diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index eee720613..782cd3d7e 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_internal.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_internal.cc b/src/iceberg/content_cache_internal.cc new file mode 100644 index 000000000..05f417c1a --- /dev/null +++ b/src/iceberg/content_cache_internal.cc @@ -0,0 +1,305 @@ +/* + * 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/content_cache_internal.h" + +#include +#include +#include +#include + +#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