From 687667b2a3d46ed1db34f573d0605c75a44fa95f Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Fri, 6 Nov 2026 14:02:24 +0900 Subject: [PATCH 1/3] feat: cache immutable manifest content per table Keep bounded content-cache state on Table and share it with manifest readers, scans, and compatible refresh and commit handles. Configure independent table budgets from catalog and table properties while keeping FileIO and ReaderOptions unchanged. Reuse weighted LruCache for binary content and bounded normalized aliases. Skip empty and oversized files, validate lengths before storage access, and close Arrow size probes on every path while preserving errors. --- mkdocs/docs/file-io.md | 28 + src/iceberg/CMakeLists.txt | 1 + src/iceberg/arrow/arrow_io.cc | 27 +- .../catalog/memory/in_memory_catalog.cc | 22 +- src/iceberg/catalog/rest/rest_catalog.cc | 8 +- src/iceberg/catalog/rest/rest_file_io.cc | 3 + src/iceberg/catalog/rest/rest_table.cc | 8 +- src/iceberg/catalog/sql/sql_catalog.cc | 13 +- src/iceberg/content_cache.cc | 304 ++++++++ src/iceberg/content_cache_internal.h | 97 +++ src/iceberg/data_table_scan.cc | 11 +- src/iceberg/incremental_append_scan.cc | 6 +- src/iceberg/incremental_changelog_scan.cc | 10 +- src/iceberg/table.cc | 62 +- src/iceberg/table.h | 31 +- src/iceberg/table_scan.cc | 1 + src/iceberg/table_scan.h | 3 + src/iceberg/test/CMakeLists.txt | 1 + src/iceberg/test/arrow_io_test.cc | 140 ++++ src/iceberg/test/cache_test.cc | 111 +++ src/iceberg/test/fast_append_test.cc | 1 + src/iceberg/test/manifest_reader_test.cc | 154 ++++ src/iceberg/test/metadata_cache_test.cc | 668 ++++++++++++++++++ src/iceberg/test/transaction_test.cc | 23 + src/iceberg/transaction.cc | 1 + src/iceberg/update/cherry_pick_operation.cc | 4 +- src/iceberg/update/expire_snapshots.cc | 10 +- src/iceberg/update/fast_append.cc | 6 +- src/iceberg/update/merging_snapshot_update.cc | 7 +- src/iceberg/update/overwrite_files.cc | 16 +- src/iceberg/update/replace_partitions.cc | 2 +- src/iceberg/update/rewrite_files.cc | 2 +- src/iceberg/update/row_delta.cc | 2 +- src/iceberg/update/snapshot_update.cc | 6 +- src/iceberg/util/cache_internal.h | 108 ++- 35 files changed, 1813 insertions(+), 84 deletions(-) create mode 100644 src/iceberg/content_cache.cc create mode 100644 src/iceberg/content_cache_internal.h create mode 100644 src/iceberg/test/metadata_cache_test.cc diff --git a/mkdocs/docs/file-io.md b/mkdocs/docs/file-io.md index 62f998b1f..823035c24 100644 --- a/mkdocs/docs/file-io.md +++ b/mkdocs/docs/file-io.md @@ -21,6 +21,34 @@ `FileIO` reads, writes, and deletes Iceberg data and metadata files. +## Manifest content cache + +Manifest content caching is owned by each loaded `Table`. Tables sharing a `FileIO` +have independent caches and byte budgets. The storage `FileIO` remains usable for +uncached data-file and table-metadata JSON reads. + +| Property | Default | Description | +|---|---|---| +| `io.manifest.cache-enabled` | `false` | Cache immutable manifest-list and manifest bytes across scans of a table | +| `io.manifest.cache.max-total-bytes` | `104857600` | Maximum combined content retained by one table's cache | +| `io.manifest.cache.max-content-length` | `8388608` | Maximum file size eligible for caching | + +Catalog I/O properties provide defaults. Table properties override them. Direct +`Table::Make` calls can supply defaults through the trailing `io_properties` +argument; properties exposed by the supplied FileIO are also used as defaults. + +Scans and manifest maintenance share the table's cache. Refreshes and transactions +preserve cached bytes when the table UUID, cache limits, and storage context are +compatible. Independently loaded table handles start with separate caches. + +`Table::ClearManifestCache()` releases retained entries. Active input handles and +streams can keep their bytes alive after clearing or eviction; those bytes and +concurrent population buffers are outside the retention budget. Normalized path +aliases use a separate index retaining at most 64 KiB of alias strings per table +and do not pin content. Empty files and +files above either byte limit bypass the cache. Time-based expiration is not +supported, so `io.manifest.cache.expiration-interval-ms` is ignored. + ## Built-in implementations Call `iceberg::arrow::RegisterAll()` to register the Arrow-backed FileIO diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index eee720613..b72333b4f 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -26,6 +26,7 @@ set(ICEBERG_SOURCES catalog/memory/in_memory_catalog.cc catalog/session_catalog.cc catalog/session_context.cc + content_cache.cc data_table_scan.cc delete_file_index.cc deletes/dv_util.cc diff --git a/src/iceberg/arrow/arrow_io.cc b/src/iceberg/arrow/arrow_io.cc index 998f8c5ce..f9bab7b80 100644 --- a/src/iceberg/arrow/arrow_io.cc +++ b/src/iceberg/arrow/arrow_io.cc @@ -22,6 +22,7 @@ #include #include #include +#include #include #include @@ -428,24 +429,41 @@ class ArrowInputFile : public InputFile { } ::arrow::fs::FileInfo file_info(path_, ::arrow::fs::FileType::File); ICEBERG_ARROW_ASSIGN_OR_RETURN(auto input, fs_->OpenInputFile(file_info)); - ICEBERG_ARROW_ASSIGN_OR_RETURN(auto size, input->GetSize()); + auto size_result = input->GetSize(); + auto close_status = input->Close(); + if (!size_result.ok() && !close_status.ok()) { + auto status = size_result.status().WithMessage( + size_result.status().message(), + "; additionally failed to close file: ", close_status.ToString()); + ICEBERG_ARROW_RETURN_NOT_OK(status); + } + ICEBERG_ARROW_ASSIGN_OR_RETURN(auto size, std::move(size_result)); + ICEBERG_ARROW_RETURN_NOT_OK(close_status); + file_size_ = size; return size; } Result> Open() override { + ICEBERG_ASSIGN_OR_RAISE(auto input, OpenArrowFile()); + return std::make_unique(std::move(input)); + } + + Result> OpenArrowFile() { ::arrow::fs::FileInfo file_info(path_, ::arrow::fs::FileType::File); if (file_size_.has_value()) { file_info.set_size(*file_size_); } ICEBERG_ARROW_ASSIGN_OR_RETURN(auto input, fs_->OpenInputFile(file_info)); - return std::make_unique(std::move(input)); + return input; } private: std::shared_ptr<::arrow::fs::FileSystem> fs_; std::string location_; std::string path_; - std::optional file_size_; + // Set by Size() so a later Open() can skip the size lookup. Not synchronized: like + // other InputFile handles, this one is not shared across threads. + mutable std::optional file_size_; }; class ArrowOutputFile : public OutputFile { @@ -530,6 +548,9 @@ Result> OpenArrowInputStream( } else { ICEBERG_ASSIGN_OR_RAISE(input_file, io->NewInputFile(path)); } + if (auto* arrow_input = dynamic_cast(input_file.get())) { + return arrow_input->OpenArrowFile(); + } ICEBERG_ASSIGN_OR_RAISE(size, input_file->Size()); if (size < 0) { return Invalid("Invalid negative file size {} for {}", size, path); diff --git a/src/iceberg/catalog/memory/in_memory_catalog.cc b/src/iceberg/catalog/memory/in_memory_catalog.cc index 2ca198335..86031f2be 100644 --- a/src/iceberg/catalog/memory/in_memory_catalog.cc +++ b/src/iceberg/catalog/memory/in_memory_catalog.cc @@ -23,6 +23,7 @@ #include #include "iceberg/catalog/catalog_util.h" +#include "iceberg/content_cache_internal.h" #include "iceberg/file_io.h" #include "iceberg/metrics/metrics_reporters.h" #include "iceberg/table.h" @@ -343,6 +344,8 @@ Result> InMemoryCatalog::Make( const std::string& name, const std::shared_ptr& file_io, const std::string& warehouse_location, const std::unordered_map& properties) { + ICEBERG_PRECHECK(file_io != nullptr, "InMemoryCatalog requires a non-null FileIO"); + ICEBERG_RETURN_UNEXPECTED(internal::ContentCacheOptions::FromProperties(properties)); std::shared_ptr reporter; auto it = properties.find(std::string(kMetricsReporterImpl)); if (it != properties.end() && !it->second.empty() && @@ -445,7 +448,8 @@ Result> InMemoryCatalog::CreateTable( root_namespace_->UpdateTableMetadataLocation(identifier, metadata_file_location)); return Table::Make(identifier, std::move(table_metadata), std::move(metadata_file_location), file_io_, shared_from_this(), - CatalogUtil::FullTableName(name(), identifier), reporter_); + CatalogUtil::FullTableName(name(), identifier), reporter_, + /*labels=*/nullptr, properties_); } Result> InMemoryCatalog::UpdateTable( @@ -497,7 +501,8 @@ Result> InMemoryCatalog::UpdateTable( return Table::Make(identifier, std::move(updated), std::move(new_metadata_location), file_io_, shared_from_this(), - CatalogUtil::FullTableName(name(), identifier), reporter_); + CatalogUtil::FullTableName(name(), identifier), reporter_, + /*labels=*/nullptr, properties_); } Result> InMemoryCatalog::StageCreateTable( @@ -517,10 +522,10 @@ Result> InMemoryCatalog::StageCreateTable( auto table_metadata, TableMetadata::Make(*schema, *spec, *order, base_location, properties)); ICEBERG_ASSIGN_OR_RAISE( - auto table, - StagedTable::Make(identifier, std::move(table_metadata), "", file_io_, - shared_from_this(), - CatalogUtil::FullTableName(name(), identifier), reporter_)); + auto table, StagedTable::Make(identifier, std::move(table_metadata), "", file_io_, + shared_from_this(), + CatalogUtil::FullTableName(name(), identifier), + reporter_, properties_)); return Transaction::Make(std::move(table), TransactionKind::kCreate); } @@ -601,7 +606,8 @@ Result> InMemoryCatalog::LoadTable( TableMetadataUtil::Read(*file_io_, metadata_location)); return Table::Make(identifier, std::move(metadata), std::move(metadata_location), file_io_, shared_from_this(), - CatalogUtil::FullTableName(name(), identifier), reporter_); + CatalogUtil::FullTableName(name(), identifier), reporter_, + /*labels=*/nullptr, properties_); } Result> InMemoryCatalog::RegisterTable( @@ -624,7 +630,7 @@ Result> InMemoryCatalog::RegisterTable( } return Table::Make(identifier, std::move(metadata), metadata_file_location, file_io_, shared_from_this(), CatalogUtil::FullTableName(name(), identifier), - reporter_); + reporter_, /*labels=*/nullptr, properties_); } } // namespace iceberg diff --git a/src/iceberg/catalog/rest/rest_catalog.cc b/src/iceberg/catalog/rest/rest_catalog.cc index a25d28eb4..431f9405f 100644 --- a/src/iceberg/catalog/rest/rest_catalog.cc +++ b/src/iceberg/catalog/rest/rest_catalog.cc @@ -923,12 +923,14 @@ Result> RestCatalog::StageCreateTable( auto table_catalog = std::make_shared( shared_from_this(), context, identifier, table_config, std::move(table_session), table_io); + auto manifest_properties = config_.configs(); + for (const auto& [key, value] : table_config) manifest_properties[key] = value; ICEBERG_ASSIGN_OR_RAISE( auto staged_table, StagedTable::Make(identifier, std::move(result.metadata), std::move(result.metadata_location), std::move(table_io), std::move(table_catalog), RestTableName(name_, identifier), - std::move(reporter))); + std::move(reporter), manifest_properties)); return Transaction::Make(std::move(staged_table), TransactionKind::kCreate); } @@ -1074,6 +1076,8 @@ Result> RestCatalog::MakeTable( std::shared_ptr table_catalog, std::shared_ptr reporter, const std::unordered_map& table_config, std::shared_ptr table_session, std::unique_ptr labels) { + auto manifest_properties = config_.configs(); + for (const auto& [key, value] : table_config) manifest_properties[key] = value; // Table config overrides client config. ICEBERG_ASSIGN_OR_RAISE(auto server_mode, RestCatalogProperties::ScanPlanningModeFrom(table_config)); @@ -1127,7 +1131,7 @@ Result> RestCatalog::MakeTable( return Table::Make(identifier, std::move(metadata), std::move(metadata_location), std::move(table_io), std::move(table_catalog), RestTableName(name_, identifier), std::move(reporter), - std::move(labels)); + std::move(labels), manifest_properties); } } // namespace iceberg::rest diff --git a/src/iceberg/catalog/rest/rest_file_io.cc b/src/iceberg/catalog/rest/rest_file_io.cc index 4cd102704..f7f7aa641 100644 --- a/src/iceberg/catalog/rest/rest_file_io.cc +++ b/src/iceberg/catalog/rest/rest_file_io.cc @@ -25,6 +25,7 @@ #include #include "iceberg/catalog/rest/types.h" +#include "iceberg/content_cache_internal.h" #include "iceberg/file_io.h" #include "iceberg/file_io_registry.h" #include "iceberg/logging/log_macros.h" @@ -48,6 +49,8 @@ std::unordered_map MergeFileIOProperties( } // namespace Result> MakeCatalogFileIO(const RestCatalogProperties& config) { + ICEBERG_RETURN_UNEXPECTED( + internal::ContentCacheOptions::FromProperties(config.configs())); const std::string io_impl = config.Get(RestCatalogProperties::kIOImpl); if (io_impl.empty()) { return std::make_unique(config.configs()); diff --git a/src/iceberg/catalog/rest/rest_table.cc b/src/iceberg/catalog/rest/rest_table.cc index f70bd9e92..c12909817 100644 --- a/src/iceberg/catalog/rest/rest_table.cc +++ b/src/iceberg/catalog/rest/rest_table.cc @@ -48,13 +48,19 @@ Result> RestTable::Make( std::shared_ptr catalog, std::string full_name, std::shared_ptr reporter, RestScanContext rest_context, std::unique_ptr labels) { + ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null"); if (metadata == nullptr) { return InvalidArgument("Metadata cannot be null"); } - return std::shared_ptr(new RestTable( + auto table = std::shared_ptr(new RestTable( std::move(identifier), std::move(metadata), std::move(metadata_location), std::move(io), std::move(catalog), std::move(full_name), std::move(reporter), std::move(rest_context), std::move(labels))); + auto properties = table->rest_context_->catalog_config; + for (const auto& [key, value] : table->rest_context_->table_config) + properties[key] = value; + ICEBERG_RETURN_UNEXPECTED(table->InitializeManifestCache(properties)); + return table; } Result> RestTable::NewScan() const { diff --git a/src/iceberg/catalog/sql/sql_catalog.cc b/src/iceberg/catalog/sql/sql_catalog.cc index e69729327..3b4bfa543 100644 --- a/src/iceberg/catalog/sql/sql_catalog.cc +++ b/src/iceberg/catalog/sql/sql_catalog.cc @@ -25,6 +25,7 @@ #include "iceberg/catalog/catalog_util.h" #include "iceberg/catalog/sql/config.h" +#include "iceberg/content_cache_internal.h" #include "iceberg/file_io.h" #include "iceberg/metrics/metrics_reporters.h" #include "iceberg/table.h" @@ -148,6 +149,7 @@ Result> SqlCatalog::Make( if (file_io == nullptr) { return InvalidArgument("SqlCatalog requires a non-null FileIO"); } + ICEBERG_RETURN_UNEXPECTED(internal::ContentCacheOptions::FromProperties(config.props)); ICEBERG_RETURN_UNEXPECTED(store->Initialize()); std::shared_ptr reporter; @@ -385,7 +387,7 @@ Result> SqlCatalog::LoadTableFrom( TableMetadataUtil::Read(*file_io_, metadata_location)); return Table::Make(identifier, std::move(metadata), metadata_location, file_io_, shared_from_this(), CatalogUtil::FullTableName(name(), identifier), - reporter_); + reporter_, /*labels=*/nullptr, config_.props); } Result> SqlCatalog::LoadTable(const TableIdentifier& identifier) { @@ -424,7 +426,7 @@ Result> SqlCatalog::CreateTable( return Table::Make(identifier, std::move(metadata), metadata_location, file_io_, shared_from_this(), CatalogUtil::FullTableName(name(), identifier), - reporter_); + reporter_, /*labels=*/nullptr, config_.props); } Result> SqlCatalog::UpdateTable( @@ -490,7 +492,7 @@ Result> SqlCatalog::UpdateTable( return Table::Make(identifier, std::move(updated), new_metadata_location, file_io_, shared_from_this(), CatalogUtil::FullTableName(name(), identifier), - reporter_); + reporter_, /*labels=*/nullptr, config_.props); } Result> SqlCatalog::StageCreateTable( @@ -518,7 +520,8 @@ Result> SqlCatalog::StageCreateTable( ICEBERG_ASSIGN_OR_RAISE( auto table, StagedTable::Make(identifier, std::move(metadata), "", file_io_, shared_from_this(), - CatalogUtil::FullTableName(name(), identifier), reporter_)); + CatalogUtil::FullTableName(name(), identifier), reporter_, + config_.props)); return Transaction::Make(std::move(table), TransactionKind::kCreate); } @@ -599,7 +602,7 @@ Result> SqlCatalog::RegisterTable( return Table::Make(identifier, std::move(metadata), metadata_file_location, file_io_, shared_from_this(), CatalogUtil::FullTableName(name(), identifier), - reporter_); + reporter_, /*labels=*/nullptr, config_.props); } // -------------------------------------------------------------------------- diff --git a/src/iceberg/content_cache.cc b/src/iceberg/content_cache.cc new file mode 100644 index 000000000..bb4f52e1f --- /dev/null +++ b/src/iceberg/content_cache.cc @@ -0,0 +1,304 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include +#include +#include +#include + +#include "iceberg/content_cache_internal.h" +#include "iceberg/util/macros.h" +#include "iceberg/util/string_util.h" + +namespace iceberg::internal { +namespace { + +template +Result ParseNumberProperty( + const std::unordered_map& properties, std::string_view key, + T fallback) { + auto it = properties.find(std::string(key)); + if (it == properties.end()) return fallback; + auto result = StringUtils::ParseNumber(it->second); + if (!result) + return InvalidArgument("Invalid numeric value '{}' for property '{}': {}", it->second, + key, result.error().message); + return result; +} + +Status FinishWithCloseStatus(Status operation_status, Status close_status) { + if (!operation_status.has_value()) { + auto error = operation_status.error(); + if (!close_status.has_value()) { + error.message += "; additionally failed to close stream: "; + error.message += close_status.error().message; + } + return ::iceberg::unexpected(std::move(error)); + } + return close_status; +} + +Result> ReadInputFile(InputFile& input_file, int64_t read_size, + std::string_view file_location) { + if (read_size < 0) { + return Invalid("Invalid negative file size {} for {}", read_size, file_location); + } + if (static_cast(read_size) > + static_cast(std::numeric_limits::max())) { + return Invalid("File size {} exceeds size_t max for {}", read_size, file_location); + } + + auto size = static_cast(read_size); + std::vector content(size); + ICEBERG_ASSIGN_OR_RAISE(auto stream, input_file.Open()); + Status read_status = {}; + if (size > 0) { + read_status = stream->ReadFully(/*position=*/0, content); + } + ICEBERG_RETURN_UNEXPECTED( + FinishWithCloseStatus(std::move(read_status), stream->Close())); + return content; +} + +class CachedSeekableInputStream : public SeekableInputStream { + public: + explicit CachedSeekableInputStream(MetadataContent content) + : content_(std::move(content)) {} + + Result Position() const override { return position_; } + + Status Seek(int64_t position) override { + ICEBERG_PRECHECK(!closed_, "Input stream is closed"); + ICEBERG_PRECHECK(position >= 0, "Position must not be negative: {}", position); + ICEBERG_PRECHECK(static_cast(position) <= content_->size(), + "Position {} exceeds file size {}", position, content_->size()); + position_ = position; + return {}; + } + + Result Read(std::span out) override { + ICEBERG_PRECHECK(!closed_, "Input stream is closed"); + auto position = static_cast(position_); + auto bytes_to_read = std::min(out.size(), content_->size() - position); + if (bytes_to_read > 0) { + std::memcpy(out.data(), content_->data() + position, bytes_to_read); + position_ += static_cast(bytes_to_read); + } + return static_cast(bytes_to_read); + } + + Status ReadFully(int64_t position, std::span out) override { + ICEBERG_PRECHECK(!closed_, "Input stream is closed"); + ICEBERG_PRECHECK(position >= 0, "Position must not be negative: {}", position); + ICEBERG_PRECHECK(static_cast(position) <= content_->size(), + "Position {} exceeds file size {}", position, content_->size()); + auto offset = static_cast(position); + ICEBERG_PRECHECK(out.size() <= content_->size() - offset, + "Read out of bounds: offset {} + length {} exceeds file size {}", + position, out.size(), content_->size()); + if (!out.empty()) { + std::memcpy(out.data(), content_->data() + offset, out.size()); + } + return {}; + } + + Status Close() override { + closed_ = true; + return {}; + } + + private: + MetadataContent content_; + int64_t position_ = 0; + bool closed_ = false; +}; + +class CachedInputFile : public InputFile { + public: + CachedInputFile(std::unique_ptr input, std::shared_ptr cache, + int64_t size) + : input_(std::move(input)), + cache_(std::move(cache)), + location_(input_->location()), + size_(size) {} + CachedInputFile(std::string location, MetadataContent content) + : location_(std::move(location)), + size_(static_cast(content->size())), + content_(std::move(content)) {} + std::string_view location() const override { return location_; } + Result Size() const override { return size_; } + Result> Open() override { + if (content_) return std::make_unique(content_); + if (auto cached = cache_->Find(location_)) + return std::make_unique(std::move(cached)); + auto loaded = ReadInputFile(*input_, size_, location_); + if (!loaded) { + auto error = std::move(loaded).error(); + if (error.kind != ErrorKind::kIOError) return unexpected(std::move(error)); + auto fallback = input_->Open(); + if (!fallback) { + error.message += "; fallback open failed: " + fallback.error().message; + return unexpected(std::move(error)); + } + return fallback; + } + auto content = + std::make_shared>(std::move(loaded).value()); + cache_->Insert(location_, content); + return std::make_unique(std::move(content)); + } + + private: + std::unique_ptr input_; + std::shared_ptr cache_; + std::string location_; + int64_t size_; + MetadataContent content_; +}; + +// Only manifest readers receive this adapter. Output and deletion operations are +// forwarded unchanged, allowing manifest maintenance to use the same read context. +class ManifestFileIO : public FileIO { + public: + ManifestFileIO(std::shared_ptr io, std::shared_ptr cache) + : io_(std::move(io)), cache_(std::move(cache)) {} + Result> NewInputFile(std::string location) override { + return cache_->NewInputFile(io_, std::move(location)); + } + Result> NewInputFile(std::string location, + size_t length) override { + return cache_->NewInputFile(io_, std::move(location), length); + } + Result> NewOutputFile(std::string location) override { + return io_->NewOutputFile(std::move(location)); + } + Status DeleteFile(const std::string& location) override { + return io_->DeleteFile(location); + } + Status DeleteFiles(const std::vector& locations) override { + return io_->DeleteFiles(locations); + } + const std::unordered_map& properties() const override { + return io_->properties(); + } + + private: + std::shared_ptr io_; + std::shared_ptr cache_; +}; + +} // namespace + +Result ContentCacheOptions::FromProperties( + const std::unordered_map& properties) { + ContentCacheOptions options; + if (auto it = properties.find(std::string(kEnabled)); it != properties.end()) + options.enabled = StringUtils::ParseBoolean(it->second); + ICEBERG_ASSIGN_OR_RAISE( + options.max_total_bytes, + ParseNumberProperty(properties, kMaxTotalBytes, options.max_total_bytes)); + ICEBERG_ASSIGN_OR_RAISE( + options.max_content_length, + ParseNumberProperty(properties, kMaxContentLength, options.max_content_length)); + if (options.enabled) { + ICEBERG_PRECHECK( + options.max_total_bytes > 0 && + std::cmp_less_equal(options.max_total_bytes, + std::numeric_limits::max()), + "Manifest cache maximum total bytes must be positive and fit in int64_t"); + ICEBERG_PRECHECK( + options.max_content_length > 0 && + std::cmp_less_equal(options.max_content_length, + std::numeric_limits::max()), + "Manifest cache maximum content length must be positive and fit in int64_t"); + } + return options; +} + +ContentCache::ContentCache(ContentCacheOptions options) + : options_(options), + lru_(static_cast(options.max_total_bytes)), + aliases_( + static_cast(std::min(options.max_total_bytes, 64 * 1024))) {} +std::shared_ptr ContentCache::Wrap(std::shared_ptr io) { + return std::make_shared(std::move(io), shared_from_this()); +} +MetadataContent ContentCache::Find(std::string_view location, + std::string* canonical_location) { + std::lock_guard lock(mutex_); + if (auto* content = lru_.Find(location)) { + if (canonical_location) *canonical_location = location; + return *content; + } + if (auto* canonical = aliases_.Find(location)) { + if (auto* content = lru_.Find(*canonical)) { + if (canonical_location) *canonical_location = *canonical; + return *content; + } + } + return nullptr; +} +void ContentCache::Insert(const std::string& location, MetadataContent content) { + // Empty/corrupt metadata must not accumulate unbounded zero-weight entries. + if (content->empty()) return; + std::lock_guard lock(mutex_); + lru_.Replace(location, std::move(content)); +} +void ContentCache::Clear() { + std::lock_guard lock(mutex_); + lru_.Clear(); + aliases_.Clear(); +} +Result> ContentCache::NewInputFile( + const std::shared_ptr& io, std::string location, + std::optional length) { + if (length && std::cmp_greater(*length, std::numeric_limits::max())) + return InvalidArgument("File length {} exceeds int64_t max", *length); + // A hit needs neither storage resolution nor working remote credentials. + std::string canonical_location; + if (auto content = Find(location, &canonical_location)) + return std::make_unique(std::move(canonical_location), + std::move(content)); + std::unique_ptr input; + if (length) { + ICEBERG_ASSIGN_OR_RAISE(input, io->NewInputFile(location, *length)); + } else { + ICEBERG_ASSIGN_OR_RAISE(input, io->NewInputFile(location)); + } + if (input->location() != location) { + std::lock_guard lock(mutex_); + aliases_.Replace(std::move(location), std::string(input->location())); + } + if (auto content = Find(input->location())) + return std::make_unique(std::string(input->location()), + std::move(content)); + int64_t size; + if (length) + size = static_cast(*length); + else { + ICEBERG_ASSIGN_OR_RAISE(size, input->Size()); + } + ICEBERG_PRECHECK(size >= 0, "Invalid negative file size {}", size); + if (size == 0 || std::cmp_greater(size, options_.max_content_length) || + std::cmp_greater(size, options_.max_total_bytes)) + return input; + return std::make_unique(std::move(input), shared_from_this(), size); +} + +} // namespace iceberg::internal diff --git a/src/iceberg/content_cache_internal.h b/src/iceberg/content_cache_internal.h new file mode 100644 index 000000000..5ec39bb7f --- /dev/null +++ b/src/iceberg/content_cache_internal.h @@ -0,0 +1,97 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "iceberg/file_io.h" +#include "iceberg/util/cache_internal.h" +#include "iceberg/util/string_util.h" + +namespace iceberg::internal { + +// Cache limits apply independently to each loaded table. Expiration is not supported. +struct ICEBERG_EXPORT ContentCacheOptions { + inline static constexpr std::string_view kEnabled = "io.manifest.cache-enabled"; + inline static constexpr std::string_view kMaxTotalBytes = + "io.manifest.cache.max-total-bytes"; + inline static constexpr std::string_view kMaxContentLength = + "io.manifest.cache.max-content-length"; + bool enabled = false; + size_t max_total_bytes = 100 * 1024 * 1024; + size_t max_content_length = 8 * 1024 * 1024; + friend bool operator==(const ContentCacheOptions&, + const ContentCacheOptions&) = default; + static Result FromProperties( + const std::unordered_map& properties); +}; + +using MetadataContent = std::shared_ptr>; +struct MetadataContentWeigher { + int64_t operator()(const std::string&, const MetadataContent& content) const { + return static_cast(content->size()); + } +}; + +// Bound the bookkeeping for normalized aliases separately from retained content. +struct LocationAliasWeigher { + int64_t operator()(const std::string& alias, const std::string& canonical) const { + constexpr size_t kMaxAliasPart = 65537; + return static_cast(std::min(alias.size(), kMaxAliasPart) + + std::min(canonical.size(), kMaxAliasPart)); + } +}; + +// A table owns this cache; readers share it through a metadata-only FileIO adapter. +// Storage FileIOs remain independent of the cache and its configuration. +class ICEBERG_EXPORT ContentCache : public std::enable_shared_from_this { + public: + explicit ContentCache(ContentCacheOptions options); + const ContentCacheOptions& options() const { return options_; } + std::shared_ptr Wrap(std::shared_ptr io); + Result> NewInputFile( + const std::shared_ptr& io, std::string location, + std::optional length = std::nullopt); + // Optionally return the canonical location represented by a cache hit. + MetadataContent Find(std::string_view location, + std::string* canonical_location = nullptr); + void Insert(const std::string& location, MetadataContent content); + void Clear(); + + private: + const ContentCacheOptions options_; + std::mutex mutex_; + LruCache + lru_; + // Alias strings do not pin content and are capped at 64 KiB in total. + LruCache + aliases_; +}; + +} // namespace iceberg::internal diff --git a/src/iceberg/data_table_scan.cc b/src/iceberg/data_table_scan.cc index f35f454c0..05ac75b34 100644 --- a/src/iceberg/data_table_scan.cc +++ b/src/iceberg/data_table_scan.cc @@ -164,8 +164,12 @@ Result DataTableScan::PlanFilesStream() const { ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id, metadata_cache.GetPartitionSpecsById()); SnapshotReader snapshot_reader(snapshot.get()); - ICEBERG_ASSIGN_OR_RAISE(auto data_manifests, snapshot_reader.DataManifests(io_)); - ICEBERG_ASSIGN_OR_RAISE(auto delete_manifests, snapshot_reader.DeleteManifests(io_)); + ICEBERG_ASSIGN_OR_RAISE( + auto data_manifests, + snapshot_reader.DataManifests(context_.manifest_io ? context_.manifest_io : io_)); + ICEBERG_ASSIGN_OR_RAISE( + auto delete_manifests, + snapshot_reader.DeleteManifests(context_.manifest_io ? context_.manifest_io : io_)); if (scan_metrics) { scan_metrics->total_data_manifests->Increment( @@ -183,7 +187,8 @@ Result DataTableScan::PlanFilesStream() const { ICEBERG_ASSIGN_OR_RAISE( auto manifest_group, - ManifestGroup::Make(io_, schema_, specs_by_id, std::move(owned_data_manifests), + ManifestGroup::Make(context_.manifest_io ? context_.manifest_io : io_, schema_, + specs_by_id, std::move(owned_data_manifests), std::move(owned_delete_manifests))); manifest_group->CaseSensitive(context_.case_sensitive) .Select(ScanColumns()) diff --git a/src/iceberg/incremental_append_scan.cc b/src/iceberg/incremental_append_scan.cc index cf01fe894..189d374eb 100644 --- a/src/iceberg/incremental_append_scan.cc +++ b/src/iceberg/incremental_append_scan.cc @@ -75,7 +75,9 @@ Result>> IncrementalAppendScan::PlanFi std::unordered_set data_manifests; for (const auto& snapshot : append_snapshots) { SnapshotReader snapshot_reader(snapshot.get()); - ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_reader.DataManifests(io_)); + ICEBERG_ASSIGN_OR_RAISE( + auto manifests, + snapshot_reader.DataManifests(context_.manifest_io ? context_.manifest_io : io_)); std::ranges::copy_if( manifests, std::inserter(data_manifests, data_manifests.end()), [&snapshot_ids](const ManifestFile& manifest) { @@ -93,7 +95,7 @@ Result>> IncrementalAppendScan::PlanFi ICEBERG_ASSIGN_OR_RAISE( auto manifest_group, ManifestGroup::Make( - io_, schema_, specs_by_id, + context_.manifest_io ? context_.manifest_io : io_, schema_, specs_by_id, std::vector(data_manifests.begin(), data_manifests.end()), {})); manifest_group->CaseSensitive(context_.case_sensitive) diff --git a/src/iceberg/incremental_changelog_scan.cc b/src/iceberg/incremental_changelog_scan.cc index 5a1dbba11..3f4dba6a0 100644 --- a/src/iceberg/incremental_changelog_scan.cc +++ b/src/iceberg/incremental_changelog_scan.cc @@ -67,7 +67,8 @@ IncrementalChangelogScan::PlanFiles(std::optional from_snapshot_id_excl if (!operation.has_value() || operation.value() != DataOperation::kReplace) { auto snapshot_reader = std::make_unique(snapshot.get()); ICEBERG_ASSIGN_OR_RAISE(auto delete_manifests, - snapshot_reader->DeleteManifests(io_)); + snapshot_reader->DeleteManifests( + context_.manifest_io ? context_.manifest_io : io_)); if (!delete_manifests.empty()) { return NotSupported( "Delete files are currently not supported in changelog scans"); @@ -93,7 +94,9 @@ IncrementalChangelogScan::PlanFiles(std::optional from_snapshot_id_excl std::vector data_manifests; std::unordered_set seen_manifest_paths; for (const auto& snapshot : changelog_snapshots) { - ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot.second->DataManifests(io_)); + ICEBERG_ASSIGN_OR_RAISE(auto manifests, + snapshot.second->DataManifests( + context_.manifest_io ? context_.manifest_io : io_)); for (auto& manifest : manifests) { if (manifest.added_snapshot_id.has_value() && snapshot_ids.contains(manifest.added_snapshot_id.value()) && @@ -111,7 +114,8 @@ IncrementalChangelogScan::PlanFiles(std::optional from_snapshot_id_excl ICEBERG_ASSIGN_OR_RAISE( auto manifest_group, - ManifestGroup::Make(io_, schema_, specs_by_id, std::move(data_manifests), + ManifestGroup::Make(context_.manifest_io ? context_.manifest_io : io_, schema_, + specs_by_id, std::move(data_manifests), /*delete_manifests=*/{})); manifest_group->CaseSensitive(context_.case_sensitive) diff --git a/src/iceberg/table.cc b/src/iceberg/table.cc index 0fa6bf405..55f3d198b 100644 --- a/src/iceberg/table.cc +++ b/src/iceberg/table.cc @@ -22,7 +22,9 @@ #include #include "iceberg/catalog.h" +#include "iceberg/content_cache_internal.h" #include "iceberg/data_table_scan.h" +#include "iceberg/file_io.h" #include "iceberg/incremental_append_scan.h" #include "iceberg/incremental_changelog_scan.h" #include "iceberg/inspect/metadata_table.h" @@ -62,7 +64,8 @@ Result> Table::Make( TableIdentifier identifier, std::shared_ptr metadata, std::string metadata_location, std::shared_ptr io, std::shared_ptr catalog, std::string full_name, - std::shared_ptr reporter, std::unique_ptr labels) { + std::shared_ptr reporter, std::unique_ptr labels, + const std::unordered_map& io_properties) { if (metadata == nullptr) [[unlikely]] { return InvalidArgument("Metadata cannot be null"); } @@ -75,10 +78,12 @@ Result> Table::Make( if (catalog == nullptr) [[unlikely]] { return InvalidArgument("Catalog cannot be null"); } - return std::shared_ptr(new Table(std::move(identifier), std::move(metadata), - std::move(metadata_location), std::move(io), - std::move(catalog), std::move(full_name), - std::move(reporter), std::move(labels))); + auto table = std::shared_ptr
( + new Table(std::move(identifier), std::move(metadata), std::move(metadata_location), + std::move(io), std::move(catalog), std::move(full_name), + std::move(reporter), std::move(labels))); + ICEBERG_RETURN_UNEXPECTED(table->InitializeManifestCache(io_properties)); + return table; } Table::~Table() = default; @@ -92,16 +97,52 @@ Table::Table(TableIdentifier identifier, std::shared_ptr metadata metadata_(std::move(metadata)), metadata_location_(std::move(metadata_location)), io_(std::move(io)), + manifest_io_(io_), catalog_(std::move(catalog)), reporter_(std::move(reporter)), metadata_cache_(std::make_unique(metadata_.get())), labels_(labels && !labels->empty() ? std::move(labels) : nullptr) {} +Status Table::InitializeManifestCache( + const std::unordered_map& io_properties) { + auto properties = io_->properties(); + for (const auto& [key, value] : io_properties) properties[key] = value; + for (const auto& [key, value] : metadata_->properties.configs()) + properties[key] = value; + ICEBERG_ASSIGN_OR_RAISE(auto options, + internal::ContentCacheOptions::FromProperties(properties)); + if (options.enabled) { + content_cache_ = std::make_shared(options); + manifest_io_ = content_cache_->Wrap(io_); + } + return {}; +} + +void Table::ClearManifestCache() { + if (content_cache_) content_cache_->Clear(); +} + +void Table::ReuseManifestCache(const Table& source) { + if (!content_cache_ || !source.content_cache_ || uuid() != source.uuid() || + content_cache_->options() != source.content_cache_->options()) + return; + // A replaced storage context with no comparable properties may represent a different + // endpoint or filesystem. Only preserve bytes when its identity is known to match. + if (io_ != source.io_ && + (io_->properties().empty() || io_->properties() != source.io_->properties())) + return; + content_cache_ = source.content_cache_; + manifest_io_ = content_cache_->Wrap(io_); +} + const std::string& Table::uuid() const { return metadata_->table_uuid; } Status Table::Refresh() { ICEBERG_ASSIGN_OR_RAISE(auto refreshed_table, catalog_->LoadTable(identifier_)); if (metadata_location_ != refreshed_table->metadata_file_location()) { + refreshed_table->ReuseManifestCache(*this); + content_cache_ = std::move(refreshed_table->content_cache_); + manifest_io_ = std::move(refreshed_table->manifest_io_); metadata_ = std::move(refreshed_table->metadata_); metadata_location_ = std::string(refreshed_table->metadata_file_location()); io_ = std::move(refreshed_table->io_); @@ -303,7 +344,8 @@ Result> StagedTable::Make( TableIdentifier identifier, std::shared_ptr metadata, std::string metadata_location, std::shared_ptr io, std::shared_ptr catalog, std::string full_name, - std::shared_ptr reporter) { + std::shared_ptr reporter, + const std::unordered_map& io_properties) { if (metadata == nullptr) [[unlikely]] { return InvalidArgument("Metadata cannot be null"); } @@ -313,10 +355,12 @@ Result> StagedTable::Make( if (catalog == nullptr) [[unlikely]] { return InvalidArgument("Catalog cannot be null"); } - return std::shared_ptr( + auto table = std::shared_ptr( new StagedTable(std::move(identifier), std::move(metadata), std::move(metadata_location), std::move(io), std::move(catalog), std::move(full_name), std::move(reporter), /*labels=*/nullptr)); + ICEBERG_RETURN_UNEXPECTED(table->InitializeManifestCache(io_properties)); + return table; } StagedTable::~StagedTable() = default; @@ -334,10 +378,12 @@ Result> StaticTable::Make( if (io == nullptr) [[unlikely]] { return InvalidArgument("FileIO cannot be null"); } - return std::shared_ptr( + auto table = std::shared_ptr( new StaticTable(std::move(identifier), std::move(metadata), std::move(metadata_location), std::move(io), /*catalog=*/nullptr, std::move(full_name), /*reporter=*/nullptr, /*labels=*/nullptr)); + ICEBERG_RETURN_UNEXPECTED(table->InitializeManifestCache()); + return table; } StaticTable::~StaticTable() = default; diff --git a/src/iceberg/table.h b/src/iceberg/table.h index 3a5f07047..78a59f30a 100644 --- a/src/iceberg/table.h +++ b/src/iceberg/table.h @@ -38,6 +38,10 @@ namespace iceberg { +namespace internal { +class ContentCache; +} + /// \brief Represents an Iceberg table /// /// Safe to read from several threads, but not while one of them commits through it @@ -55,12 +59,15 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this
{ /// \param[in] reporter Optional metrics reporter for this table. Defaults to nullptr /// (noop). /// \param[in] labels Labels for this table, or nullptr when it has none. + /// \param[in] io_properties Catalog I/O defaults for the table manifest cache. + /// Table properties override these defaults; the budget is per loaded table. static Result> Make( TableIdentifier identifier, std::shared_ptr metadata, std::string metadata_location, std::shared_ptr io, std::shared_ptr catalog, std::string full_name = "", std::shared_ptr reporter = nullptr, - std::unique_ptr labels = nullptr); + std::unique_ptr labels = nullptr, + const std::unordered_map& io_properties = {}); virtual ~Table(); @@ -127,6 +134,16 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this
{ /// \brief Returns a FileIO to read and write table data and metadata files const std::shared_ptr& io() const; + /// \brief I/O context for immutable manifest lists and manifests only. + /// Each loaded table has its own optional content cache, shared by its readers. + const std::shared_ptr& manifest_io() const { return manifest_io_; } + + /// \brief Return whether this table caches manifest content. + bool ManifestCacheEnabled() const { return content_cache_ != nullptr; } + + /// \brief Release manifest content retained by this table. + void ClearManifestCache(); + /// \brief Returns the current metadata for this table const std::shared_ptr& metadata() const; @@ -231,6 +248,9 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this
{ Result> Inspect(MetadataTableKind kind) const; protected: + Status InitializeManifestCache( + const std::unordered_map& io_properties = {}); + Table(TableIdentifier identifier, std::shared_ptr metadata, std::string metadata_location, std::shared_ptr io, std::shared_ptr catalog, std::string full_name, @@ -242,10 +262,16 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this
{ std::shared_ptr metadata_; std::string metadata_location_; std::shared_ptr io_; + std::shared_ptr content_cache_; + std::shared_ptr manifest_io_; std::shared_ptr catalog_; std::shared_ptr reporter_; std::unique_ptr metadata_cache_; const std::unique_ptr labels_; + + private: + friend class Transaction; + void ReuseManifestCache(const Table& source); }; /// \brief A table created by stage-create and not yet committed. @@ -255,7 +281,8 @@ class ICEBERG_EXPORT StagedTable final : public Table { TableIdentifier identifier, std::shared_ptr metadata, std::string metadata_location, std::shared_ptr io, std::shared_ptr catalog, std::string full_name = "", - std::shared_ptr reporter = nullptr); + std::shared_ptr reporter = nullptr, + const std::unordered_map& io_properties = {}); ~StagedTable() override; diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc index d04e3385b..723091b41 100644 --- a/src/iceberg/table_scan.cc +++ b/src/iceberg/table_scan.cc @@ -299,6 +299,7 @@ Result>> TableScanBuilder:: auto builder = std::unique_ptr>(new TableScanBuilder( table.metadata(), table.io(), table.full_name(), table.reporter())); + builder->context_.manifest_io = table.manifest_io(); return builder; } diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h index 1307089e3..b7e5659dd 100644 --- a/src/iceberg/table_scan.h +++ b/src/iceberg/table_scan.h @@ -277,6 +277,9 @@ struct ICEBERG_EXPORT TableScanContext { std::string table_name; std::shared_ptr metrics_reporter; + // Table-owned manifest content cache, independent from data-file I/O. + std::shared_ptr manifest_io; + // Validate the context parameters to see if they have conflicts. [[nodiscard]] Status Validate() const; }; diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index b2fd4f9be..7499f507a 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -155,6 +155,7 @@ add_iceberg_test(util_test lazy_test.cc location_util_test.cc math_util_internal_test.cc + metadata_cache_test.cc executor_util_test.cc roaring_position_bitmap_test.cc position_delete_index_test.cc diff --git a/src/iceberg/test/arrow_io_test.cc b/src/iceberg/test/arrow_io_test.cc index 7bc9ebba5..3ed0e39ba 100644 --- a/src/iceberg/test/arrow_io_test.cc +++ b/src/iceberg/test/arrow_io_test.cc @@ -24,16 +24,23 @@ #include #include +#include #include +#include #include #include #include #include "iceberg/arrow/arrow_io_internal.h" #include "iceberg/arrow/arrow_register.h" +#include "iceberg/content_cache_internal.h" +#include "iceberg/file_io.h" #include "iceberg/file_io_registry.h" #include "iceberg/resolving_file_io.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" #include "iceberg/test/matchers.h" +#include "iceberg/test/mock_catalog.h" #include "iceberg/test/std_io.h" #include "iceberg/test/temp_file_test_base.h" @@ -50,6 +57,74 @@ struct CloseState { bool closed = false; }; +class TrackingLocalFileSystem : public ::arrow::fs::LocalFileSystem { + public: + using ::arrow::fs::LocalFileSystem::OpenInputFile; + + ::arrow::Result> OpenInputFile( + const ::arrow::fs::FileInfo& info) override { + ARROW_ASSIGN_OR_RAISE(auto input, + ::arrow::fs::LocalFileSystem::OpenInputFile(info.path())); + opened_inputs.emplace_back(input); + return input; + } + + // Strong references keep every opened file alive, so tests can tell an explicit + // Close() apart from the file simply going out of scope. + std::vector> opened_inputs; +}; + +class SizeFailureArrowFile : public ::arrow::io::RandomAccessFile { + public: + explicit SizeFailureArrowFile(bool fail_close) : fail_close_(fail_close) {} + ::arrow::Result GetSize() override { + return ::arrow::Status::IOError("size failed"); + } + ::arrow::Status Close() override { + ++close_calls; + closed_ = true; + return fail_close_ ? ::arrow::Status::IOError("close failed") : ::arrow::Status::OK(); + } + bool closed() const override { return closed_; } + ::arrow::Result Tell() const override { return 0; } + ::arrow::Status Seek(int64_t) override { return ::arrow::Status::OK(); } + ::arrow::Result Read(int64_t, void*) override { return 0; } + ::arrow::Result> Read(int64_t) override { + return ::arrow::Buffer::FromString(""); + } + int close_calls = 0; + + private: + bool fail_close_; + bool closed_ = false; +}; + +class SizeFailureFileSystem : public ::arrow::fs::LocalFileSystem { + public: + using ::arrow::fs::LocalFileSystem::OpenInputFile; + explicit SizeFailureFileSystem(bool fail_close) + : input(std::make_shared(fail_close)) {} + ::arrow::Result> OpenInputFile( + const ::arrow::fs::FileInfo&) override { + return input; + } + std::shared_ptr input; +}; + +TEST(ArrowFileIOTest, SizeClosesTemporaryInputOnFailure) { + for (bool fail_close : {false, true}) { + SCOPED_TRACE(fail_close); + auto fs = std::make_shared(fail_close); + auto io = std::make_shared(fs); + ICEBERG_UNWRAP_OR_FAIL(auto input, io->NewInputFile("manifest.avro")); + auto result = input->Size(); + EXPECT_THAT(result, HasErrorMessage("size failed")); + if (fail_close) EXPECT_THAT(result, HasErrorMessage("close failed")); + EXPECT_TRUE(fs->input->closed()); + EXPECT_EQ(fs->input->close_calls, 1); + } +} + class ReadFailureInputStream : public SeekableInputStream { public: explicit ReadFailureInputStream(std::shared_ptr state) @@ -356,6 +431,51 @@ TEST_F(LocalFileIOTest, ReadWriteFile) { EXPECT_THAT(read_res, HasValue(::testing::Eq("hello world"))); } +TEST_F(LocalFileIOTest, SizeReleasesTemporaryArrowFile) { + auto fs = std::make_shared(); + auto file_io = std::make_shared(fs); + ASSERT_THAT(file_io->WriteFile(temp_filepath_, "hello"), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto input_file, file_io->NewInputFile(temp_filepath_)); + + EXPECT_THAT(input_file->Size(), HasValue(::testing::Eq(5))); + ASSERT_EQ(fs->opened_inputs.size(), 1); + EXPECT_TRUE(fs->opened_inputs.front()->closed()); + + EXPECT_THAT(input_file->Size(), HasValue(::testing::Eq(5))); + EXPECT_EQ(fs->opened_inputs.size(), 1); + ICEBERG_UNWRAP_OR_FAIL(auto stream, input_file->Open()); + ASSERT_EQ(fs->opened_inputs.size(), 2); + EXPECT_FALSE(fs->opened_inputs.back()->closed()); + std::array data; + ASSERT_THAT(stream->ReadFully(0, data), IsOk()); + EXPECT_EQ(std::string(reinterpret_cast(data.data()), data.size()), + "hello"); + ASSERT_THAT(stream->Close(), IsOk()); + EXPECT_TRUE(fs->opened_inputs.back()->closed()); +} + +TEST_F(LocalFileIOTest, CachedInputCreationReleasesTemporaryArrowFile) { + auto fs = std::make_shared(); + auto file_io = std::make_shared(fs); + ASSERT_THAT(file_io->WriteFile(temp_filepath_, "hello"), IsOk()); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + Table::Make({.ns = {}, .name = "test"}, std::make_shared(), + "metadata.json", file_io, std::make_shared(), "", nullptr, + nullptr, + {{std::string(internal::ContentCacheOptions::kEnabled), "true"}})); + + ICEBERG_UNWRAP_OR_FAIL(auto input_file, + table->manifest_io()->NewInputFile(temp_filepath_)); + ASSERT_EQ(fs->opened_inputs.size(), 1); + EXPECT_TRUE(fs->opened_inputs.front()->closed()); + // Opening loads the content into the cache and closes the underlying file. + ICEBERG_UNWRAP_OR_FAIL(auto stream, input_file->Open()); + ASSERT_EQ(fs->opened_inputs.size(), 2); + EXPECT_TRUE(fs->opened_inputs.back()->closed()); + ASSERT_THAT(stream->Close(), IsOk()); +} + TEST_F(LocalFileIOTest, DeleteFile) { auto write_res = file_io_->WriteFile(temp_filepath_, "hello world"); EXPECT_THAT(write_res, IsOk()); @@ -446,6 +566,26 @@ TEST(ArrowFileIOTest, OutputStoredLengthAfterClose) { EXPECT_THAT(output->StoredLength(), HasValue(::testing::Eq(3))); } +TEST(ArrowFileIOTest, OversizedCachedInputRetainsNativeArrowStream) { + std::shared_ptr file_io = arrow::ArrowFileSystemFileIO::MakeMockFileIO(); + ASSERT_THAT(file_io->WriteFile("input", "manifest"), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto uncached, arrow::OpenArrowInputStream(file_io, "input")); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + Table::Make( + {.ns = {}, .name = "test"}, std::make_shared(), "metadata.json", + file_io, std::make_shared(), "", nullptr, nullptr, + {{std::string(internal::ContentCacheOptions::kEnabled), "true"}, + {std::string(internal::ContentCacheOptions::kMaxContentLength), "4"}})); + ICEBERG_UNWRAP_OR_FAIL(auto oversized, + arrow::OpenArrowInputStream(table->manifest_io(), "input")); + + const auto& oversized_stream = *oversized; + const auto& uncached_stream = *uncached; + EXPECT_TRUE(typeid(oversized_stream) == typeid(uncached_stream)); +} + TEST_F(LocalFileIOTest, ResolvesForeignSchemeToUnderlyingPath) { ASSERT_THAT(file_io_->WriteFile(temp_filepath_, "hello world"), IsOk()); diff --git a/src/iceberg/test/cache_test.cc b/src/iceberg/test/cache_test.cc index fbc29ab9c..c406febdd 100644 --- a/src/iceberg/test/cache_test.cc +++ b/src/iceberg/test/cache_test.cc @@ -21,6 +21,7 @@ #include #include #include +#include #include @@ -43,6 +44,116 @@ TEST(LruCacheTest, Eviction) { EXPECT_EQ(*cache.Find(1), "one"); ASSERT_NE(cache.Find(3), nullptr); EXPECT_EQ(*cache.Find(3), "three"); + EXPECT_EQ(cache.size(), 2); + EXPECT_EQ(cache.total_weight(), 2); +} + +struct StringLengthWeigher { + int64_t operator()(int32_t /*key*/, const std::string& value) const { + return static_cast(value.size()); + } +}; + +using WeightedCache = LruCache; + +TEST(LruCacheTest, WeightedEviction) { + WeightedCache cache(10); + + EXPECT_TRUE(cache.Replace(1, "aaaa").first); + EXPECT_TRUE(cache.Replace(2, "bbb").first); + ASSERT_NE(cache.Find(1), nullptr); + EXPECT_EQ(cache.total_weight(), 7); + + // 7 + 6 exceeds 10, so the least recently used entry 2 is evicted. + auto [inserted, value] = cache.Replace(3, "cccccc"); + EXPECT_TRUE(inserted); + ASSERT_NE(value, nullptr); + EXPECT_EQ(*value, "cccccc"); + EXPECT_EQ(cache.Find(2), nullptr); + ASSERT_NE(cache.Find(1), nullptr); + EXPECT_EQ(*cache.Find(1), "aaaa"); + EXPECT_EQ(cache.size(), 2); + EXPECT_EQ(cache.total_weight(), 10); +} + +TEST(LruCacheTest, WeightedEvictionRemovesMultipleEntries) { + WeightedCache cache(10); + cache.Replace(1, "aaa"); + cache.Replace(2, "bbb"); + cache.Replace(3, "ccc"); + + // Making room for 7 requires evicting the two least recently used entries. + ASSERT_NE(cache.Replace(4, "ddddddd").second, nullptr); + EXPECT_EQ(cache.Find(1), nullptr); + EXPECT_EQ(cache.Find(2), nullptr); + EXPECT_NE(cache.Find(3), nullptr); + EXPECT_NE(cache.Find(4), nullptr); + EXPECT_EQ(cache.size(), 2); + EXPECT_EQ(cache.total_weight(), 10); +} + +TEST(LruCacheTest, OversizedValueIsNotCached) { + WeightedCache cache(5); + cache.Replace(1, "aa"); + + auto [inserted, value] = cache.Replace(2, "bbbbbb"); + EXPECT_FALSE(inserted); + EXPECT_EQ(value, nullptr); + EXPECT_EQ(cache.Find(2), nullptr); + // Rejecting the oversized value must not evict anything. + ASSERT_NE(cache.Find(1), nullptr); + EXPECT_EQ(cache.size(), 1); + EXPECT_EQ(cache.total_weight(), 2); + + // A value exactly at the maximum weight fits. + ASSERT_NE(cache.Replace(3, "ccccc").second, nullptr); + EXPECT_EQ(cache.Find(1), nullptr); + EXPECT_EQ(cache.total_weight(), 5); +} + +TEST(LruCacheTest, ReplaceReweighsEntry) { + WeightedCache cache(10); + cache.Replace(1, "aaa"); + cache.Replace(2, "bbb"); + + // Growing entry 2 from 3 to 7 still fits next to entry 1. + EXPECT_FALSE(cache.Replace(2, "bbbbbbb").first); + EXPECT_EQ(cache.size(), 2); + EXPECT_EQ(cache.total_weight(), 10); + + // Growing entry 1 to 8 makes it most recently used and evicts entry 2. + auto [inserted, value] = cache.Replace(1, "aaaaaaaa"); + EXPECT_FALSE(inserted); + ASSERT_NE(value, nullptr); + EXPECT_EQ(*value, "aaaaaaaa"); + EXPECT_EQ(cache.Find(2), nullptr); + EXPECT_EQ(cache.size(), 1); + EXPECT_EQ(cache.total_weight(), 8); + + // Shrinking releases weight. + cache.Replace(1, "a"); + EXPECT_EQ(cache.total_weight(), 1); + + // Replacing with an oversized value removes the existing entry. + std::tie(inserted, value) = cache.Replace(1, std::string(11, 'a')); + EXPECT_FALSE(inserted); + EXPECT_EQ(value, nullptr); + EXPECT_EQ(cache.Find(1), nullptr); + EXPECT_EQ(cache.size(), 0); + EXPECT_EQ(cache.total_weight(), 0); +} + +TEST(LruCacheTest, ClearResetsWeight) { + WeightedCache cache(10); + cache.Replace(1, "aaaa"); + cache.Replace(2, "bbbb"); + + cache.Clear(); + EXPECT_EQ(cache.size(), 0); + EXPECT_EQ(cache.total_weight(), 0); + + ASSERT_NE(cache.Replace(3, std::string(10, 'c')).second, nullptr); + EXPECT_EQ(cache.total_weight(), 10); } TEST(MemoizeLruTest, Caches) { diff --git a/src/iceberg/test/fast_append_test.cc b/src/iceberg/test/fast_append_test.cc index d75e1ef23..98b48160e 100644 --- a/src/iceberg/test/fast_append_test.cc +++ b/src/iceberg/test/fast_append_test.cc @@ -31,6 +31,7 @@ #include #include +#include #include #include diff --git a/src/iceberg/test/manifest_reader_test.cc b/src/iceberg/test/manifest_reader_test.cc index 6a201bd56..311398f1f 100644 --- a/src/iceberg/test/manifest_reader_test.cc +++ b/src/iceberg/test/manifest_reader_test.cc @@ -24,16 +24,22 @@ #include #include #include +#include #include #include #include +#include #include #include +#include "iceberg/arrow/arrow_io_internal.h" #include "iceberg/arrow/arrow_io_util.h" #include "iceberg/avro/avro_register.h" +#include "iceberg/content_cache_internal.h" +#include "iceberg/data_table_scan.h" #include "iceberg/expression/expressions.h" +#include "iceberg/file_io.h" #include "iceberg/file_writer.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/manifest/manifest_list.h" @@ -41,12 +47,36 @@ #include "iceberg/partition_spec.h" #include "iceberg/schema.h" #include "iceberg/schema_internal.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_scan.h" #include "iceberg/test/matchers.h" +#include "iceberg/test/mock_catalog.h" #include "iceberg/transform.h" #include "iceberg/type.h" namespace iceberg { +namespace { + +// Counts how often each path is opened for reading. FileSystem::OpenInputFile(FileInfo) +// forwards to the path overload, so every open is counted exactly once. +class CountingMockFileSystem : public ::arrow::fs::internal::MockFileSystem { + public: + using ::arrow::fs::internal::MockFileSystem::MockFileSystem; + using ::arrow::fs::internal::MockFileSystem::OpenInputFile; + + ::arrow::Result> OpenInputFile( + const std::string& path) override { + ++open_counts[path]; + return ::arrow::fs::internal::MockFileSystem::OpenInputFile(path); + } + + std::unordered_map open_counts; +}; + +} // namespace + class TestManifestReader : public testing::TestWithParam { protected: void SetUp() override { @@ -765,6 +795,130 @@ TEST(ManifestReaderStaticTest, WithStatsColumnsHandlesEmptyAndSelectAll) { testing::ElementsAre(Schema::kAllColumns)); } +TEST_P(TestManifestReader, ReadsManifestListAndManifestFromCachedContent) { + const auto version = GetParam(); + auto fs = std::make_shared(std::chrono::system_clock::now()); + file_io_ = std::make_shared(fs); + auto table_metadata = std::make_shared(); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + Table::Make({.ns = {}, .name = "test"}, table_metadata, "metadata.json", file_io_, + std::make_shared(), "", nullptr, nullptr, + {{std::string(internal::ContentCacheOptions::kEnabled), "true"}})); + + std::vector entries; + entries.push_back(MakeEntry( + ManifestStatus::kAdded, /*snapshot_id=*/1000L, + MakeDataFile("/path/to/data-a.parquet", PartitionValues({Literal::Int(0)})))); + entries.push_back(MakeEntry( + ManifestStatus::kAdded, /*snapshot_id=*/1000L, + MakeDataFile("/path/to/data-b.parquet", PartitionValues({Literal::Int(1)})))); + auto written_manifest = WriteManifest(version, /*snapshot_id=*/1000L, entries); + + const std::string manifest_list_path = std::format("snap-{}.avro", MakeManifestPath()); + ICEBERG_UNWRAP_OR_FAIL( + auto list_writer, + ManifestListWriter::MakeWriter(version, /*snapshot_id=*/1000L, + /*parent_snapshot_id=*/std::nullopt, + manifest_list_path, file_io_, + /*sequence_number=*/1, /*first_row_id=*/0L)); + ASSERT_THAT(list_writer->Add(written_manifest), IsOk()); + ASSERT_THAT(list_writer->Close(), IsOk()); + + auto read_manifest_list = [&]() -> std::vector { + auto reader = ManifestListReader::Make(manifest_list_path, table->manifest_io()); + EXPECT_THAT(reader, IsOk()); + if (!reader.has_value()) { + return {}; + } + auto files = reader.value()->Files(); + EXPECT_THAT(files, IsOk()); + return files.has_value() ? std::move(files).value() : std::vector{}; + }; + auto read_data_file_paths = [&](const ManifestFile& manifest) { + std::vector paths; + auto reader = ManifestReader::Make(manifest, table->manifest_io(), schema_, spec_); + EXPECT_THAT(reader, IsOk()); + if (!reader.has_value()) { + return paths; + } + auto read_entries = reader.value()->Entries(); + EXPECT_THAT(read_entries, IsOk()); + if (read_entries.has_value()) { + for (const auto& entry : read_entries.value()) { + EXPECT_EQ(entry.status, ManifestStatus::kAdded); + EXPECT_EQ(entry.snapshot_id, 1000L); + paths.push_back(entry.data_file->file_path); + } + } + return paths; + }; + const std::vector expected_paths = {"/path/to/data-a.parquet", + "/path/to/data-b.parquet"}; + + // First read: both files are loaded from the file system into the cache. + auto first_files = read_manifest_list(); + ASSERT_EQ(first_files.size(), 1); + EXPECT_EQ(first_files[0].manifest_path, written_manifest.manifest_path); + EXPECT_EQ(first_files[0].added_snapshot_id, 1000L); + EXPECT_EQ(read_data_file_paths(first_files[0]), expected_paths); + const int list_opens = fs->open_counts[manifest_list_path]; + const int manifest_opens = fs->open_counts[written_manifest.manifest_path]; + EXPECT_GT(list_opens, 0); + EXPECT_GT(manifest_opens, 0); + + // Second read: the same rows are decoded from cached content without opening either + // file again. + auto second_files = read_manifest_list(); + ASSERT_EQ(second_files.size(), 1); + EXPECT_EQ(second_files[0], first_files[0]); + EXPECT_EQ(read_data_file_paths(second_files[0]), expected_paths); + EXPECT_EQ(fs->open_counts[manifest_list_path], list_opens); + EXPECT_EQ(fs->open_counts[written_manifest.manifest_path], manifest_opens); + + // The public scan builder must carry this same table-owned cache into both + // manifest-list and manifest readers. Each builder captures its own read context. + table_metadata->format_version = version; + table_metadata->schemas = {schema_}; + table_metadata->current_schema_id = schema_->schema_id(); + table_metadata->partition_specs = {spec_}; + table_metadata->default_spec_id = spec_->spec_id(); + table_metadata->current_snapshot_id = 1000; + table_metadata->snapshots = {std::make_shared(Snapshot{ + .snapshot_id = 1000, + .sequence_number = 1, + .manifest_list = manifest_list_path, + .summary = {{"operation", "append"}}, + .schema_id = schema_->schema_id(), + })}; + for (int iteration = 0; iteration < 2; ++iteration) { + ICEBERG_UNWRAP_OR_FAIL(auto builder, table->NewScan()); + ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); + EXPECT_EQ(scan->io(), file_io_); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles()); + ASSERT_EQ(tasks.size(), 2); + EXPECT_EQ(tasks[0]->data_file()->file_path, expected_paths[0]); + EXPECT_EQ(tasks[1]->data_file()->file_path, expected_paths[1]); + } + EXPECT_EQ(fs->open_counts[manifest_list_path], list_opens); + EXPECT_EQ(fs->open_counts[written_manifest.manifest_path], manifest_opens); + + ICEBERG_UNWRAP_OR_FAIL(auto captured_builder, table->NewScan()); + table.reset(); + ICEBERG_UNWRAP_OR_FAIL(auto captured_scan, captured_builder->Build()); + ICEBERG_UNWRAP_OR_FAIL(auto captured_tasks, captured_scan->PlanFiles()); + EXPECT_EQ(captured_tasks.size(), 2); + EXPECT_EQ(fs->open_counts[manifest_list_path], list_opens); + EXPECT_EQ(fs->open_counts[written_manifest.manifest_path], manifest_opens); + + // Control: without caching, reading the manifest opens it again. + ICEBERG_UNWRAP_OR_FAIL( + table, Table::Make({.ns = {}, .name = "test"}, table_metadata, "metadata.json", + file_io_, std::make_shared())); + EXPECT_EQ(read_data_file_paths(second_files[0]), expected_paths); + EXPECT_GT(fs->open_counts[written_manifest.manifest_path], manifest_opens); +} + INSTANTIATE_TEST_SUITE_P(ManifestReaderVersions, TestManifestReader, testing::Values(1, 2, 3)); diff --git a/src/iceberg/test/metadata_cache_test.cc b/src/iceberg/test/metadata_cache_test.cc new file mode 100644 index 000000000..d98bf1629 --- /dev/null +++ b/src/iceberg/test/metadata_cache_test.cc @@ -0,0 +1,668 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "iceberg/content_cache_internal.h" +#include "iceberg/file_io.h" +#include "iceberg/file_io_registry.h" +#include "iceberg/file_reader.h" +#include "iceberg/schema.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/mock_catalog.h" +#include "iceberg/test/mock_io.h" +#include "iceberg/util/macros.h" + +namespace iceberg { +namespace { +using internal::ContentCacheOptions; + +class CountingInputStream : public SeekableInputStream { + public: + CountingInputStream(std::unique_ptr stream, int* read_count) + : stream_(std::move(stream)), read_count_(read_count) {} + + Result Position() const override { return stream_->Position(); } + Status Seek(int64_t position) override { return stream_->Seek(position); } + + Result Read(std::span out) override { + ++*read_count_; + return stream_->Read(out); + } + + Status ReadFully(int64_t position, std::span out) override { + ++*read_count_; + return stream_->ReadFully(position, out); + } + + Status Close() override { return stream_->Close(); } + + private: + std::unique_ptr stream_; + int* read_count_; +}; + +class CountingInputFile : public InputFile { + public: + CountingInputFile(std::unique_ptr input_file, int* open_count, + int* size_count, int* read_count, int* failing_opens) + : input_file_(std::move(input_file)), + open_count_(open_count), + size_count_(size_count), + read_count_(read_count), + failing_opens_(failing_opens) {} + + std::string_view location() const override { return input_file_->location(); } + + Result Size() const override { + ++*size_count_; + return input_file_->Size(); + } + + Result> Open() override { + ++*open_count_; + if (*failing_opens_ > 0) { + --*failing_opens_; + return NotFound("Simulated open failure"); + } + ICEBERG_ASSIGN_OR_RAISE(auto stream, input_file_->Open()); + return std::make_unique(std::move(stream), read_count_); + } + + private: + std::unique_ptr input_file_; + int* open_count_; + int* size_count_; + int* read_count_; + int* failing_opens_; +}; + +class CountingMockFileIO : public MockFileIO { + public: + Result> NewInputFile(std::string file_location) override { + ICEBERG_ASSIGN_OR_RAISE(auto input_file, + MockFileIO::NewInputFile(std::move(file_location))); + return std::make_unique(std::move(input_file), &open_count, + &size_count, &read_count, &failing_opens); + } + + int open_count = 0; + int size_count = 0; + int read_count = 0; + // Number of upcoming Open calls that fail with kNotFound. + int failing_opens = 0; +}; + +// Create independently configured tables sharing the same storage FileIO. +Result> MakeCachedTable( + FileIO& io, const std::unordered_map& properties = {}, + std::string metadata_location = "metadata.json") { + auto metadata = std::make_shared(); + metadata->table_uuid = "cache-test-table"; + auto storage = std::shared_ptr(&io, [](FileIO*) {}); + return Table::Make({.ns = {}, .name = "test"}, std::move(metadata), + std::move(metadata_location), std::move(storage), + std::make_shared(), "", nullptr, nullptr, properties); +} + +// Read a whole immutable file through a table's manifest I/O context. +Result ReadCached(FileIO& file_io, const std::string& location) { + ICEBERG_ASSIGN_OR_RAISE(auto input_file, file_io.NewInputFile(location)); + ICEBERG_ASSIGN_OR_RAISE(auto size, input_file->Size()); + ICEBERG_ASSIGN_OR_RAISE(auto stream, input_file->Open()); + std::string content(static_cast(size), '\0'); + ICEBERG_RETURN_UNEXPECTED( + stream->ReadFully(/*position=*/0, std::as_writable_bytes(std::span(content)))); + ICEBERG_RETURN_UNEXPECTED(stream->Close()); + return content; +} + +class NormalizingMockFileIO : public CountingMockFileIO { + public: + Result> NewInputFile(std::string file_location) override { + constexpr std::string_view kAliasPrefix = "alias://"; + if (file_location.starts_with(kAliasPrefix)) { + file_location.erase(0, kAliasPrefix.size()); + } + return CountingMockFileIO::NewInputFile(std::move(file_location)); + } +}; + +struct FailingOpenState { + int open_count = 0; + Error first_error; + Error fallback_error; +}; + +class FailingOpenInputFile : public InputFile { + public: + explicit FailingOpenInputFile(std::shared_ptr state) + : state_(std::move(state)) {} + + std::string_view location() const override { return "manifest.avro"; } + Result Size() const override { return 8; } + + Result> Open() override { + ++state_->open_count; + if (state_->open_count == 1) { + return unexpected(state_->first_error); + } + return unexpected(state_->fallback_error); + } + + private: + std::shared_ptr state_; +}; + +class FailingOpenFileIO : public FileIO { + public: + explicit FailingOpenFileIO(std::shared_ptr state) + : state_(std::move(state)) {} + + Result> NewInputFile( + std::string /*file_location*/) override { + return std::make_unique(state_); + } + + private: + std::shared_ptr state_; +}; + +TEST(MetadataCacheTest, TableReusesCachedContentAcrossInputFiles) { + CountingMockFileIO file_io; + file_io.AddFile("manifest.avro", "manifest"); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + MakeCachedTable(file_io, {{std::string(ContentCacheOptions::kEnabled), "true"}})); + + auto first_file = table->manifest_io()->NewInputFile("manifest.avro"); + ASSERT_THAT(first_file, IsOk()); + auto first_stream = first_file.value()->Open(); + ASSERT_THAT(first_stream, IsOk()); + ASSERT_THAT(first_stream.value()->Close(), IsOk()); + + auto second_file = table->manifest_io()->NewInputFile("manifest.avro"); + ASSERT_THAT(second_file, IsOk()); + auto second_stream = second_file.value()->Open(); + ASSERT_THAT(second_stream, IsOk()); + ASSERT_THAT(second_stream.value()->Close(), IsOk()); + + EXPECT_EQ(file_io.open_count, 1); +} + +TEST(MetadataCacheTest, TableUsesNormalizedLocationForCacheHits) { + NormalizingMockFileIO file_io; + file_io.AddFile("manifest.avro", "manifest"); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + MakeCachedTable(file_io, {{std::string(ContentCacheOptions::kEnabled), "true"}})); + + for (const auto* location : + {"alias://manifest.avro", "alias://manifest.avro", "manifest.avro"}) { + ICEBERG_UNWRAP_OR_FAIL(auto input_file, table->manifest_io()->NewInputFile(location)); + EXPECT_EQ(input_file->location(), "manifest.avro"); + ICEBERG_UNWRAP_OR_FAIL(auto stream, input_file->Open()); + ASSERT_THAT(stream->Close(), IsOk()); + } + + EXPECT_EQ(file_io.size_count, 1); + EXPECT_EQ(file_io.open_count, 1); +} + +TEST(MetadataCacheTest, CachedInputFileDoesNotRetryNonIOErrors) { + auto state = std::make_shared(FailingOpenState{ + .first_error = {.kind = ErrorKind::kNotFound, .message = "missing"}, + .fallback_error = {.kind = ErrorKind::kIOError, .message = "unexpected retry"}, + }); + FailingOpenFileIO file_io(state); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + MakeCachedTable(file_io, {{std::string(ContentCacheOptions::kEnabled), "true"}})); + ICEBERG_UNWRAP_OR_FAIL(auto input_file, + table->manifest_io()->NewInputFile("manifest.avro")); + + EXPECT_THAT(input_file->Open(), IsError(ErrorKind::kNotFound)); + EXPECT_EQ(state->open_count, 1); +} + +TEST(MetadataCacheTest, CachedInputFilePreservesReadAheadAndFallbackErrors) { + auto state = std::make_shared(FailingOpenState{ + .first_error = {.kind = ErrorKind::kIOError, .message = "read-ahead failed"}, + .fallback_error = {.kind = ErrorKind::kIOError, .message = "open failed"}, + }); + FailingOpenFileIO file_io(state); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + MakeCachedTable(file_io, {{std::string(ContentCacheOptions::kEnabled), "true"}})); + ICEBERG_UNWRAP_OR_FAIL(auto input_file, + table->manifest_io()->NewInputFile("manifest.avro")); + + auto result = input_file->Open(); + EXPECT_THAT(result, IsError(ErrorKind::kIOError)); + EXPECT_THAT(result, HasErrorMessage("read-ahead failed")); + EXPECT_THAT(result, HasErrorMessage("fallback open failed: open failed")); + EXPECT_EQ(state->open_count, 2); +} + +TEST(MetadataCacheTest, TableDoesNotCacheLoadFailures) { + CountingMockFileIO file_io; + file_io.AddFile("manifest.avro", "manifest"); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + MakeCachedTable(file_io, {{std::string(ContentCacheOptions::kEnabled), "true"}})); + file_io.failing_opens = 1; + + EXPECT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), + IsError(ErrorKind::kNotFound)); + EXPECT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), + HasValue(::testing::Eq("manifest"))); + EXPECT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), + HasValue(::testing::Eq("manifest"))); + // The failed open, then the successful load. The last read comes from the cache. + EXPECT_EQ(file_io.open_count, 2); +} + +TEST(MetadataCacheTest, TableReadsUncacheableFilesDirectly) { + for (auto limit : + {ContentCacheOptions::kMaxTotalBytes, ContentCacheOptions::kMaxContentLength}) { + SCOPED_TRACE(limit); + CountingMockFileIO file_io; + file_io.AddFile("manifest.avro", "manifest"); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + MakeCachedTable(file_io, {{std::string(ContentCacheOptions::kEnabled), "true"}, + {std::string(limit), "4"}})); + + // The 8-byte file can never be cached, so opening it must not load the whole file + // into memory first. + ICEBERG_UNWRAP_OR_FAIL(auto input_file, + table->manifest_io()->NewInputFile("manifest.avro")); + ICEBERG_UNWRAP_OR_FAIL(auto stream, input_file->Open()); + EXPECT_EQ(file_io.read_count, 0); + ASSERT_THAT(stream->Close(), IsOk()); + + EXPECT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), + HasValue(::testing::Eq("manifest"))); + EXPECT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), + HasValue(::testing::Eq("manifest"))); + EXPECT_EQ(file_io.open_count, 3); + } +} + +TEST(MetadataCacheTest, TableEvictsLeastRecentlyUsedContentByBytes) { + CountingMockFileIO file_io; + file_io.AddFile("a", "aa"); + file_io.AddFile("b", "bbb"); + file_io.AddFile("c", "cccc"); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + MakeCachedTable(file_io, + {{std::string(ContentCacheOptions::kEnabled), "true"}, + {std::string(ContentCacheOptions::kMaxTotalBytes), "6"}})); + + // Return the number of underlying opens after reading `location`. + auto read = [&file_io, &table](const std::string& location) { + EXPECT_THAT(ReadCached(*table->manifest_io(), location), IsOk()); + return file_io.open_count; + }; + EXPECT_EQ(read("a"), 1); + EXPECT_EQ(read("b"), 2); + EXPECT_EQ(read("a"), 2); // Cached, and now the most recently used. + EXPECT_EQ(read("c"), 3); // 2 + 3 + 4 bytes exceed 6, so "b" is evicted. + EXPECT_EQ(read("a"), 3); + EXPECT_EQ(read("c"), 3); + EXPECT_EQ(read("b"), 4); +} + +TEST(MetadataCacheTest, TableClearsCachedContent) { + CountingMockFileIO file_io; + file_io.AddFile("snap.avro", "manifest-list"); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + MakeCachedTable(file_io, {{std::string(ContentCacheOptions::kEnabled), "true"}})); + ASSERT_THAT(ReadCached(*table->manifest_io(), "snap.avro"), IsOk()); + + table->ClearManifestCache(); + + ASSERT_THAT(ReadCached(*table->manifest_io(), "snap.avro"), IsOk()); + ASSERT_THAT(ReadCached(*table->manifest_io(), "snap.avro"), IsOk()); + EXPECT_EQ(file_io.open_count, 2); +} + +TEST(MetadataCacheTest, TableConcurrentReadsReturnFileContent) { + // CountingMockFileIO's counters are not thread-safe, so use the plain mock. + MockFileIO file_io; + constexpr int kLocations = 8; + std::vector contents; + for (int id = 0; id < kLocations; ++id) { + contents.emplace_back(static_cast(id % 4 + 1), static_cast('a' + id)); + file_io.AddFile(std::to_string(id), contents.back()); + } + // The files add up to 20 bytes, so reads keep evicting from the 16-byte cache. + ICEBERG_UNWRAP_OR_FAIL( + auto table, + MakeCachedTable(file_io, + {{std::string(ContentCacheOptions::kEnabled), "true"}, + {std::string(ContentCacheOptions::kMaxTotalBytes), "16"}})); + + constexpr int kThreads = 4; + constexpr int kIterations = 1000; + std::atomic mismatch = false; + std::vector> workers; + for (int thread = 0; thread < kThreads; ++thread) { + workers.push_back(std::async(std::launch::async, [&, thread] { + for (int i = 0; i < kIterations; ++i) { + const int id = (thread + i) % kLocations; + auto content = ReadCached(*table->manifest_io(), std::to_string(id)); + if (!content.has_value() || content.value() != contents[id]) { + mismatch.store(true, std::memory_order_relaxed); + } + } + })); + } + for (auto& worker : workers) { + worker.get(); + } + + EXPECT_FALSE(mismatch.load(std::memory_order_relaxed)); +} + +TEST(MetadataCacheTest, ParsesJavaCompatibleProperties) { + std::unordered_map properties = { + {std::string(ContentCacheOptions::kEnabled), "true"}, + {std::string(ContentCacheOptions::kMaxTotalBytes), "200"}, + {std::string(ContentCacheOptions::kMaxContentLength), "50"}, + }; + + auto options = ContentCacheOptions::FromProperties(properties); + + ASSERT_THAT(options, IsOk()); + EXPECT_TRUE(options->enabled); + EXPECT_EQ(options->max_total_bytes, 200); + EXPECT_EQ(options->max_content_length, 50); +} + +TEST(MetadataCacheTest, ParsesEnabledPropertyIgnoringCase) { + for (const auto* value : {"true", "TRUE", "True", "tRuE"}) { + SCOPED_TRACE(value); + ICEBERG_UNWRAP_OR_FAIL(auto options, + ContentCacheOptions::FromProperties( + {{std::string(ContentCacheOptions::kEnabled), value}})); + EXPECT_TRUE(options.enabled); + } +} + +TEST(MetadataCacheTest, TreatsOtherEnabledPropertyValuesAsFalseLikeJava) { + for (const auto* value : + {"false", "FALSE", "False", "fAlSe", "", "yes", "1", "ture", " true "}) { + SCOPED_TRACE(value); + ICEBERG_UNWRAP_OR_FAIL(auto options, + ContentCacheOptions::FromProperties( + {{std::string(ContentCacheOptions::kEnabled), value}})); + EXPECT_FALSE(options.enabled); + } +} + +TEST(MetadataCacheTest, UsesJavaCompatibleDefaults) { + auto options = ContentCacheOptions::FromProperties( + {{std::string(ContentCacheOptions::kEnabled), "true"}}); + + ASSERT_THAT(options, IsOk()); + EXPECT_TRUE(options->enabled); + EXPECT_EQ(options->max_total_bytes, 100 * 1024 * 1024); + EXPECT_EQ(options->max_content_length, 8 * 1024 * 1024); +} + +TEST(MetadataCacheTest, InvalidNumericPropertyErrorIncludesPropertyName) { + auto options = ContentCacheOptions::FromProperties( + {{std::string(ContentCacheOptions::kEnabled), "true"}, + {std::string(ContentCacheOptions::kMaxTotalBytes), "invalid"}}); + + EXPECT_THAT(options, HasErrorMessage(std::string(ContentCacheOptions::kMaxTotalBytes))); +} + +TEST(MetadataCacheTest, ParsesTuningPropertiesWhenDisabled) { + auto options = ContentCacheOptions::FromProperties( + {{std::string(ContentCacheOptions::kEnabled), "false"}, + {std::string(ContentCacheOptions::kMaxTotalBytes), "1024"}}); + + ASSERT_THAT(options, IsOk()); + EXPECT_FALSE(options->enabled); + EXPECT_EQ(options->max_total_bytes, 1024); +} + +TEST(MetadataCacheTest, RejectsInvalidNumericPropertyWhenDisabled) { + auto options = ContentCacheOptions::FromProperties( + {{std::string(ContentCacheOptions::kEnabled), "false"}, + {std::string(ContentCacheOptions::kMaxTotalBytes), "invalid"}}); + + EXPECT_THAT(options, HasErrorMessage(std::string(ContentCacheOptions::kMaxTotalBytes))); +} + +TEST(MetadataCacheTest, ReaderOptionsKeepsPositionalAggregateInitialization) { + auto io = std::make_shared(); + auto schema = std::make_shared(std::vector{}); + ReaderOptions options{"manifest.avro", {}, {}, io, schema}; + EXPECT_EQ(options.path, "manifest.avro"); + EXPECT_EQ(options.io, io); + EXPECT_EQ(options.projection, schema); +} + +class LengthTrackingFileIO : public MockFileIO { + public: + using MockFileIO::NewInputFile; + int length_calls = 0; + Result> NewInputFile(std::string location, + size_t length) override { + ++length_calls; + return MockFileIO::NewInputFile(std::move(location), length); + } +}; + +TEST(MetadataCacheTest, RejectsOversizedLengthBeforeCallingStorage) { + if constexpr (sizeof(size_t) <= sizeof(int32_t)) return; + LengthTrackingFileIO io; + ICEBERG_UNWRAP_OR_FAIL( + auto table, + MakeCachedTable(io, {{std::string(ContentCacheOptions::kEnabled), "true"}})); + EXPECT_THAT(table->manifest_io()->NewInputFile("manifest.avro", + std::numeric_limits::max()), + IsError(ErrorKind::kInvalidArgument)); + EXPECT_EQ(io.length_calls, 0); +} + +TEST(MetadataCacheTest, DisabledTableUsesStorageDirectly) { + CountingMockFileIO io; + io.AddFile("manifest.avro", "manifest"); + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeCachedTable(io)); + EXPECT_FALSE(table->ManifestCacheEnabled()); + EXPECT_EQ(table->manifest_io(), table->io()); + ASSERT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), IsOk()); + ASSERT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), IsOk()); + EXPECT_EQ(io.open_count, 2); +} + +TEST(MetadataCacheTest, TablesSharingStorageHaveIndependentCachesAndLimits) { + CountingMockFileIO io; + io.AddFile("manifest.avro", "manifest"); + ICEBERG_UNWRAP_OR_FAIL( + auto first, + MakeCachedTable(io, {{std::string(ContentCacheOptions::kEnabled), "true"}, + {std::string(ContentCacheOptions::kMaxTotalBytes), "16"}})); + ICEBERG_UNWRAP_OR_FAIL( + auto second, + MakeCachedTable(io, {{std::string(ContentCacheOptions::kEnabled), "true"}, + {std::string(ContentCacheOptions::kMaxTotalBytes), "4"}})); + ASSERT_THAT(ReadCached(*first->manifest_io(), "manifest.avro"), IsOk()); + ASSERT_THAT(ReadCached(*second->manifest_io(), "manifest.avro"), IsOk()); + ASSERT_THAT(ReadCached(*first->manifest_io(), "manifest.avro"), IsOk()); + ASSERT_THAT(ReadCached(*second->manifest_io(), "manifest.avro"), IsOk()); + EXPECT_EQ(io.open_count, 3); + // Ordinary FileIO input remains independent of either table's cache. + ASSERT_THAT(ReadCached(io, "manifest.avro"), IsOk()); + EXPECT_EQ(io.open_count, 4); +} + +TEST(MetadataCacheTest, CompatibleRefreshPreservesCachedContent) { + CountingMockFileIO io; + io.AddFile("manifest.avro", "manifest"); + std::unordered_map props = { + {std::string(ContentCacheOptions::kEnabled), "true"}}; + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeCachedTable(io, props)); + ASSERT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto refreshed, MakeCachedTable(io, props, "metadata-2.json")); + auto catalog = std::dynamic_pointer_cast(table->catalog()); + EXPECT_CALL(*catalog, LoadTable(::testing::_)).WillOnce(::testing::Return(refreshed)); + ASSERT_THAT(table->Refresh(), IsOk()); + ASSERT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), IsOk()); + EXPECT_EQ(io.open_count, 1); + table->ClearManifestCache(); + ASSERT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), IsOk()); + EXPECT_EQ(io.open_count, 2); +} + +TEST(MetadataCacheTest, RefreshWithChangedLimitsReplacesCache) { + CountingMockFileIO io; + io.AddFile("manifest.avro", "manifest"); + std::unordered_map props = { + {std::string(ContentCacheOptions::kEnabled), "true"}}; + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeCachedTable(io, props)); + ASSERT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), IsOk()); + props[std::string(ContentCacheOptions::kMaxTotalBytes)] = "4"; + ICEBERG_UNWRAP_OR_FAIL(auto refreshed, MakeCachedTable(io, props, "metadata-2.json")); + auto catalog = std::dynamic_pointer_cast(table->catalog()); + EXPECT_CALL(*catalog, LoadTable(::testing::_)).WillOnce(::testing::Return(refreshed)); + ASSERT_THAT(table->Refresh(), IsOk()); + ASSERT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), IsOk()); + ASSERT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), IsOk()); + EXPECT_EQ(io.open_count, 3); +} + +TEST(MetadataCacheTest, RefreshWithDifferentStorageDiscardsCachedBytes) { + CountingMockFileIO first_io; + CountingMockFileIO second_io; + first_io.AddFile("manifest.avro", "first"); + second_io.AddFile("manifest.avro", "second"); + std::unordered_map props = { + {std::string(ContentCacheOptions::kEnabled), "true"}}; + ICEBERG_UNWRAP_OR_FAIL(auto table, MakeCachedTable(first_io, props)); + ASSERT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), + HasValue(::testing::Eq("first"))); + ICEBERG_UNWRAP_OR_FAIL(auto refreshed, + MakeCachedTable(second_io, props, "metadata-2.json")); + auto catalog = std::dynamic_pointer_cast(table->catalog()); + EXPECT_CALL(*catalog, LoadTable(::testing::_)).WillOnce(::testing::Return(refreshed)); + ASSERT_THAT(table->Refresh(), IsOk()); + EXPECT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), + HasValue(::testing::Eq("second"))); +} + +TEST(MetadataCacheTest, RejectsInvalidCacheLimitsAtTableCreation) { + for (const auto key : + {ContentCacheOptions::kMaxTotalBytes, ContentCacheOptions::kMaxContentLength}) { + for (const auto value : {"0", "-1", "9223372036854775808", "invalid"}) { + CountingMockFileIO io; + EXPECT_THAT( + MakeCachedTable(io, {{std::string(ContentCacheOptions::kEnabled), "true"}, + {std::string(key), value}}), + IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT( + MakeCachedTable(io, {{std::string(ContentCacheOptions::kEnabled), "true"}}), + IsOk()); + } + } +} + +TEST(MetadataCacheTest, EmptyFilesAreNotRetained) { + CountingMockFileIO io; + ICEBERG_UNWRAP_OR_FAIL( + auto table, + MakeCachedTable(io, {{std::string(ContentCacheOptions::kEnabled), "true"}, + {std::string(ContentCacheOptions::kMaxTotalBytes), "1"}})); + for (int i = 0; i < 100; ++i) { + io.AddFile(std::to_string(i), ""); + ASSERT_THAT(ReadCached(*table->manifest_io(), std::to_string(i)), IsOk()); + } + ASSERT_THAT(ReadCached(*table->manifest_io(), "0"), IsOk()); + EXPECT_EQ(io.open_count, 101); +} + +class UnavailableFileIO : public CountingMockFileIO { + public: + bool unavailable = false; + Result> NewInputFile(std::string location) override { + if (unavailable) return IOError("storage resolution unavailable"); + return CountingMockFileIO::NewInputFile(std::move(location)); + } +}; + +class UnavailableNormalizingFileIO : public NormalizingMockFileIO { + public: + bool unavailable = false; + Result> NewInputFile(std::string location) override { + if (unavailable) return IOError("storage resolution unavailable"); + return NormalizingMockFileIO::NewInputFile(std::move(location)); + } +}; + +TEST(MetadataCacheTest, NormalizedAliasHitDoesNotResolveStorage) { + UnavailableNormalizingFileIO io; + io.AddFile("manifest.avro", "manifest"); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + MakeCachedTable(io, {{std::string(ContentCacheOptions::kEnabled), "true"}})); + ASSERT_THAT(ReadCached(*table->manifest_io(), "alias://manifest.avro"), IsOk()); + io.unavailable = true; + EXPECT_THAT(ReadCached(*table->manifest_io(), "alias://manifest.avro"), + HasValue(::testing::Eq("manifest"))); + EXPECT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), + HasValue(::testing::Eq("manifest"))); + EXPECT_EQ(io.open_count, 1); +} + +TEST(MetadataCacheTest, HitDoesNotResolveStorage) { + UnavailableFileIO io; + io.AddFile("manifest.avro", "manifest"); + ICEBERG_UNWRAP_OR_FAIL( + auto table, + MakeCachedTable(io, {{std::string(ContentCacheOptions::kEnabled), "true"}})); + ASSERT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), IsOk()); + io.unavailable = true; + EXPECT_THAT(ReadCached(*table->manifest_io(), "manifest.avro"), + HasValue(::testing::Eq("manifest"))); +} + +} // namespace +} // namespace iceberg diff --git a/src/iceberg/test/transaction_test.cc b/src/iceberg/test/transaction_test.cc index 3a1e3bcdd..16a5e356d 100644 --- a/src/iceberg/test/transaction_test.cc +++ b/src/iceberg/test/transaction_test.cc @@ -81,6 +81,29 @@ TEST_F(TransactionTest, CommitTransactionWithPropertyUpdate) { EXPECT_EQ(props.at("txn.property"), "txn.value"); } +TEST_F(TransactionTest, CommitPreservesTableManifestContentCache) { + ICEBERG_UNWRAP_OR_FAIL(auto enable_txn, table_->NewTransaction()); + ICEBERG_UNWRAP_OR_FAIL(auto enable, enable_txn->NewUpdateProperties()); + enable->Set("io.manifest.cache-enabled", "true"); + ASSERT_THAT(enable->Commit(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(table_, enable_txn->Commit()); + ASSERT_TRUE(table_->ManifestCacheEnabled()); + + const auto manifest_path = table_location_ + "/metadata/cached-manifest.avro"; + ASSERT_THAT(file_io_->WriteFile(manifest_path, "manifest bytes"), IsOk()); + ASSERT_THAT(table_->manifest_io()->ReadFile(manifest_path, std::nullopt), + HasValue(::testing::Eq("manifest bytes"))); + + ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction()); + ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewUpdateProperties()); + update->Set("txn.property", "txn.value"); + ASSERT_THAT(update->Commit(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto committed, txn->Commit()); + ASSERT_THAT(file_io_->DeleteFile(manifest_path), IsOk()); + EXPECT_THAT(committed->manifest_io()->ReadFile(manifest_path, std::nullopt), + HasValue(::testing::Eq("manifest bytes"))); +} + TEST_F(TransactionTest, MultipleUpdatesInTransaction) { ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction()); diff --git a/src/iceberg/transaction.cc b/src/iceberg/transaction.cc index 0eaa3c976..13903203a 100644 --- a/src/iceberg/transaction.cc +++ b/src/iceberg/transaction.cc @@ -506,6 +506,7 @@ Result> Transaction::Commit() { return commit_result; } + commit_result.value()->ReuseManifestCache(*ctx_->table); ctx_->table = std::move(commit_result.value()); state_ = TransactionState::kCommitted; FinalizeUpdates(*ctx_->table->metadata()); diff --git a/src/iceberg/update/cherry_pick_operation.cc b/src/iceberg/update/cherry_pick_operation.cc index d0a330760..3856c98da 100644 --- a/src/iceberg/update/cherry_pick_operation.cc +++ b/src/iceberg/update/cherry_pick_operation.cc @@ -241,7 +241,7 @@ CherryPickOperation& CherryPickOperation::Cherrypick(int64_t snapshot_id) { replaced_partitions_.emplace(); } - auto io = ctx_->table->io(); + auto io = ctx_->table->manifest_io(); ICEBERG_BUILDER_ASSIGN_OR_RETURN( auto changes, ReadSnapshotChanges(*cherrypick_snapshot_, io, current)); @@ -306,7 +306,7 @@ Status CherryPickOperation::ValidateReplacedPartitions(const TableMetadata& meta // Walk back from the current snapshot to the picked snapshot's parent and // reject any file added into a partition this pick replaces. - auto io = ctx_->table->io(); + auto io = ctx_->table->manifest_io(); ICEBERG_ASSIGN_OR_RAISE(auto snapshots, SnapshotUtil::AncestorsBetween( meta, meta.current_snapshot_id, parent_id)); // TODO: Parallelize added-file validation using the planning executor. diff --git a/src/iceberg/update/expire_snapshots.cc b/src/iceberg/update/expire_snapshots.cc index fd14308e8..8805f6d15 100644 --- a/src/iceberg/update/expire_snapshots.cc +++ b/src/iceberg/update/expire_snapshots.cc @@ -978,8 +978,8 @@ Result ExpireSnapshots::Apply() { std::unordered_set schema_ids; ICEBERG_ASSIGN_OR_RAISE(auto snapshot, base.SnapshotById(snapshot_id)); SnapshotReader snapshot_reader(snapshot.get()); - ICEBERG_ASSIGN_OR_RAISE(auto manifests, - snapshot_reader.Manifests(ctx_->table->io())); + ICEBERG_ASSIGN_OR_RAISE( + auto manifests, snapshot_reader.Manifests(ctx_->table->manifest_io())); for (const auto& manifest : manifests) { spec_ids.insert(manifest.partition_spec_id); } @@ -1038,12 +1038,12 @@ Status ExpireSnapshots::Finalize(const TableMetadata& metadata_after_expiration) !HasNonMainSnapshots(metadata_after_expiration); if (can_use_incremental) { - return IncrementalFileCleanup(ctx_->table->io(), delete_func_, delete_executor_, - plan_executor_) + return IncrementalFileCleanup(ctx_->table->manifest_io(), delete_func_, + delete_executor_, plan_executor_) .CleanFiles(metadata_before_expiration, metadata_after_expiration, cleanup_level_); } - return ReachableFileCleanup(ctx_->table->io(), delete_func_, delete_executor_, + return ReachableFileCleanup(ctx_->table->manifest_io(), delete_func_, delete_executor_, plan_executor_) .CleanFiles(metadata_before_expiration, metadata_after_expiration, cleanup_level_); } diff --git a/src/iceberg/update/fast_append.cc b/src/iceberg/update/fast_append.cc index 1cb1e41cf..58476556e 100644 --- a/src/iceberg/update/fast_append.cc +++ b/src/iceberg/update/fast_append.cc @@ -130,7 +130,7 @@ Result> FastAppend::Apply( if (snapshot != nullptr) { auto snapshot_reader = SnapshotReader(snapshot.get()); ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests, - snapshot_reader.Manifests(ctx_->table->io())); + snapshot_reader.Manifests(ctx_->table->manifest_io())); manifests.insert(manifests.end(), snapshot_manifests.begin(), snapshot_manifests.end()); } @@ -200,8 +200,8 @@ Result FastAppend::CopyManifest(const ManifestFile& manifest) { int64_t snapshot_id = SnapshotId(); // Copy the manifest with the new snapshot ID. - return CopyAppendManifest(manifest, ctx_->table->io(), schema, spec, snapshot_id, - new_manifest_path, current.format_version, + return CopyAppendManifest(manifest, ctx_->table->manifest_io(), schema, spec, + snapshot_id, new_manifest_path, current.format_version, &appended_manifests_summary_); } diff --git a/src/iceberg/update/merging_snapshot_update.cc b/src/iceberg/update/merging_snapshot_update.cc index 154a27b91..bcb31fbcc 100644 --- a/src/iceberg/update/merging_snapshot_update.cc +++ b/src/iceberg/update/merging_snapshot_update.cc @@ -409,7 +409,7 @@ MergingSnapshotUpdate::MergingSnapshotUpdate(std::string table_name, : SnapshotUpdate(std::move(ctx)), table_name_(std::move(table_name)), delete_expression_(Expressions::AlwaysFalse()) { - auto file_io = ctx_->table->io(); + auto file_io = ctx_->table->manifest_io(); auto data_filter_manager = ManifestFilterManager::Make( ManifestContent::kData, file_io, [this](const std::string& location) { return DeleteFile(location); }); @@ -675,8 +675,9 @@ Result MergingSnapshotUpdate::CopyManifest(const ManifestFile& man ICEBERG_ASSIGN_OR_RAISE(auto spec, current.PartitionSpecById(manifest.partition_spec_id)); std::string path = ManifestPath(); - return CopyAppendManifest(manifest, ctx_->table->io(), schema, spec, SnapshotId(), path, - current.format_version, &appended_manifests_summary_); + return CopyAppendManifest(manifest, ctx_->table->manifest_io(), schema, spec, + SnapshotId(), path, current.format_version, + &appended_manifests_summary_); } // ------------------------------------------------------------------------- diff --git a/src/iceberg/update/overwrite_files.cc b/src/iceberg/update/overwrite_files.cc index cba6e028c..90b90a144 100644 --- a/src/iceberg/update/overwrite_files.cc +++ b/src/iceberg/update/overwrite_files.cc @@ -215,25 +215,25 @@ Status OverwriteFiles::Validate(const TableMetadata& current_metadata, if (validate_new_data_files_) { ICEBERG_RETURN_UNEXPECTED(ValidateAddedDataFiles( current_metadata, starting_snapshot_id_, DataConflictDetectionFilter(), snapshot, - ctx_->table->io(), IsCaseSensitive())); + ctx_->table->manifest_io(), IsCaseSensitive())); } if (validate_new_deletes_) { if (row_filter != nullptr && row_filter != Expressions::AlwaysFalse()) { auto filter = conflict_detection_filter_ != nullptr ? conflict_detection_filter_ : row_filter; - ICEBERG_RETURN_UNEXPECTED( - ValidateNoNewDeleteFiles(current_metadata, starting_snapshot_id_, filter, - snapshot, ctx_->table->io(), IsCaseSensitive())); - ICEBERG_RETURN_UNEXPECTED( - ValidateDeletedDataFiles(current_metadata, starting_snapshot_id_, filter, - snapshot, ctx_->table->io(), IsCaseSensitive())); + ICEBERG_RETURN_UNEXPECTED(ValidateNoNewDeleteFiles( + current_metadata, starting_snapshot_id_, filter, snapshot, + ctx_->table->manifest_io(), IsCaseSensitive())); + ICEBERG_RETURN_UNEXPECTED(ValidateDeletedDataFiles( + current_metadata, starting_snapshot_id_, filter, snapshot, + ctx_->table->manifest_io(), IsCaseSensitive())); } if (!deleted_data_files_.empty()) { ICEBERG_RETURN_UNEXPECTED(ValidateNoNewDeletesForDataFiles( current_metadata, starting_snapshot_id_, conflict_detection_filter_, - deleted_data_files_, snapshot, ctx_->table->io(), IsCaseSensitive())); + deleted_data_files_, snapshot, ctx_->table->manifest_io(), IsCaseSensitive())); } } diff --git a/src/iceberg/update/replace_partitions.cc b/src/iceberg/update/replace_partitions.cc index 4487751bf..8ad35cc4b 100644 --- a/src/iceberg/update/replace_partitions.cc +++ b/src/iceberg/update/replace_partitions.cc @@ -113,7 +113,7 @@ Status ReplacePartitions::Validate(const TableMetadata& current_metadata, return {}; } - auto io = ctx_->table->io(); + auto io = ctx_->table->manifest_io(); if (validate_conflicting_data_) { if (data_spec->IsUnpartitioned()) { ICEBERG_RETURN_UNEXPECTED(ValidateAddedDataFiles( diff --git a/src/iceberg/update/rewrite_files.cc b/src/iceberg/update/rewrite_files.cc index 3c743c01d..9df62f553 100644 --- a/src/iceberg/update/rewrite_files.cc +++ b/src/iceberg/update/rewrite_files.cc @@ -172,7 +172,7 @@ Status RewriteFiles::Validate(const TableMetadata& current_metadata, if (!replaced_data_files_.empty()) { // If there are replaced data files, there cannot be any new row-level deletes // for those data files. - auto io = ctx_->table->io(); + auto io = ctx_->table->manifest_io(); ICEBERG_RETURN_UNEXPECTED(MergingSnapshotUpdate::ValidateNoNewDeletesForDataFiles( current_metadata, starting_snapshot_id_, replaced_data_files_, snapshot, std::move(io), HasDataSequenceNumber())); diff --git a/src/iceberg/update/row_delta.cc b/src/iceberg/update/row_delta.cc index c58b5233d..deac1a2dc 100644 --- a/src/iceberg/update/row_delta.cc +++ b/src/iceberg/update/row_delta.cc @@ -139,7 +139,7 @@ Status RowDelta::Validate(const TableMetadata& current_metadata, starting_snapshot_id_.value(), snapshot->snapshot_id); } - auto io = ctx_->table->io(); + auto io = ctx_->table->manifest_io(); if (!referenced_data_files_.empty()) { ICEBERG_RETURN_UNEXPECTED(MergingSnapshotUpdate::ValidateDataFilesExist( current_metadata, starting_snapshot_id_, referenced_data_files_, diff --git a/src/iceberg/update/snapshot_update.cc b/src/iceberg/update/snapshot_update.cc index d6a1066ac..bdfc17cad 100644 --- a/src/iceberg/update/snapshot_update.cc +++ b/src/iceberg/update/snapshot_update.cc @@ -328,7 +328,8 @@ Result SnapshotUpdate::Apply() { continue; } metadata_tasks.Submit([&manifest, this]() -> Status { - ICEBERG_ASSIGN_OR_RAISE(manifest, AddMetadata(manifest, ctx_->table->io(), base())); + ICEBERG_ASSIGN_OR_RAISE(manifest, + AddMetadata(manifest, ctx_->table->manifest_io(), base())); return {}; }); } @@ -392,7 +393,8 @@ Status SnapshotUpdate::Finalize([[maybe_unused]] const TableMetadata& metadata) // Only files created by this update are tracked, and committed paths are kept below. internal::LogAndIgnoreFailure("Snapshot cleanup", [this]() -> Status { auto snapshot_reader = SnapshotReader(staged_snapshot_.get()); - ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_reader.Manifests(ctx_->table->io())); + ICEBERG_ASSIGN_OR_RAISE(auto manifests, + snapshot_reader.Manifests(ctx_->table->manifest_io())); auto committed = manifests | std::views::transform([](const auto& manifest) { return manifest.manifest_path; }) | diff --git a/src/iceberg/util/cache_internal.h b/src/iceberg/util/cache_internal.h index d82a93b80..4aded5a61 100644 --- a/src/iceberg/util/cache_internal.h +++ b/src/iceberg/util/cache_internal.h @@ -25,6 +25,7 @@ /// \file iceberg/util/cache_internal.h /// \brief Provide internal LRU cache and memoization helpers. +#include #include #include #include @@ -37,13 +38,31 @@ namespace iceberg::internal { +// Gives every entry a weight of one, so that the maximum weight of a cache is the +// maximum number of entries. +struct UnitWeigher { + template + constexpr int64_t operator()(const Key& /*key*/, + const Value& /*value*/) const noexcept { + return 1; + } +}; + // A LRU (Least recently used) replacement cache -template +// +// The combined weight of the cached entries never exceeds `max_weight`; least recently +// used entries are evicted first to make room. `Weigher` computes the weight of an entry +// from its key and value when the entry is inserted or replaced, e.g. its size in bytes. +// Weights must be non-negative. With the default UnitWeigher, `max_weight` is the +// maximum number of entries. +template , typename KeyEqual = std::equal_to> class LruCache { public: - explicit LruCache(int32_t capacity) : capacity_(capacity) { - // The map size can temporarily exceed the cache capacity, see Replace() - map_.reserve(capacity_ + 1); + explicit LruCache(int64_t max_weight, Weigher weigher = Weigher{}) + : max_weight_(max_weight), weigher_(std::move(weigher)) { + ICEBERG_DCHECK(max_weight_ >= 0, "Cache maximum weight must be non-negative"); + ReserveForUnitWeight(); } LruCache(const LruCache&) = delete; @@ -54,15 +73,23 @@ class LruCache { void Clear() { items_.clear(); map_.clear(); + total_weight_ = 0; // The C++ spec doesn't tell whether map_.clear() will shrink the map capacity - map_.reserve(capacity_ + 1); + ReserveForUnitWeight(); } + // Number of retained entries. int32_t size() const { ICEBERG_DCHECK(items_.size() == map_.size(), "Cache item and map sizes differ"); return static_cast(items_.size()); } + // Combined weight of retained entries. + int64_t total_weight() const { return total_weight_; } + + // Maximum combined weight allowed in this cache. + int64_t max_weight() const { return max_weight_; } + template Value* Find(K&& key) { const auto it = map_.find(key); @@ -76,29 +103,47 @@ class LruCache { return &list_it->value; } + // Insert or update the value for `key`, mark it as most recently used, and evict least + // recently used entries until the cache is within its maximum weight again. + // + // Returns whether a new entry was added, and a pointer to the cached value. A value + // heavier than the maximum weight is not cached: any existing entry for `key` is + // removed, other entries are kept, and {false, nullptr} is returned. template std::pair Replace(K&& key, V&& value) { // Try to insert temporary iterator auto [it, inserted] = map_.emplace(std::forward(key), ListIt{}); + ListIt list_it; if (inserted) { // Inserted => push item at front of the list, and update iterator items_.emplace_front(&it->first, std::forward(value)); - it->second = items_.begin(); - // Did we exceed the cache capacity? If so, remove least recently used item - if (static_cast(items_.size()) > capacity_) { - const bool erased = map_.erase(*items_.back().key); - ICEBERG_DCHECK(erased, "Failed to erase least recently used cache item"); - static_cast(erased); - items_.pop_back(); - } - return {true, &it->second->value}; + list_it = items_.begin(); + it->second = list_it; + } else { + // Already exists => move item at front of the list, and update value + list_it = it->second; + items_.splice(items_.begin(), items_, list_it); + list_it->value = std::forward(value); + total_weight_ -= list_it->weight; + list_it->weight = 0; } - // Already exists => move item at front of the list, and update value - auto list_it = it->second; - items_.splice(items_.begin(), items_, list_it); - list_it->value = std::forward(value); - return {false, &list_it->value}; + const int64_t weight = weigher_(it->first, list_it->value); + ICEBERG_DCHECK(weight >= 0, "Cache entry weight must be non-negative"); + if (weight > max_weight_) { + items_.erase(list_it); + map_.erase(it); + return {false, nullptr}; + } + + // The item is at the front of the list and not yet counted in total_weight_, and + // `max_weight_ - weight` is non-negative, so eviction stops before reaching it. + while (total_weight_ > max_weight_ - weight) { + EvictLeastRecentlyUsed(); + } + list_it->weight = weight; + total_weight_ += weight; + return {inserted, &list_it->value}; } private: @@ -106,14 +151,35 @@ class LruCache { // Pointer to the key inside the unordered_map const Key* key; Value value; + int64_t weight = 0; }; using List = std::list; using ListIt = typename List::iterator; - const int32_t capacity_; + void EvictLeastRecentlyUsed() { + ICEBERG_DCHECK(!items_.empty(), "Cannot evict from an empty cache"); + const Item& item = items_.back(); + total_weight_ -= item.weight; + const bool erased = map_.erase(*item.key); + ICEBERG_DCHECK(erased, "Failed to erase least recently used cache item"); + static_cast(erased); + items_.pop_back(); + } + + void ReserveForUnitWeight() { + // Only with unit weights does the maximum weight bound the number of entries. The + // map size can temporarily exceed it by one, see Replace(). + if constexpr (std::is_same_v) { + map_.reserve(static_cast(max_weight_) + 1); + } + } + + const int64_t max_weight_; + Weigher weigher_; + int64_t total_weight_ = 0; // In most to least recently used order std::list items_; - std::unordered_map map_; + std::unordered_map map_; }; template From 339798ba6d3ab44dd7f6a952b518117a17939690 Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Fri, 6 Nov 2026 23:25:05 +0800 Subject: [PATCH 2/3] docs: remove manifest cache changes from FileIO guide --- mkdocs/docs/file-io.md | 28 ---------------------------- 1 file changed, 28 deletions(-) diff --git a/mkdocs/docs/file-io.md b/mkdocs/docs/file-io.md index 823035c24..62f998b1f 100644 --- a/mkdocs/docs/file-io.md +++ b/mkdocs/docs/file-io.md @@ -21,34 +21,6 @@ `FileIO` reads, writes, and deletes Iceberg data and metadata files. -## Manifest content cache - -Manifest content caching is owned by each loaded `Table`. Tables sharing a `FileIO` -have independent caches and byte budgets. The storage `FileIO` remains usable for -uncached data-file and table-metadata JSON reads. - -| Property | Default | Description | -|---|---|---| -| `io.manifest.cache-enabled` | `false` | Cache immutable manifest-list and manifest bytes across scans of a table | -| `io.manifest.cache.max-total-bytes` | `104857600` | Maximum combined content retained by one table's cache | -| `io.manifest.cache.max-content-length` | `8388608` | Maximum file size eligible for caching | - -Catalog I/O properties provide defaults. Table properties override them. Direct -`Table::Make` calls can supply defaults through the trailing `io_properties` -argument; properties exposed by the supplied FileIO are also used as defaults. - -Scans and manifest maintenance share the table's cache. Refreshes and transactions -preserve cached bytes when the table UUID, cache limits, and storage context are -compatible. Independently loaded table handles start with separate caches. - -`Table::ClearManifestCache()` releases retained entries. Active input handles and -streams can keep their bytes alive after clearing or eviction; those bytes and -concurrent population buffers are outside the retention budget. Normalized path -aliases use a separate index retaining at most 64 KiB of alias strings per table -and do not pin content. Empty files and -files above either byte limit bypass the cache. Time-based expiration is not -supported, so `io.manifest.cache.expiration-interval-ms` is ignored. - ## Built-in implementations Call `iceberg::arrow::RegisterAll()` to register the Arrow-backed FileIO From 77677717fc313ed1abead9dda6401259436f5c42 Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Fri, 6 Nov 2026 23:50:38 +0800 Subject: [PATCH 3/3] refactor: align content cache source and header names --- src/iceberg/CMakeLists.txt | 2 +- src/iceberg/{content_cache.cc => content_cache_internal.cc} | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) rename src/iceberg/{content_cache.cc => content_cache_internal.cc} (99%) diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index b72333b4f..782cd3d7e 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -26,7 +26,7 @@ set(ICEBERG_SOURCES catalog/memory/in_memory_catalog.cc catalog/session_catalog.cc catalog/session_context.cc - content_cache.cc + content_cache_internal.cc data_table_scan.cc delete_file_index.cc deletes/dv_util.cc diff --git a/src/iceberg/content_cache.cc b/src/iceberg/content_cache_internal.cc similarity index 99% rename from src/iceberg/content_cache.cc rename to src/iceberg/content_cache_internal.cc index bb4f52e1f..05f417c1a 100644 --- a/src/iceberg/content_cache.cc +++ b/src/iceberg/content_cache_internal.cc @@ -17,12 +17,13 @@ * under the License. */ +#include "iceberg/content_cache_internal.h" + #include #include #include #include -#include "iceberg/content_cache_internal.h" #include "iceberg/util/macros.h" #include "iceberg/util/string_util.h"