From 3a165cd01d7dee7cc5c6cb08e87c79bf34456e94 Mon Sep 17 00:00:00 2001 From: manuzhang Date: Thu, 6 Aug 2026 16:13:21 +0800 Subject: [PATCH 1/5] feat: add metadata and snapshot caching Co-authored-by: Codex --- src/iceberg/CMakeLists.txt | 1 + src/iceberg/arrow/arrow_io.cc | 26 +- src/iceberg/arrow/arrow_io_internal.h | 3 +- src/iceberg/avro/avro_reader.cc | 3 +- .../catalog/memory/in_memory_catalog.cc | 5 + src/iceberg/catalog/sql/sql_catalog.cc | 4 + src/iceberg/file_io.cc | 217 +++++++++- src/iceberg/file_io.h | 33 ++ src/iceberg/file_io_registry.cc | 7 +- src/iceberg/file_reader.h | 2 + src/iceberg/manifest/manifest_reader.cc | 2 + src/iceberg/meson.build | 2 + src/iceberg/metadata_cache.cc | 341 ++++++++++++++++ src/iceberg/metadata_cache.h | 109 +++++ src/iceberg/snapshot.cc | 94 ++++- src/iceberg/snapshot.h | 25 +- src/iceberg/table_metadata.cc | 2 +- src/iceberg/test/CMakeLists.txt | 1 + src/iceberg/test/meson.build | 1 + src/iceberg/test/metadata_cache_test.cc | 371 ++++++++++++++++++ src/iceberg/update/expire_snapshots.cc | 3 +- 21 files changed, 1207 insertions(+), 45 deletions(-) create mode 100644 src/iceberg/metadata_cache.cc create mode 100644 src/iceberg/metadata_cache.h create mode 100644 src/iceberg/test/metadata_cache_test.cc diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index e48209d62..79cceb152 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -68,6 +68,7 @@ set(ICEBERG_SOURCES manifest/v1_metadata.cc manifest/v2_metadata.cc manifest/v3_metadata.cc + metadata_cache.cc metadata_columns.cc metrics_config.cc metrics/commit_report.cc diff --git a/src/iceberg/arrow/arrow_io.cc b/src/iceberg/arrow/arrow_io.cc index 4c795badf..8c92fa57e 100644 --- a/src/iceberg/arrow/arrow_io.cc +++ b/src/iceberg/arrow/arrow_io.cc @@ -508,24 +508,28 @@ Result ArrowFileSystemFileIO::ResolvePath(const std::string& file_l Result> OpenArrowInputStream( const std::shared_ptr& io, const std::string& path, - std::optional length) { + std::optional length, bool cache_content) { ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null"); - if (auto arrow_io = std::dynamic_pointer_cast(io)) { - ICEBERG_ASSIGN_OR_RAISE(auto resolved_path, arrow_io->ResolvePath(path)); - ::arrow::fs::FileInfo file_info(resolved_path, ::arrow::fs::FileType::File); - if (length.has_value()) { - ICEBERG_ASSIGN_OR_RAISE(auto size, ToInt64Length(*length)); - file_info.set_size(size); + if (!cache_content || !io->MetadataCacheEnabled()) { + if (auto arrow_io = std::dynamic_pointer_cast(io)) { + ICEBERG_ASSIGN_OR_RAISE(auto resolved_path, arrow_io->ResolvePath(path)); + ::arrow::fs::FileInfo file_info(resolved_path, ::arrow::fs::FileType::File); + if (length.has_value()) { + ICEBERG_ASSIGN_OR_RAISE(auto size, ToInt64Length(*length)); + file_info.set_size(size); + } + ICEBERG_ARROW_ASSIGN_OR_RETURN(auto input, + arrow_io->arrow_fs_->OpenInputFile(file_info)); + return input; } - ICEBERG_ARROW_ASSIGN_OR_RETURN(auto input, - arrow_io->arrow_fs_->OpenInputFile(file_info)); - return input; } int64_t size; std::unique_ptr input_file; - if (length.has_value()) { + if (cache_content) { + ICEBERG_ASSIGN_OR_RAISE(input_file, io->NewCachedInputFile(path, length)); + } else if (length.has_value()) { ICEBERG_ASSIGN_OR_RAISE(input_file, io->NewInputFile(path, *length)); } else { ICEBERG_ASSIGN_OR_RAISE(input_file, io->NewInputFile(path)); diff --git a/src/iceberg/arrow/arrow_io_internal.h b/src/iceberg/arrow/arrow_io_internal.h index a6b85b6c9..8f46130e6 100644 --- a/src/iceberg/arrow/arrow_io_internal.h +++ b/src/iceberg/arrow/arrow_io_internal.h @@ -40,7 +40,8 @@ namespace iceberg::arrow { /// implement NewInputFile. ICEBERG_BUNDLE_EXPORT Result> OpenArrowInputStream(const std::shared_ptr& io, const std::string& path, - std::optional length = std::nullopt); + std::optional length = std::nullopt, + bool cache_content = false); /// \brief Open a FileIO output as an Arrow output stream. /// diff --git a/src/iceberg/avro/avro_reader.cc b/src/iceberg/avro/avro_reader.cc index f333a38cb..a71023c9e 100644 --- a/src/iceberg/avro/avro_reader.cc +++ b/src/iceberg/avro/avro_reader.cc @@ -54,7 +54,8 @@ namespace { Result> CreateInputStream(const ReaderOptions& options, int64_t buffer_size) { ICEBERG_ASSIGN_OR_RAISE( - auto file, arrow::OpenArrowInputStream(options.io, options.path, options.length)); + auto file, arrow::OpenArrowInputStream(options.io, options.path, options.length, + options.cache_content)); return std::make_unique(file, buffer_size); } diff --git a/src/iceberg/catalog/memory/in_memory_catalog.cc b/src/iceberg/catalog/memory/in_memory_catalog.cc index d0797e728..bc28a1677 100644 --- a/src/iceberg/catalog/memory/in_memory_catalog.cc +++ b/src/iceberg/catalog/memory/in_memory_catalog.cc @@ -24,6 +24,7 @@ #include "iceberg/catalog/catalog_util.h" #include "iceberg/file_io.h" +#include "iceberg/metadata_cache.h" #include "iceberg/metrics/metrics_reporters.h" #include "iceberg/table.h" #include "iceberg/table_identifier.h" @@ -343,6 +344,10 @@ 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"); + if (properties.contains(std::string(MetadataCacheOptions::kEnabled))) { + ICEBERG_RETURN_UNEXPECTED(file_io->ConfigureMetadataCache(properties)); + } std::shared_ptr reporter; auto it = properties.find(std::string(kMetricsReporterImpl)); if (it != properties.end() && !it->second.empty() && diff --git a/src/iceberg/catalog/sql/sql_catalog.cc b/src/iceberg/catalog/sql/sql_catalog.cc index e69729327..21fef31dc 100644 --- a/src/iceberg/catalog/sql/sql_catalog.cc +++ b/src/iceberg/catalog/sql/sql_catalog.cc @@ -26,6 +26,7 @@ #include "iceberg/catalog/catalog_util.h" #include "iceberg/catalog/sql/config.h" #include "iceberg/file_io.h" +#include "iceberg/metadata_cache.h" #include "iceberg/metrics/metrics_reporters.h" #include "iceberg/table.h" #include "iceberg/table_identifier.h" @@ -148,6 +149,9 @@ Result> SqlCatalog::Make( if (file_io == nullptr) { return InvalidArgument("SqlCatalog requires a non-null FileIO"); } + if (config.props.contains(std::string(MetadataCacheOptions::kEnabled))) { + ICEBERG_RETURN_UNEXPECTED(file_io->ConfigureMetadataCache(config.props)); + } ICEBERG_RETURN_UNEXPECTED(store->Initialize()); std::shared_ptr reporter; diff --git a/src/iceberg/file_io.cc b/src/iceberg/file_io.cc index e4223182e..345549454 100644 --- a/src/iceberg/file_io.cc +++ b/src/iceberg/file_io.cc @@ -19,9 +19,12 @@ #include "iceberg/file_io.h" +#include +#include #include #include +#include "iceberg/metadata_cache.h" #include "iceberg/util/macros.h" namespace iceberg { @@ -40,6 +43,123 @@ Status FinishWithCloseStatus(Status operation_status, Status close_status) { 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::string content(size, '\0'); + ICEBERG_ASSIGN_OR_RAISE(auto stream, input_file.Open()); + Status read_status = {}; + if (size > 0) { + auto bytes = std::as_writable_bytes(std::span(content.data(), content.size())); + read_status = stream->ReadFully(/*position=*/0, bytes); + } + ICEBERG_RETURN_UNEXPECTED( + FinishWithCloseStatus(std::move(read_status), stream->Close())); + return content; +} + +class CachedSeekableInputStream : public SeekableInputStream { + public: + explicit CachedSeekableInputStream(std::shared_ptr 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: + std::shared_ptr content_; + int64_t position_ = 0; + bool closed_ = false; +}; + +class CachedInputFile : public InputFile { + public: + CachedInputFile(std::unique_ptr input_file, + std::shared_ptr cache, int64_t size) + : input_file_(std::move(input_file)), + cache_(std::move(cache)), + location_(input_file_->location()), + size_(size) {} + + std::string_view location() const override { return location_; } + + Result Size() const override { + if (auto content = cache_->GetIfPresent(location_)) { + return static_cast(content->size()); + } + return size_; + } + + Result> Open() override { + auto content = + cache_->Get(location_, static_cast(size_), + [this]() -> Result { + ICEBERG_ASSIGN_OR_RAISE( + auto loaded, ReadInputFile(*input_file_, size_, location_)); + return std::make_shared(std::move(loaded)); + }); + if (!content.has_value()) { + // Cache loading is an optimization. Match Java's ContentCache by falling back to + // the underlying input when a read-ahead attempt fails. + return input_file_->Open(); + } + return std::make_unique(std::move(content).value()); + } + + private: + std::unique_ptr input_file_; + std::shared_ptr cache_; + std::string location_; + int64_t size_; +}; + } // namespace Result> FileIO::NewInputFile(std::string file_location) { @@ -69,21 +189,94 @@ Result FileIO::ReadFile(const std::string& file_location, ICEBERG_ASSIGN_OR_RAISE(input_file, NewInputFile(file_location)); ICEBERG_ASSIGN_OR_RAISE(read_size, input_file->Size()); } - if (read_size < 0) { - return Invalid("Invalid negative file size {} for {}", read_size, file_location); + return ReadInputFile(*input_file, read_size, file_location); +} + +Result> FileIO::NewCachedInputFile( + std::string file_location, std::optional length) { + std::unique_ptr input_file; + if (length.has_value()) { + ICEBERG_ASSIGN_OR_RAISE(input_file, NewInputFile(file_location, *length)); + } else { + ICEBERG_ASSIGN_OR_RAISE(input_file, NewInputFile(file_location)); } - auto size = static_cast(read_size); - std::string content(size, '\0'); - ICEBERG_ASSIGN_OR_RAISE(auto stream, input_file->Open()); - Status read_status = {}; - if (size > 0) { - auto bytes = std::as_writable_bytes(std::span(content.data(), content.size())); - read_status = stream->ReadFully(/*position=*/0, bytes); + auto cache = metadata_cache_.load(std::memory_order_acquire); + if (cache == nullptr || !cache->options().enabled) { + return input_file; + } + + if (auto cached = cache->GetIfPresent(file_location)) { + return std::make_unique(std::move(input_file), std::move(cache), + static_cast(cached->size())); + } + + int64_t size; + if (length.has_value()) { + if (*length > static_cast(std::numeric_limits::max())) { + return InvalidArgument("File length {} exceeds int64_t max", *length); + } + size = static_cast(*length); + } else { + ICEBERG_ASSIGN_OR_RAISE(size, input_file->Size()); + } + if (size < 0) { + return Invalid("Invalid negative file size {} for {}", size, file_location); + } + if (static_cast(size) > cache->options().max_content_length) { + return input_file; + } + return std::make_unique(std::move(input_file), std::move(cache), size); +} + +Result FileIO::ReadFileCached(const std::string& file_location, + std::optional length) { + auto cache = metadata_cache_.load(std::memory_order_acquire); + if (cache == nullptr || !cache->options().enabled) { + return ReadFile(file_location, length); + } + ICEBERG_ASSIGN_OR_RAISE( + auto content, + cache->Get(file_location, length, + [this, &file_location, length]() -> Result { + ICEBERG_ASSIGN_OR_RAISE(auto loaded, ReadFile(file_location, length)); + return std::make_shared(std::move(loaded)); + })); + return *content; +} + +Status FileIO::ConfigureMetadataCache( + const std::unordered_map& properties) { + ICEBERG_ASSIGN_OR_RAISE(auto options, MetadataCacheOptions::FromProperties(properties)); + ICEBERG_ASSIGN_OR_RAISE(auto cache, MetadataCache::Make(options)); + std::shared_ptr expected; + while (expected == nullptr) { + if (metadata_cache_.compare_exchange_strong( + expected, cache, std::memory_order_release, std::memory_order_acquire)) { + return {}; + } + } + if (expected->options() == options) { + return {}; + } + return InvalidArgument("Metadata cache is already configured with different options"); +} + +bool FileIO::MetadataCacheEnabled() const noexcept { + auto cache = metadata_cache_.load(std::memory_order_acquire); + return cache != nullptr && cache->options().enabled; +} + +void FileIO::InvalidateMetadataCache(std::string_view file_location) { + if (auto cache = metadata_cache_.load(std::memory_order_acquire)) { + cache->Invalidate(file_location); + } +} + +void FileIO::ClearMetadataCache() { + if (auto cache = metadata_cache_.load(std::memory_order_acquire)) { + cache->Clear(); } - ICEBERG_RETURN_UNEXPECTED( - FinishWithCloseStatus(std::move(read_status), stream->Close())); - return content; } Status FileIO::WriteFile(const std::string& file_location, std::string_view content) { diff --git a/src/iceberg/file_io.h b/src/iceberg/file_io.h index 3ea4afa49..bed694360 100644 --- a/src/iceberg/file_io.h +++ b/src/iceberg/file_io.h @@ -22,6 +22,7 @@ /// \file iceberg/file_io.h /// \brief Define the FileIO abstraction and file stream interfaces. +#include #include #include #include @@ -29,6 +30,7 @@ #include #include #include +#include #include #include "iceberg/iceberg_export.h" @@ -38,6 +40,7 @@ namespace iceberg { class SupportsStorageCredentials; +class MetadataCache; /// \brief Seekable byte stream for reading file contents. class ICEBERG_EXPORT SeekableInputStream { @@ -153,6 +156,33 @@ class ICEBERG_EXPORT FileIO { virtual Result ReadFile(const std::string& file_location, std::optional length); + /// \brief Create an input handle backed by the metadata content cache when enabled. + /// + /// This is intended for immutable Iceberg metadata files, including metadata JSON, + /// manifest lists, and manifests. Data files should use NewInputFile directly. + Result> NewCachedInputFile( + std::string file_location, std::optional length = std::nullopt); + + /// \brief Read an immutable Iceberg metadata file through the content cache. + Result ReadFileCached(const std::string& file_location, + std::optional length); + + /// \brief Configure metadata caching from catalog/FileIO properties. + /// + /// Configuration is immutable after the first call. Repeating the same configuration + /// is allowed; attempting to replace it returns an error. + Status ConfigureMetadataCache( + const std::unordered_map& properties); + + /// \brief Return whether metadata content caching is enabled. + bool MetadataCacheEnabled() const noexcept; + + /// \brief Invalidate a cached metadata file location. + void InvalidateMetadataCache(std::string_view file_location); + + /// \brief Clear cached metadata file content. + void ClearMetadataCache(); + /// \brief Write the given content to the file at the given location. /// /// \param file_location The location of the file to write. @@ -180,6 +210,9 @@ class ICEBERG_EXPORT FileIO { /// \brief Return storage-credential support when implemented by this FileIO. virtual SupportsStorageCredentials* AsSupportsStorageCredentials() { return nullptr; } + + private: + std::atomic> metadata_cache_; }; /// \brief Mix-in for FileIO implementations that route object paths to diff --git a/src/iceberg/file_io_registry.cc b/src/iceberg/file_io_registry.cc index 77ff4a9d7..dc88a6197 100644 --- a/src/iceberg/file_io_registry.cc +++ b/src/iceberg/file_io_registry.cc @@ -22,6 +22,8 @@ #include #include +#include "iceberg/util/macros.h" + namespace iceberg { namespace { @@ -57,7 +59,10 @@ Result> FileIORegistry::Load( } factory = it->second; } - return factory(properties); + ICEBERG_ASSIGN_OR_RAISE(auto io, factory(properties)); + ICEBERG_PRECHECK(io != nullptr, "FileIO factory returned null for {}", name); + ICEBERG_RETURN_UNEXPECTED(io->ConfigureMetadataCache(properties)); + return io; } } // namespace iceberg diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h index cefc688c0..56685ef9b 100644 --- a/src/iceberg/file_reader.h +++ b/src/iceberg/file_reader.h @@ -98,6 +98,8 @@ struct ICEBERG_EXPORT ReaderOptions { std::optional split; /// \brief FileIO instance to open the file. std::shared_ptr io; + /// \brief Cache this immutable metadata file's content when FileIO caching is enabled. + bool cache_content = false; /// \brief The projection schema to read from the file. This field is required. std::shared_ptr projection; /// \brief The filter to apply to the data. Reader implementations may ignore this if diff --git a/src/iceberg/manifest/manifest_reader.cc b/src/iceberg/manifest/manifest_reader.cc index 68760b440..638414f13 100644 --- a/src/iceberg/manifest/manifest_reader.cc +++ b/src/iceberg/manifest/manifest_reader.cc @@ -849,6 +849,7 @@ Status ManifestReaderImpl::OpenReader(std::shared_ptr projection) { ReaderOptions options; options.path = manifest_path_; options.io = file_io_; + options.cache_content = true; options.projection = file_schema_; if (manifest_length_.has_value()) { options.length = manifest_length_; @@ -1058,6 +1059,7 @@ Result> ManifestListReader::Make( { .path = std::string(manifest_list_location), .io = std::move(file_io), + .cache_content = true, .projection = schema, })); return std::make_unique(std::move(reader), std::move(schema)); diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build index 39e8b2939..930a4eaa2 100644 --- a/src/iceberg/meson.build +++ b/src/iceberg/meson.build @@ -116,6 +116,7 @@ iceberg_sources = files( 'manifest/v1_metadata.cc', 'manifest/v2_metadata.cc', 'manifest/v3_metadata.cc', + 'metadata_cache.cc', 'metadata_columns.cc', 'metrics/commit_report.cc', 'metrics/counter.cc', @@ -320,6 +321,7 @@ install_headers( 'iceberg_export.h', 'inheritable_metadata.h', 'location_provider.h', + 'metadata_cache.h', 'metadata_columns.h', 'metrics.h', 'metrics_config.h', diff --git a/src/iceberg/metadata_cache.cc b/src/iceberg/metadata_cache.cc new file mode 100644 index 000000000..661090f20 --- /dev/null +++ b/src/iceberg/metadata_cache.cc @@ -0,0 +1,341 @@ +/* + * 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 "iceberg/metadata_cache.h" + +#include +#include +#include +#include +#include +#include +#include + +#include "iceberg/util/macros.h" +#include "iceberg/util/string_util.h" + +namespace iceberg { + +namespace { + +Result ParseBoolean(std::string_view key, const std::string& value) { + if (value == "true") { + return true; + } + if (value == "false") { + return false; + } + return InvalidArgument("Invalid boolean value '{}' for property {}", value, key); +} + +template +Result ParseNumberProperty( + const std::unordered_map& properties, std::string_view key, + T default_value) { + auto it = properties.find(std::string(key)); + if (it == properties.end()) { + return default_value; + } + return StringUtils::ParseNumber(it->second); +} + +} // namespace + +Result MetadataCacheOptions::FromProperties( + const std::unordered_map& properties) { + MetadataCacheOptions options; + if (auto it = properties.find(std::string(kEnabled)); it != properties.end()) { + ICEBERG_ASSIGN_OR_RAISE(options.enabled, ParseBoolean(kEnabled, it->second)); + } + + if (!options.enabled) { + return options; + } + + ICEBERG_ASSIGN_OR_RAISE(options.expiration_interval_ms, + ParseNumberProperty(properties, kExpirationIntervalMs, + options.expiration_interval_ms)); + 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)); + return options; +} + +class MetadataCache::Impl { + public: + using Clock = std::chrono::steady_clock; + + explicit Impl(MetadataCacheOptions options) : options_(std::move(options)) {} + + struct Entry { + bool loading = true; + Content content; + std::optional error; + Clock::time_point last_access; + std::list::iterator lru_position; + std::condition_variable loaded; + }; + + bool IsExpired(const Entry& entry, Clock::time_point now) const { + return options_.expiration_interval_ms > 0 && + now - entry.last_access >= + std::chrono::milliseconds(options_.expiration_interval_ms); + } + + void Touch(const std::string& location, Entry& entry, Clock::time_point now) { + entry.last_access = now; + lru_.splice(lru_.begin(), lru_, entry.lru_position); + entry.lru_position = lru_.begin(); + } + + void EraseLoaded(std::unordered_map>::iterator it) { + auto& entry = *it->second; + total_bytes_ -= entry.content->size(); + lru_.erase(entry.lru_position); + entries_.erase(it); + } + + void PruneExpired(Clock::time_point now) { + while (!lru_.empty()) { + auto it = entries_.find(lru_.back()); + if (it == entries_.end()) { + lru_.pop_back(); + continue; + } + if (!IsExpired(*it->second, now)) { + break; + } + EraseLoaded(it); + } + } + + void EvictToFit(size_t incoming_bytes) { + while (!lru_.empty() && incoming_bytes > options_.max_total_bytes - total_bytes_) { + auto it = entries_.find(lru_.back()); + if (it == entries_.end()) { + lru_.pop_back(); + } else { + EraseLoaded(it); + } + } + } + + MetadataCacheOptions options_; + mutable std::mutex mutex_; + std::unordered_map> entries_; + std::list lru_; + size_t total_bytes_ = 0; + bool clearing_ = false; + std::condition_variable clear_completed_; +}; + +Result> MetadataCache::Make(MetadataCacheOptions options) { + if (options.enabled) { + ICEBERG_PRECHECK(options.expiration_interval_ms >= 0, + "Metadata cache expiration interval must not be negative: {}", + options.expiration_interval_ms); + ICEBERG_PRECHECK(options.max_total_bytes > 0, + "Metadata cache maximum total bytes must be positive"); + ICEBERG_PRECHECK(options.max_content_length > 0, + "Metadata cache maximum content length must be positive"); + ICEBERG_PRECHECK(options.max_content_length <= + static_cast(std::numeric_limits::max()), + "Metadata cache maximum content length exceeds int64_t max: {}", + options.max_content_length); + } + return std::shared_ptr(new MetadataCache(std::move(options))); +} + +MetadataCache::MetadataCache(MetadataCacheOptions options) + : impl_(std::make_unique(std::move(options))) {} + +MetadataCache::~MetadataCache() = default; + +const MetadataCacheOptions& MetadataCache::options() const noexcept { + return impl_->options_; +} + +Result MetadataCache::Get(std::string location, + std::optional length, + Loader loader) { + if (!impl_->options_.enabled || + (length.has_value() && *length > impl_->options_.max_content_length)) { + auto loaded = loader(); + if (loaded.has_value() && loaded.value() == nullptr) { + return Invalid("Metadata cache loader returned null content for {}", location); + } + return loaded; + } + + std::shared_ptr entry; + { + std::unique_lock lock(impl_->mutex_); + impl_->clear_completed_.wait(lock, [this] { return !impl_->clearing_; }); + auto now = Impl::Clock::now(); + impl_->PruneExpired(now); + + while (true) { + auto it = impl_->entries_.find(location); + if (it == impl_->entries_.end()) { + entry = std::make_shared(); + impl_->entries_.emplace(location, entry); + break; + } + + entry = it->second; + if (entry->loading) { + entry->loaded.wait(lock, [&entry] { return !entry->loading; }); + if (entry->error.has_value()) { + return std::unexpected(*entry->error); + } + if (entry->content != nullptr) { + if (auto current = impl_->entries_.find(location); + current != impl_->entries_.end() && current->second == entry) { + impl_->Touch(location, *entry, Impl::Clock::now()); + } + return entry->content; + } + continue; + } + + now = Impl::Clock::now(); + if (impl_->IsExpired(*entry, now)) { + impl_->EraseLoaded(it); + continue; + } + impl_->Touch(location, *entry, now); + return entry->content; + } + } + + auto loaded = loader(); + if (loaded.has_value() && loaded.value() == nullptr) { + loaded = Invalid("Metadata cache loader returned null content for {}", location); + } + std::unique_lock lock(impl_->mutex_); + auto it = impl_->entries_.find(location); + if (it == impl_->entries_.end() || it->second != entry) { + if (loaded.has_value()) { + entry->content = loaded.value(); + } else { + entry->error = loaded.error(); + } + entry->loading = false; + entry->loaded.notify_all(); + return loaded; + } + + if (!loaded.has_value()) { + impl_->entries_.erase(it); + entry->error = loaded.error(); + entry->loading = false; + entry->loaded.notify_all(); + return loaded; + } + + if (loaded.value()->size() > impl_->options_.max_content_length || + loaded.value()->size() > impl_->options_.max_total_bytes) { + impl_->entries_.erase(it); + entry->content = loaded.value(); + entry->loading = false; + entry->loaded.notify_all(); + return loaded; + } + + impl_->PruneExpired(Impl::Clock::now()); + impl_->EvictToFit(loaded.value()->size()); + entry->content = loaded.value(); + entry->last_access = Impl::Clock::now(); + impl_->lru_.push_front(location); + entry->lru_position = impl_->lru_.begin(); + entry->loading = false; + impl_->total_bytes_ += entry->content->size(); + entry->loaded.notify_all(); + return entry->content; +} + +MetadataCache::Content MetadataCache::GetIfPresent(std::string_view location) { + if (!impl_->options_.enabled) { + return nullptr; + } + std::lock_guard lock(impl_->mutex_); + auto it = impl_->entries_.find(std::string(location)); + if (it == impl_->entries_.end() || it->second->loading) { + return nullptr; + } + auto now = Impl::Clock::now(); + if (impl_->IsExpired(*it->second, now)) { + impl_->EraseLoaded(it); + return nullptr; + } + impl_->Touch(it->first, *it->second, now); + return it->second->content; +} + +void MetadataCache::Invalidate(std::string_view location) { + std::unique_lock lock(impl_->mutex_); + while (true) { + auto it = impl_->entries_.find(std::string(location)); + if (it == impl_->entries_.end()) { + return; + } + auto entry = it->second; + if (entry->loading) { + entry->loaded.wait(lock, [&entry] { return !entry->loading; }); + continue; + } + impl_->EraseLoaded(it); + return; + } +} + +void MetadataCache::Clear() { + std::unique_lock lock(impl_->mutex_); + impl_->clearing_ = true; + while (true) { + auto loading = std::ranges::find_if( + impl_->entries_, [](const auto& item) { return item.second->loading; }); + if (loading == impl_->entries_.end()) { + break; + } + auto entry = loading->second; + entry->loaded.wait(lock, [&entry] { return !entry->loading; }); + } + impl_->entries_.clear(); + impl_->lru_.clear(); + impl_->total_bytes_ = 0; + impl_->clearing_ = false; + lock.unlock(); + impl_->clear_completed_.notify_all(); +} + +size_t MetadataCache::size() const { + std::lock_guard lock(impl_->mutex_); + return impl_->lru_.size(); +} + +size_t MetadataCache::total_bytes() const { + std::lock_guard lock(impl_->mutex_); + return impl_->total_bytes_; +} + +} // namespace iceberg diff --git a/src/iceberg/metadata_cache.h b/src/iceberg/metadata_cache.h new file mode 100644 index 000000000..7a9c26e12 --- /dev/null +++ b/src/iceberg/metadata_cache.h @@ -0,0 +1,109 @@ +/* + * 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 + +/// \file iceberg/metadata_cache.h +/// \brief A bounded cache for immutable Iceberg metadata file content. + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "iceberg/iceberg_export.h" +#include "iceberg/result.h" + +namespace iceberg { + +/// \brief Configuration for cached metadata file content. +/// +/// The property names and defaults match Apache Iceberg Java's manifest content cache. +/// The cache is disabled by default. +struct ICEBERG_EXPORT MetadataCacheOptions { + inline static constexpr std::string_view kEnabled = "io.manifest.cache-enabled"; + inline static constexpr std::string_view kExpirationIntervalMs = + "io.manifest.cache.expiration-interval-ms"; + 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; + /// Zero disables time-based expiration; positive values expire after last access. + int64_t expiration_interval_ms = 60 * 1000; + /// Maximum combined weight of cached entries. + size_t max_total_bytes = 100 * 1024 * 1024; + /// Files larger than this value are read without being cached. + size_t max_content_length = 8 * 1024 * 1024; + + friend bool operator==(const MetadataCacheOptions&, + const MetadataCacheOptions&) = default; + + /// \brief Parse cache options from catalog/FileIO properties. + static Result FromProperties( + const std::unordered_map& properties); +}; + +/// \brief Thread-safe, size-bounded cache of immutable metadata file bytes. +/// +/// Entries expire after the configured interval since their last access and are evicted +/// in least-recently-used order to enforce the total byte limit. Concurrent loads for +/// the same location are coalesced. Load failures are not cached. +class ICEBERG_EXPORT MetadataCache { + public: + using Content = std::shared_ptr; + using Loader = std::function()>; + + static Result> Make(MetadataCacheOptions options); + + ~MetadataCache(); + MetadataCache(const MetadataCache&) = delete; + MetadataCache& operator=(const MetadataCache&) = delete; + + const MetadataCacheOptions& options() const noexcept; + + /// \brief Return cached content or invoke loader and cache its result when eligible. + Result Get(std::string location, std::optional length, Loader loader); + + /// \brief Return an unexpired cached entry without loading it. + Content GetIfPresent(std::string_view location); + + /// \brief Invalidate one location, waiting for an in-progress load if necessary. + void Invalidate(std::string_view location); + + /// \brief Invalidate all currently cached entries. + void Clear(); + + size_t size() const; + size_t total_bytes() const; + + private: + class Impl; + + explicit MetadataCache(MetadataCacheOptions options); + + std::unique_ptr impl_; +}; + +} // namespace iceberg diff --git a/src/iceberg/snapshot.cc b/src/iceberg/snapshot.cc index 421d7439b..ded3f37c7 100644 --- a/src/iceberg/snapshot.cc +++ b/src/iceberg/snapshot.cc @@ -19,7 +19,10 @@ #include "iceberg/snapshot.h" +#include +#include #include +#include #include #include @@ -32,6 +35,67 @@ namespace iceberg { +namespace internal { + +class SnapshotCacheData { + public: + Result> Get( + const Snapshot* snapshot, std::shared_ptr file_io) { + std::shared_ptr entry; + { + std::unique_lock lock(mutex_); + while (true) { + auto [it, inserted] = entries_.try_emplace(snapshot->manifest_list); + if (inserted) { + it->second = std::make_shared(); + } + entry = it->second; + if (entry->value.has_value()) { + return std::ref(*entry->value); + } + if (!entry->loading) { + entry->loading = true; + break; + } + entry->loaded.wait(lock, [&entry] { return !entry->loading; }); + } + } + + auto loaded = SnapshotCache::InitManifestsCache(snapshot, std::move(file_io)); + std::lock_guard lock(mutex_); + auto it = entries_.find(snapshot->manifest_list); + if (!loaded.has_value()) { + if (it != entries_.end() && it->second == entry) { + entries_.erase(it); + } + entry->loading = false; + entry->loaded.notify_all(); + return std::unexpected(std::move(loaded).error()); + } + + entry->value = std::move(loaded).value(); + entry->loading = false; + entry->loaded.notify_all(); + return std::ref(*entry->value); + } + + private: + struct Entry { + bool loading = false; + std::optional value; + std::condition_variable loaded; + }; + + std::mutex mutex_; + std::unordered_map> entries_; +}; + +std::shared_ptr MakeSnapshotCacheData() { + return std::make_shared(); +} + +} // namespace internal + namespace { /// \brief Helper function to conditionally add a property to the summary @@ -236,27 +300,39 @@ Result SnapshotCache::InitManifestsCache( return std::make_pair(std::move(manifests), data_manifests_count); } -Result> SnapshotCache::Manifests( +Result> SnapshotCache::Manifests( std::shared_ptr file_io) const { - ICEBERG_ASSIGN_OR_RAISE(auto cache_ref, manifests_cache_.Get(snapshot_, file_io)); + ICEBERG_PRECHECK(snapshot_ != nullptr, "Cannot cache manifests for a null snapshot"); + ICEBERG_PRECHECK(snapshot_->cache_data != nullptr, + "Snapshot manifest cache state is null"); + ICEBERG_ASSIGN_OR_RAISE(auto cache_ref, + snapshot_->cache_data->Get(snapshot_, std::move(file_io))); auto& cache = cache_ref.get(); - return std::span(cache.first.data(), cache.first.size()); + return std::span(cache.first.data(), cache.first.size()); } -Result> SnapshotCache::DataManifests( +Result> SnapshotCache::DataManifests( std::shared_ptr file_io) const { - ICEBERG_ASSIGN_OR_RAISE(auto cache_ref, manifests_cache_.Get(snapshot_, file_io)); + ICEBERG_PRECHECK(snapshot_ != nullptr, "Cannot cache manifests for a null snapshot"); + ICEBERG_PRECHECK(snapshot_->cache_data != nullptr, + "Snapshot manifest cache state is null"); + ICEBERG_ASSIGN_OR_RAISE(auto cache_ref, + snapshot_->cache_data->Get(snapshot_, std::move(file_io))); auto& cache = cache_ref.get(); - return std::span(cache.first.data(), cache.second); + return std::span(cache.first.data(), cache.second); } -Result> SnapshotCache::DeleteManifests( +Result> SnapshotCache::DeleteManifests( std::shared_ptr file_io) const { - ICEBERG_ASSIGN_OR_RAISE(auto cache_ref, manifests_cache_.Get(snapshot_, file_io)); + ICEBERG_PRECHECK(snapshot_ != nullptr, "Cannot cache manifests for a null snapshot"); + ICEBERG_PRECHECK(snapshot_->cache_data != nullptr, + "Snapshot manifest cache state is null"); + ICEBERG_ASSIGN_OR_RAISE(auto cache_ref, + snapshot_->cache_data->Get(snapshot_, std::move(file_io))); auto& cache = cache_ref.get(); const size_t delete_start = cache.second; const size_t delete_count = cache.first.size() - delete_start; - return std::span(cache.first.data() + delete_start, delete_count); + return std::span(cache.first.data() + delete_start, delete_count); } // SnapshotSummaryBuilder::UpdateMetrics implementation diff --git a/src/iceberg/snapshot.h b/src/iceberg/snapshot.h index 9dc162be0..ce65ceb85 100644 --- a/src/iceberg/snapshot.h +++ b/src/iceberg/snapshot.h @@ -34,11 +34,15 @@ #include "iceberg/manifest/manifest_list.h" #include "iceberg/result.h" #include "iceberg/type_fwd.h" -#include "iceberg/util/lazy.h" #include "iceberg/util/timepoint.h" namespace iceberg { +namespace internal { +class SnapshotCacheData; +ICEBERG_EXPORT std::shared_ptr MakeSnapshotCacheData(); +} // namespace internal + /// \brief The type of snapshot reference enum class SnapshotRefType { /// Branches are mutable named references that can be updated by committing a new @@ -413,6 +417,10 @@ struct ICEBERG_EXPORT Snapshot { /// The upper bound of rows with assigned row IDs in this snapshot. std::optional added_rows; + /// Internal lazy state shared by Snapshot copies so manifest lists are parsed once. + mutable std::shared_ptr cache_data = + internal::MakeSnapshotCacheData(); + /// \brief Create a new Snapshot instance with validation on the inputs. static Result> Make( int64_t sequence_number, int64_t snapshot_id, @@ -464,7 +472,9 @@ struct ICEBERG_EXPORT Snapshot { /// \brief A snapshot with cached manifest loading capabilities. /// -/// This class wraps a Snapshot pointer and provides lazy-loading of manifests. +/// This class wraps a Snapshot pointer and provides lazy-loading of manifests. Parsed +/// manifest-list entries are stored on the Snapshot so separate wrappers and scans reuse +/// them. class ICEBERG_EXPORT SnapshotCache { public: explicit SnapshotCache(const Snapshot* snapshot) : snapshot_(snapshot) {} @@ -477,19 +487,21 @@ class ICEBERG_EXPORT SnapshotCache { /// /// \param file_io The FileIO instance to use for reading the manifest list /// \return A span of ManifestFile instances, or an error - Result> Manifests(std::shared_ptr file_io) const; + Result> Manifests(std::shared_ptr file_io) const; /// \brief Returns a ManifestFile for each data manifest in this snapshot. /// /// \param file_io The FileIO instance to use for reading the manifest list /// \return A span of ManifestFile instances, or an error - Result> DataManifests(std::shared_ptr file_io) const; + Result> DataManifests( + std::shared_ptr file_io) const; /// \brief Returns a ManifestFile for each delete manifest in this snapshot. /// /// \param file_io The FileIO instance to use for reading the manifest list /// \return A span of ManifestFile instances, or an error - Result> DeleteManifests(std::shared_ptr file_io) const; + Result> DeleteManifests( + std::shared_ptr file_io) const; private: /// \brief Cache structure for storing loaded manifests @@ -508,8 +520,7 @@ class ICEBERG_EXPORT SnapshotCache { /// The underlying snapshot data const Snapshot* snapshot_; - /// Lazy-loaded manifests cache - Lazy manifests_cache_; + friend class internal::SnapshotCacheData; }; } // namespace iceberg diff --git a/src/iceberg/table_metadata.cc b/src/iceberg/table_metadata.cc index 94a154501..5bf314098 100644 --- a/src/iceberg/table_metadata.cc +++ b/src/iceberg/table_metadata.cc @@ -464,7 +464,7 @@ Result> TableMetadataUtil::Read( FileIO& io, const std::string& location, std::optional length) { ICEBERG_ASSIGN_OR_RAISE(auto codec_type, Codec::FromFileName(location)); - ICEBERG_ASSIGN_OR_RAISE(auto content, io.ReadFile(location, length)); + ICEBERG_ASSIGN_OR_RAISE(auto content, io.ReadFileCached(location, length)); if (codec_type == MetadataFileCodecType::kGzip) { auto gzip_decompressor = std::make_unique(); ICEBERG_RETURN_UNEXPECTED(gzip_decompressor->Init()); diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index f1e2a3a78..5abed6ab1 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -135,6 +135,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/meson.build b/src/iceberg/test/meson.build index 09fa12061..f28f404d2 100644 --- a/src/iceberg/test/meson.build +++ b/src/iceberg/test/meson.build @@ -110,6 +110,7 @@ iceberg_tests = { 'lazy_test.cc', 'location_util_test.cc', 'math_util_internal_test.cc', + 'metadata_cache_test.cc', 'position_delete_index_test.cc', 'position_delete_range_consumer_test.cc', 'retry_util_test.cc', diff --git a/src/iceberg/test/metadata_cache_test.cc b/src/iceberg/test/metadata_cache_test.cc new file mode 100644 index 000000000..94325d2d0 --- /dev/null +++ b/src/iceberg/test/metadata_cache_test.cc @@ -0,0 +1,371 @@ +/* + * 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 "iceberg/metadata_cache.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "iceberg/test/matchers.h" +#include "iceberg/test/mock_io.h" +#include "iceberg/util/macros.h" + +namespace iceberg { +namespace { + +using namespace std::chrono_literals; + +MetadataCacheOptions EnabledOptions(size_t max_total_bytes = 100, + size_t max_content_length = 100) { + return { + .enabled = true, + .expiration_interval_ms = 0, + .max_total_bytes = max_total_bytes, + .max_content_length = max_content_length, + }; +} + +class CountingInputFile : public InputFile { + public: + CountingInputFile(std::unique_ptr input_file, int* open_count) + : input_file_(std::move(input_file)), open_count_(open_count) {} + + std::string_view location() const override { return input_file_->location(); } + + Result Size() const override { return input_file_->Size(); } + + Result> Open() override { + ++*open_count_; + return input_file_->Open(); + } + + private: + std::unique_ptr input_file_; + int* open_count_; +}; + +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); + } + + int open_count = 0; +}; + +TEST(MetadataCacheTest, ReusesContentAcrossLoads) { + auto cache_result = MetadataCache::Make(EnabledOptions()); + ASSERT_THAT(cache_result, IsOk()); + auto cache = std::move(cache_result).value(); + + int loads = 0; + auto loader = [&]() -> Result { + ++loads; + return std::make_shared("metadata"); + }; + + auto first = cache->Get("s3://warehouse/table/metadata/v1.metadata.json", 8, loader); + auto second = cache->Get("s3://warehouse/table/metadata/v1.metadata.json", 8, loader); + + ASSERT_THAT(first, IsOk()); + ASSERT_THAT(second, IsOk()); + EXPECT_EQ(loads, 1); + EXPECT_EQ(first.value(), second.value()); + EXPECT_EQ(cache->size(), 1); + EXPECT_EQ(cache->total_bytes(), 8); +} + +TEST(MetadataCacheTest, FileIOReusesCachedContentAcrossInputFiles) { + CountingMockFileIO file_io; + file_io.AddFile("manifest.avro", "manifest"); + ASSERT_THAT(file_io.ConfigureMetadataCache( + {{std::string(MetadataCacheOptions::kEnabled), "true"}}), + IsOk()); + + auto first_file = file_io.NewCachedInputFile("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 = file_io.NewCachedInputFile("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, FileIORejectsConflictingReconfiguration) { + CountingMockFileIO file_io; + std::unordered_map properties = { + {std::string(MetadataCacheOptions::kEnabled), "true"}, + }; + ASSERT_THAT(file_io.ConfigureMetadataCache(properties), IsOk()); + ASSERT_THAT(file_io.ConfigureMetadataCache(properties), IsOk()); + properties[std::string(MetadataCacheOptions::kMaxTotalBytes)] = "1024"; + + EXPECT_THAT(file_io.ConfigureMetadataCache(properties), + IsError(ErrorKind::kInvalidArgument)); +} + +TEST(MetadataCacheTest, FileIORejectsConcurrentConflictingConfiguration) { + CountingMockFileIO file_io; + std::barrier start(3); + auto configure = [&](size_t max_total_bytes) { + start.arrive_and_wait(); + return file_io.ConfigureMetadataCache( + {{std::string(MetadataCacheOptions::kEnabled), "true"}, + {std::string(MetadataCacheOptions::kMaxTotalBytes), + std::to_string(max_total_bytes)}}); + }; + auto first = std::async(std::launch::async, configure, 1024); + auto second = std::async(std::launch::async, configure, 2048); + start.arrive_and_wait(); + + auto first_result = first.get(); + auto second_result = second.get(); + + EXPECT_NE(first_result.has_value(), second_result.has_value()); + const auto& failed = first_result.has_value() ? second_result : first_result; + EXPECT_THAT(failed, IsError(ErrorKind::kInvalidArgument)); +} + +TEST(MetadataCacheTest, DisabledCacheAlwaysLoads) { + auto cache_result = MetadataCache::Make(MetadataCacheOptions{}); + ASSERT_THAT(cache_result, IsOk()); + auto cache = std::move(cache_result).value(); + int loads = 0; + auto loader = [&]() -> Result { + ++loads; + return std::make_shared("manifest"); + }; + + ASSERT_THAT(cache->Get("manifest.avro", 8, loader), IsOk()); + ASSERT_THAT(cache->Get("manifest.avro", 8, loader), IsOk()); + + EXPECT_EQ(loads, 2); + EXPECT_EQ(cache->size(), 0); +} + +TEST(MetadataCacheTest, DoesNotCacheLoadFailures) { + auto cache_result = MetadataCache::Make(EnabledOptions()); + ASSERT_THAT(cache_result, IsOk()); + auto cache = std::move(cache_result).value(); + int loads = 0; + auto loader = [&]() -> Result { + ++loads; + if (loads == 1) { + return IOError("temporary read failure"); + } + return std::make_shared("manifest"); + }; + + EXPECT_THAT(cache->Get("manifest.avro", 8, loader), IsError(ErrorKind::kIOError)); + EXPECT_THAT(cache->Get("manifest.avro", 8, loader), IsOk()); + + EXPECT_EQ(loads, 2); + EXPECT_EQ(cache->size(), 1); +} + +TEST(MetadataCacheTest, CoalescesConcurrentLoadFailures) { + auto cache_result = MetadataCache::Make(EnabledOptions()); + ASSERT_THAT(cache_result, IsOk()); + auto cache = std::move(cache_result).value(); + std::atomic loads = 0; + std::promise load_started; + std::promise release_load; + auto release = release_load.get_future().share(); + auto loader = [&]() -> Result { + if (++loads == 1) { + load_started.set_value(); + release.wait(); + } + return IOError("temporary read failure"); + }; + + auto first = std::async(std::launch::async, + [&] { return cache->Get("manifest.avro", 8, loader); }); + load_started.get_future().wait(); + std::promise second_started; + auto second = std::async(std::launch::async, [&] { + second_started.set_value(); + return cache->Get("manifest.avro", 8, loader); + }); + second_started.get_future().wait(); + EXPECT_EQ(second.wait_for(20ms), std::future_status::timeout); + release_load.set_value(); + + EXPECT_THAT(first.get(), IsError(ErrorKind::kIOError)); + EXPECT_THAT(second.get(), IsError(ErrorKind::kIOError)); + EXPECT_EQ(loads.load(), 1); +} + +TEST(MetadataCacheTest, SkipsContentAboveMaximumLength) { + auto cache_result = MetadataCache::Make(EnabledOptions(100, 4)); + ASSERT_THAT(cache_result, IsOk()); + auto cache = std::move(cache_result).value(); + int loads = 0; + auto loader = [&]() -> Result { + ++loads; + return std::make_shared("manifest"); + }; + + ASSERT_THAT(cache->Get("manifest.avro", 8, loader), IsOk()); + ASSERT_THAT(cache->Get("manifest.avro", 8, loader), IsOk()); + + EXPECT_EQ(loads, 2); + EXPECT_EQ(cache->size(), 0); +} + +TEST(MetadataCacheTest, EvictsLeastRecentlyUsedContentByWeight) { + auto cache_result = MetadataCache::Make(EnabledOptions(6)); + ASSERT_THAT(cache_result, IsOk()); + auto cache = std::move(cache_result).value(); + auto load = [](std::string content) { + return [content = std::move(content)]() -> Result { + return std::make_shared(content); + }; + }; + + ASSERT_THAT(cache->Get("a", 2, load("aa")), IsOk()); + ASSERT_THAT(cache->Get("b", 3, load("bbb")), IsOk()); + ASSERT_NE(cache->GetIfPresent("a"), nullptr); + ASSERT_THAT(cache->Get("c", 4, load("cccc")), IsOk()); + + EXPECT_NE(cache->GetIfPresent("a"), nullptr); + EXPECT_EQ(cache->GetIfPresent("b"), nullptr); + EXPECT_NE(cache->GetIfPresent("c"), nullptr); + EXPECT_EQ(cache->total_bytes(), 6); +} + +TEST(MetadataCacheTest, InvalidatesOneLocation) { + auto cache_result = MetadataCache::Make(EnabledOptions()); + ASSERT_THAT(cache_result, IsOk()); + auto cache = std::move(cache_result).value(); + auto loader = []() -> Result { + return std::make_shared("manifest-list"); + }; + ASSERT_THAT(cache->Get("snap.avro", 13, loader), IsOk()); + + cache->Invalidate("snap.avro"); + EXPECT_EQ(cache->GetIfPresent("snap.avro"), nullptr); + EXPECT_EQ(cache->size(), 0); + EXPECT_EQ(cache->total_bytes(), 0); +} + +TEST(MetadataCacheTest, ClearsCachedContent) { + auto cache_result = MetadataCache::Make(EnabledOptions()); + ASSERT_THAT(cache_result, IsOk()); + auto cache = std::move(cache_result).value(); + int loads = 0; + auto loader = [&]() -> Result { + ++loads; + return std::make_shared("manifest-list"); + }; + ASSERT_THAT(cache->Get("snap.avro", 13, loader), IsOk()); + + cache->Clear(); + + EXPECT_EQ(cache->size(), 0); + EXPECT_EQ(cache->total_bytes(), 0); + ASSERT_THAT(cache->Get("snap.avro", 13, loader), IsOk()); + EXPECT_EQ(loads, 2); +} + +TEST(MetadataCacheTest, ClearWaitsForInflightLoad) { + auto cache_result = MetadataCache::Make(EnabledOptions()); + ASSERT_THAT(cache_result, IsOk()); + auto cache = std::move(cache_result).value(); + std::promise load_started; + std::promise release_load; + auto release = release_load.get_future().share(); + auto load = std::async(std::launch::async, [&] { + return cache->Get("snap.avro", 13, [&]() -> Result { + load_started.set_value(); + release.wait(); + return std::make_shared("manifest-list"); + }); + }); + load_started.get_future().wait(); + std::promise clear_started; + auto clear = std::async(std::launch::async, [&] { + clear_started.set_value(); + cache->Clear(); + }); + clear_started.get_future().wait(); + EXPECT_EQ(clear.wait_for(20ms), std::future_status::timeout); + + release_load.set_value(); + EXPECT_THAT(load.get(), IsOk()); + clear.get(); + + EXPECT_EQ(cache->size(), 0); + EXPECT_EQ(cache->total_bytes(), 0); +} + +TEST(MetadataCacheTest, ParsesJavaCompatibleProperties) { + std::unordered_map properties = { + {std::string(MetadataCacheOptions::kEnabled), "true"}, + {std::string(MetadataCacheOptions::kExpirationIntervalMs), "1234"}, + {std::string(MetadataCacheOptions::kMaxTotalBytes), "200"}, + {std::string(MetadataCacheOptions::kMaxContentLength), "50"}, + }; + + auto options = MetadataCacheOptions::FromProperties(properties); + + ASSERT_THAT(options, IsOk()); + EXPECT_TRUE(options->enabled); + EXPECT_EQ(options->expiration_interval_ms, 1234); + EXPECT_EQ(options->max_total_bytes, 200); + EXPECT_EQ(options->max_content_length, 50); +} + +TEST(MetadataCacheTest, UsesJavaCompatibleDefaults) { + auto options = MetadataCacheOptions::FromProperties( + {{std::string(MetadataCacheOptions::kEnabled), "true"}}); + + ASSERT_THAT(options, IsOk()); + EXPECT_TRUE(options->enabled); + EXPECT_EQ(options->expiration_interval_ms, 60 * 1000); + EXPECT_EQ(options->max_total_bytes, 100 * 1024 * 1024); + EXPECT_EQ(options->max_content_length, 8 * 1024 * 1024); +} + +TEST(MetadataCacheTest, RejectsInvalidEnabledConfiguration) { + auto options = EnabledOptions(); + options.expiration_interval_ms = -1; + + EXPECT_THAT(MetadataCache::Make(options), IsError(ErrorKind::kInvalidArgument)); +} + +} // namespace +} // namespace iceberg diff --git a/src/iceberg/update/expire_snapshots.cc b/src/iceberg/update/expire_snapshots.cc index 5573efa77..ff7d808a9 100644 --- a/src/iceberg/update/expire_snapshots.cc +++ b/src/iceberg/update/expire_snapshots.cc @@ -268,8 +268,7 @@ class ReachableFileCleanup : public FileCleanupStrategy { SnapshotCache snapshot_cache(snapshot.get()); ICEBERG_ASSIGN_OR_RAISE(auto snapshot_manifests, snapshot_cache.Manifests(file_io_)); - return snapshot_manifests | std::views::as_rvalue | - std::ranges::to>(); + return snapshot_manifests | std::ranges::to>(); } /// \brief Collect manifests for a set of snapshots. From e3a8bb2002e43e47970f833050e17db6a889b878 Mon Sep 17 00:00:00 2001 From: manuzhang Date: Thu, 6 Aug 2026 19:09:39 +0800 Subject: [PATCH 2/5] fix: address metadata cache CI failures Co-authored-by: Codex --- src/iceberg/arrow/arrow_io_internal.h | 2 +- src/iceberg/file_io.cc | 18 +++++++++------ src/iceberg/file_io.h | 3 +-- src/iceberg/file_io_registry.cc | 5 ++++- src/iceberg/metadata_cache.cc | 7 +++++- src/iceberg/test/metadata_cache_test.cc | 29 +++++++++++++++++++++++++ 6 files changed, 52 insertions(+), 12 deletions(-) diff --git a/src/iceberg/arrow/arrow_io_internal.h b/src/iceberg/arrow/arrow_io_internal.h index 8f46130e6..14c824bb7 100644 --- a/src/iceberg/arrow/arrow_io_internal.h +++ b/src/iceberg/arrow/arrow_io_internal.h @@ -88,7 +88,7 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO { private: friend Result> OpenArrowInputStream( const std::shared_ptr& io, const std::string& path, - std::optional length); + std::optional length, bool cache_content); friend Result> OpenArrowOutputStream( const std::shared_ptr& io, const std::string& path, bool overwrite); diff --git a/src/iceberg/file_io.cc b/src/iceberg/file_io.cc index 345549454..76472f416 100644 --- a/src/iceberg/file_io.cc +++ b/src/iceberg/file_io.cc @@ -20,6 +20,7 @@ #include "iceberg/file_io.h" #include +#include #include #include #include @@ -201,7 +202,7 @@ Result> FileIO::NewCachedInputFile( ICEBERG_ASSIGN_OR_RAISE(input_file, NewInputFile(file_location)); } - auto cache = metadata_cache_.load(std::memory_order_acquire); + auto cache = std::atomic_load_explicit(&metadata_cache_, std::memory_order_acquire); if (cache == nullptr || !cache->options().enabled) { return input_file; } @@ -231,7 +232,7 @@ Result> FileIO::NewCachedInputFile( Result FileIO::ReadFileCached(const std::string& file_location, std::optional length) { - auto cache = metadata_cache_.load(std::memory_order_acquire); + auto cache = std::atomic_load_explicit(&metadata_cache_, std::memory_order_acquire); if (cache == nullptr || !cache->options().enabled) { return ReadFile(file_location, length); } @@ -251,8 +252,9 @@ Status FileIO::ConfigureMetadataCache( ICEBERG_ASSIGN_OR_RAISE(auto cache, MetadataCache::Make(options)); std::shared_ptr expected; while (expected == nullptr) { - if (metadata_cache_.compare_exchange_strong( - expected, cache, std::memory_order_release, std::memory_order_acquire)) { + if (std::atomic_compare_exchange_strong_explicit( + &metadata_cache_, &expected, cache, std::memory_order_acq_rel, + std::memory_order_acquire)) { return {}; } } @@ -263,18 +265,20 @@ Status FileIO::ConfigureMetadataCache( } bool FileIO::MetadataCacheEnabled() const noexcept { - auto cache = metadata_cache_.load(std::memory_order_acquire); + auto cache = std::atomic_load_explicit(&metadata_cache_, std::memory_order_acquire); return cache != nullptr && cache->options().enabled; } void FileIO::InvalidateMetadataCache(std::string_view file_location) { - if (auto cache = metadata_cache_.load(std::memory_order_acquire)) { + if (auto cache = + std::atomic_load_explicit(&metadata_cache_, std::memory_order_acquire)) { cache->Invalidate(file_location); } } void FileIO::ClearMetadataCache() { - if (auto cache = metadata_cache_.load(std::memory_order_acquire)) { + if (auto cache = + std::atomic_load_explicit(&metadata_cache_, std::memory_order_acquire)) { cache->Clear(); } } diff --git a/src/iceberg/file_io.h b/src/iceberg/file_io.h index bed694360..b27633a6c 100644 --- a/src/iceberg/file_io.h +++ b/src/iceberg/file_io.h @@ -22,7 +22,6 @@ /// \file iceberg/file_io.h /// \brief Define the FileIO abstraction and file stream interfaces. -#include #include #include #include @@ -212,7 +211,7 @@ class ICEBERG_EXPORT FileIO { virtual SupportsStorageCredentials* AsSupportsStorageCredentials() { return nullptr; } private: - std::atomic> metadata_cache_; + std::shared_ptr metadata_cache_; }; /// \brief Mix-in for FileIO implementations that route object paths to diff --git a/src/iceberg/file_io_registry.cc b/src/iceberg/file_io_registry.cc index dc88a6197..09ed2c464 100644 --- a/src/iceberg/file_io_registry.cc +++ b/src/iceberg/file_io_registry.cc @@ -22,6 +22,7 @@ #include #include +#include "iceberg/metadata_cache.h" #include "iceberg/util/macros.h" namespace iceberg { @@ -61,7 +62,9 @@ Result> FileIORegistry::Load( } ICEBERG_ASSIGN_OR_RAISE(auto io, factory(properties)); ICEBERG_PRECHECK(io != nullptr, "FileIO factory returned null for {}", name); - ICEBERG_RETURN_UNEXPECTED(io->ConfigureMetadataCache(properties)); + if (properties.contains(std::string(MetadataCacheOptions::kEnabled))) { + ICEBERG_RETURN_UNEXPECTED(io->ConfigureMetadataCache(properties)); + } return io; } diff --git a/src/iceberg/metadata_cache.cc b/src/iceberg/metadata_cache.cc index 661090f20..75ac98a05 100644 --- a/src/iceberg/metadata_cache.cc +++ b/src/iceberg/metadata_cache.cc @@ -52,7 +52,12 @@ Result ParseNumberProperty( if (it == properties.end()) { return default_value; } - return StringUtils::ParseNumber(it->second); + auto value = StringUtils::ParseNumber(it->second); + if (!value.has_value()) { + return InvalidArgument("Invalid numeric value '{}' for property '{}': {}", it->second, + key, value.error().message); + } + return value; } } // namespace diff --git a/src/iceberg/test/metadata_cache_test.cc b/src/iceberg/test/metadata_cache_test.cc index 94325d2d0..8a5236a29 100644 --- a/src/iceberg/test/metadata_cache_test.cc +++ b/src/iceberg/test/metadata_cache_test.cc @@ -31,6 +31,7 @@ #include +#include "iceberg/file_io_registry.h" #include "iceberg/test/matchers.h" #include "iceberg/test/mock_io.h" #include "iceberg/util/macros.h" @@ -137,6 +138,25 @@ TEST(MetadataCacheTest, FileIORejectsConflictingReconfiguration) { IsError(ErrorKind::kInvalidArgument)); } +TEST(MetadataCacheTest, RegistryWithoutCachePropertiesAllowsLaterConfiguration) { + constexpr std::string_view kFileIOName = "metadata-cache-test"; + FileIORegistry::Register( + std::string(kFileIOName), + [](const std::unordered_map& /*properties*/) + -> Result> { + return std::make_unique(); + }); + + auto loaded = FileIORegistry::Load(std::string(kFileIOName), {}); + ASSERT_THAT(loaded, IsOk()); + auto file_io = std::move(loaded).value(); + EXPECT_FALSE(file_io->MetadataCacheEnabled()); + EXPECT_THAT(file_io->ConfigureMetadataCache( + {{std::string(MetadataCacheOptions::kEnabled), "true"}}), + IsOk()); + EXPECT_TRUE(file_io->MetadataCacheEnabled()); +} + TEST(MetadataCacheTest, FileIORejectsConcurrentConflictingConfiguration) { CountingMockFileIO file_io; std::barrier start(3); @@ -367,5 +387,14 @@ TEST(MetadataCacheTest, RejectsInvalidEnabledConfiguration) { EXPECT_THAT(MetadataCache::Make(options), IsError(ErrorKind::kInvalidArgument)); } +TEST(MetadataCacheTest, InvalidNumericPropertyErrorIncludesPropertyName) { + auto options = MetadataCacheOptions::FromProperties( + {{std::string(MetadataCacheOptions::kEnabled), "true"}, + {std::string(MetadataCacheOptions::kMaxTotalBytes), "invalid"}}); + + EXPECT_THAT(options, + HasErrorMessage(std::string(MetadataCacheOptions::kMaxTotalBytes))); +} + } // namespace } // namespace iceberg From 6ddf055751952ef25be89331e5ff3b8f6426e743 Mon Sep 17 00:00:00 2001 From: manuzhang Date: Thu, 6 Aug 2026 19:31:05 +0800 Subject: [PATCH 3/5] fix: make metadata cache synchronization portable Co-authored-by: Codex --- src/iceberg/file_io.cc | 35 ++++++++++++++++++----------------- src/iceberg/file_io.h | 6 +++++- 2 files changed, 23 insertions(+), 18 deletions(-) diff --git a/src/iceberg/file_io.cc b/src/iceberg/file_io.cc index 76472f416..126180a1a 100644 --- a/src/iceberg/file_io.cc +++ b/src/iceberg/file_io.cc @@ -20,7 +20,6 @@ #include "iceberg/file_io.h" #include -#include #include #include #include @@ -202,7 +201,7 @@ Result> FileIO::NewCachedInputFile( ICEBERG_ASSIGN_OR_RAISE(input_file, NewInputFile(file_location)); } - auto cache = std::atomic_load_explicit(&metadata_cache_, std::memory_order_acquire); + auto cache = GetMetadataCache(); if (cache == nullptr || !cache->options().enabled) { return input_file; } @@ -232,7 +231,7 @@ Result> FileIO::NewCachedInputFile( Result FileIO::ReadFileCached(const std::string& file_location, std::optional length) { - auto cache = std::atomic_load_explicit(&metadata_cache_, std::memory_order_acquire); + auto cache = GetMetadataCache(); if (cache == nullptr || !cache->options().enabled) { return ReadFile(file_location, length); } @@ -250,39 +249,41 @@ Status FileIO::ConfigureMetadataCache( const std::unordered_map& properties) { ICEBERG_ASSIGN_OR_RAISE(auto options, MetadataCacheOptions::FromProperties(properties)); ICEBERG_ASSIGN_OR_RAISE(auto cache, MetadataCache::Make(options)); - std::shared_ptr expected; - while (expected == nullptr) { - if (std::atomic_compare_exchange_strong_explicit( - &metadata_cache_, &expected, cache, std::memory_order_acq_rel, - std::memory_order_acquire)) { - return {}; - } + std::lock_guard lock(metadata_cache_mutex_); + if (metadata_cache_ == nullptr) { + metadata_cache_ = std::move(cache); + return {}; } - if (expected->options() == options) { + if (metadata_cache_->options() == options) { return {}; } return InvalidArgument("Metadata cache is already configured with different options"); } -bool FileIO::MetadataCacheEnabled() const noexcept { - auto cache = std::atomic_load_explicit(&metadata_cache_, std::memory_order_acquire); +bool FileIO::MetadataCacheEnabled() const { + auto cache = GetMetadataCache(); return cache != nullptr && cache->options().enabled; } void FileIO::InvalidateMetadataCache(std::string_view file_location) { - if (auto cache = - std::atomic_load_explicit(&metadata_cache_, std::memory_order_acquire)) { + auto cache = GetMetadataCache(); + if (cache != nullptr) { cache->Invalidate(file_location); } } void FileIO::ClearMetadataCache() { - if (auto cache = - std::atomic_load_explicit(&metadata_cache_, std::memory_order_acquire)) { + auto cache = GetMetadataCache(); + if (cache != nullptr) { cache->Clear(); } } +std::shared_ptr FileIO::GetMetadataCache() const { + std::lock_guard lock(metadata_cache_mutex_); + return metadata_cache_; +} + Status FileIO::WriteFile(const std::string& file_location, std::string_view content) { ICEBERG_ASSIGN_OR_RAISE(auto output_file, NewOutputFile(file_location)); ICEBERG_ASSIGN_OR_RAISE(auto stream, output_file->CreateOrOverwrite()); diff --git a/src/iceberg/file_io.h b/src/iceberg/file_io.h index b27633a6c..9c5d211df 100644 --- a/src/iceberg/file_io.h +++ b/src/iceberg/file_io.h @@ -25,6 +25,7 @@ #include #include #include +#include #include #include #include @@ -174,7 +175,7 @@ class ICEBERG_EXPORT FileIO { const std::unordered_map& properties); /// \brief Return whether metadata content caching is enabled. - bool MetadataCacheEnabled() const noexcept; + bool MetadataCacheEnabled() const; /// \brief Invalidate a cached metadata file location. void InvalidateMetadataCache(std::string_view file_location); @@ -211,6 +212,9 @@ class ICEBERG_EXPORT FileIO { virtual SupportsStorageCredentials* AsSupportsStorageCredentials() { return nullptr; } private: + std::shared_ptr GetMetadataCache() const; + + mutable std::mutex metadata_cache_mutex_; std::shared_ptr metadata_cache_; }; From 54434f440bdcb1da6854b329b543caecdacba8ab Mon Sep 17 00:00:00 2001 From: manuzhang Date: Thu, 6 Aug 2026 19:44:11 +0800 Subject: [PATCH 4/5] fix: preserve FileIO copyability Co-authored-by: Codex --- src/iceberg/file_io.cc | 12 ++++++------ src/iceberg/file_io.h | 9 +++++++-- 2 files changed, 13 insertions(+), 8 deletions(-) diff --git a/src/iceberg/file_io.cc b/src/iceberg/file_io.cc index 126180a1a..fe28dff83 100644 --- a/src/iceberg/file_io.cc +++ b/src/iceberg/file_io.cc @@ -249,12 +249,12 @@ Status FileIO::ConfigureMetadataCache( const std::unordered_map& properties) { ICEBERG_ASSIGN_OR_RAISE(auto options, MetadataCacheOptions::FromProperties(properties)); ICEBERG_ASSIGN_OR_RAISE(auto cache, MetadataCache::Make(options)); - std::lock_guard lock(metadata_cache_mutex_); - if (metadata_cache_ == nullptr) { - metadata_cache_ = std::move(cache); + std::lock_guard lock(metadata_cache_state_->mutex); + if (metadata_cache_state_->cache == nullptr) { + metadata_cache_state_->cache = std::move(cache); return {}; } - if (metadata_cache_->options() == options) { + if (metadata_cache_state_->cache->options() == options) { return {}; } return InvalidArgument("Metadata cache is already configured with different options"); @@ -280,8 +280,8 @@ void FileIO::ClearMetadataCache() { } std::shared_ptr FileIO::GetMetadataCache() const { - std::lock_guard lock(metadata_cache_mutex_); - return metadata_cache_; + std::lock_guard lock(metadata_cache_state_->mutex); + return metadata_cache_state_->cache; } Status FileIO::WriteFile(const std::string& file_location, std::string_view content) { diff --git a/src/iceberg/file_io.h b/src/iceberg/file_io.h index 9c5d211df..7f0d6431c 100644 --- a/src/iceberg/file_io.h +++ b/src/iceberg/file_io.h @@ -212,10 +212,15 @@ class ICEBERG_EXPORT FileIO { virtual SupportsStorageCredentials* AsSupportsStorageCredentials() { return nullptr; } private: + struct MetadataCacheState { + std::mutex mutex; + std::shared_ptr cache; + }; + std::shared_ptr GetMetadataCache() const; - mutable std::mutex metadata_cache_mutex_; - std::shared_ptr metadata_cache_; + std::shared_ptr metadata_cache_state_ = + std::make_shared(); }; /// \brief Mix-in for FileIO implementations that route object paths to From d70ca602ca44f69ef6a40054776486e8f91daf7c Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Thu, 6 Aug 2026 21:52:20 +0800 Subject: [PATCH 5/5] fix: address metadata cache review feedback Co-authored-by: Codex --- src/iceberg/file_io.cc | 11 ++++++++++- src/iceberg/file_io.h | 8 ++++++++ src/iceberg/snapshot.cc | 21 +++++++++------------ src/iceberg/snapshot.h | 8 ++------ 4 files changed, 29 insertions(+), 19 deletions(-) diff --git a/src/iceberg/file_io.cc b/src/iceberg/file_io.cc index fe28dff83..45db72fc8 100644 --- a/src/iceberg/file_io.cc +++ b/src/iceberg/file_io.cc @@ -25,6 +25,7 @@ #include #include "iceberg/metadata_cache.h" +#include "iceberg/snapshot.h" #include "iceberg/util/macros.h" namespace iceberg { @@ -223,7 +224,7 @@ Result> FileIO::NewCachedInputFile( if (size < 0) { return Invalid("Invalid negative file size {} for {}", size, file_location); } - if (static_cast(size) > cache->options().max_content_length) { + if (std::cmp_greater(size, cache->options().max_content_length)) { return input_file; } return std::make_unique(std::move(input_file), std::move(cache), size); @@ -284,6 +285,14 @@ std::shared_ptr FileIO::GetMetadataCache() const { return metadata_cache_state_->cache; } +std::shared_ptr FileIO::GetSnapshotCacheData() const { + std::lock_guard lock(metadata_cache_state_->mutex); + if (metadata_cache_state_->snapshot_cache == nullptr) { + metadata_cache_state_->snapshot_cache = internal::MakeSnapshotCacheData(); + } + return metadata_cache_state_->snapshot_cache; +} + Status FileIO::WriteFile(const std::string& file_location, std::string_view content) { ICEBERG_ASSIGN_OR_RAISE(auto output_file, NewOutputFile(file_location)); ICEBERG_ASSIGN_OR_RAISE(auto stream, output_file->CreateOrOverwrite()); diff --git a/src/iceberg/file_io.h b/src/iceberg/file_io.h index 7f0d6431c..9c5d69d9d 100644 --- a/src/iceberg/file_io.h +++ b/src/iceberg/file_io.h @@ -41,6 +41,10 @@ namespace iceberg { class SupportsStorageCredentials; class MetadataCache; +class SnapshotCache; +namespace internal { +class SnapshotCacheData; +} /// \brief Seekable byte stream for reading file contents. class ICEBERG_EXPORT SeekableInputStream { @@ -215,12 +219,16 @@ class ICEBERG_EXPORT FileIO { struct MetadataCacheState { std::mutex mutex; std::shared_ptr cache; + std::shared_ptr snapshot_cache; }; std::shared_ptr GetMetadataCache() const; + std::shared_ptr GetSnapshotCacheData() const; std::shared_ptr metadata_cache_state_ = std::make_shared(); + + friend class SnapshotCache; }; /// \brief Mix-in for FileIO implementations that route object paths to diff --git a/src/iceberg/snapshot.cc b/src/iceberg/snapshot.cc index ded3f37c7..a5346225a 100644 --- a/src/iceberg/snapshot.cc +++ b/src/iceberg/snapshot.cc @@ -303,10 +303,9 @@ Result SnapshotCache::InitManifestsCache( Result> SnapshotCache::Manifests( std::shared_ptr file_io) const { ICEBERG_PRECHECK(snapshot_ != nullptr, "Cannot cache manifests for a null snapshot"); - ICEBERG_PRECHECK(snapshot_->cache_data != nullptr, - "Snapshot manifest cache state is null"); - ICEBERG_ASSIGN_OR_RAISE(auto cache_ref, - snapshot_->cache_data->Get(snapshot_, std::move(file_io))); + ICEBERG_PRECHECK(file_io != nullptr, "Cannot cache manifests: FileIO is null"); + auto cache_data = file_io->GetSnapshotCacheData(); + ICEBERG_ASSIGN_OR_RAISE(auto cache_ref, cache_data->Get(snapshot_, std::move(file_io))); auto& cache = cache_ref.get(); return std::span(cache.first.data(), cache.first.size()); } @@ -314,10 +313,9 @@ Result> SnapshotCache::Manifests( Result> SnapshotCache::DataManifests( std::shared_ptr file_io) const { ICEBERG_PRECHECK(snapshot_ != nullptr, "Cannot cache manifests for a null snapshot"); - ICEBERG_PRECHECK(snapshot_->cache_data != nullptr, - "Snapshot manifest cache state is null"); - ICEBERG_ASSIGN_OR_RAISE(auto cache_ref, - snapshot_->cache_data->Get(snapshot_, std::move(file_io))); + ICEBERG_PRECHECK(file_io != nullptr, "Cannot cache manifests: FileIO is null"); + auto cache_data = file_io->GetSnapshotCacheData(); + ICEBERG_ASSIGN_OR_RAISE(auto cache_ref, cache_data->Get(snapshot_, std::move(file_io))); auto& cache = cache_ref.get(); return std::span(cache.first.data(), cache.second); } @@ -325,10 +323,9 @@ Result> SnapshotCache::DataManifests( Result> SnapshotCache::DeleteManifests( std::shared_ptr file_io) const { ICEBERG_PRECHECK(snapshot_ != nullptr, "Cannot cache manifests for a null snapshot"); - ICEBERG_PRECHECK(snapshot_->cache_data != nullptr, - "Snapshot manifest cache state is null"); - ICEBERG_ASSIGN_OR_RAISE(auto cache_ref, - snapshot_->cache_data->Get(snapshot_, std::move(file_io))); + ICEBERG_PRECHECK(file_io != nullptr, "Cannot cache manifests: FileIO is null"); + auto cache_data = file_io->GetSnapshotCacheData(); + ICEBERG_ASSIGN_OR_RAISE(auto cache_ref, cache_data->Get(snapshot_, std::move(file_io))); auto& cache = cache_ref.get(); const size_t delete_start = cache.second; const size_t delete_count = cache.first.size() - delete_start; diff --git a/src/iceberg/snapshot.h b/src/iceberg/snapshot.h index ce65ceb85..d66c06c32 100644 --- a/src/iceberg/snapshot.h +++ b/src/iceberg/snapshot.h @@ -417,10 +417,6 @@ struct ICEBERG_EXPORT Snapshot { /// The upper bound of rows with assigned row IDs in this snapshot. std::optional added_rows; - /// Internal lazy state shared by Snapshot copies so manifest lists are parsed once. - mutable std::shared_ptr cache_data = - internal::MakeSnapshotCacheData(); - /// \brief Create a new Snapshot instance with validation on the inputs. static Result> Make( int64_t sequence_number, int64_t snapshot_id, @@ -473,8 +469,8 @@ struct ICEBERG_EXPORT Snapshot { /// \brief A snapshot with cached manifest loading capabilities. /// /// This class wraps a Snapshot pointer and provides lazy-loading of manifests. Parsed -/// manifest-list entries are stored on the Snapshot so separate wrappers and scans reuse -/// them. +/// manifest-list entries are stored in FileIO-owned internal state so separate wrappers +/// and scans reuse them without changing Snapshot's public layout. class ICEBERG_EXPORT SnapshotCache { public: explicit SnapshotCache(const Snapshot* snapshot) : snapshot_(snapshot) {}