feat: cache immutable manifest content per table

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

src/iceberg/arrow/arrow_io.cc

@@ -22,6 +22,7 @@
 #include <limits>
 #include <mutex>
 #include <optional>
+#include <utility>
 #include <vector>
 
 #include <arrow/buffer.h>
@@ -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<std::unique_ptr<SeekableInputStream>> Open() override {
+    ICEBERG_ASSIGN_OR_RAISE(auto input, OpenArrowFile());
+    return std::make_unique<ArrowSeekableInputStream>(std::move(input));
+  }
+
+  Result<std::shared_ptr<::arrow::io::RandomAccessFile>> 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<ArrowSeekableInputStream>(std::move(input));
+    return input;
   }
 
  private:
   std::shared_ptr<::arrow::fs::FileSystem> fs_;
   std::string location_;
   std::string path_;
-  std::optional<int64_t> 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<int64_t> file_size_;
 };
 
 class ArrowOutputFile : public OutputFile {
@@ -530,6 +548,9 @@ Result<std::shared_ptr<::arrow::io::RandomAccessFile>> OpenArrowInputStream(
   } else {
     ICEBERG_ASSIGN_OR_RAISE(input_file, io->NewInputFile(path));
   }
+  if (auto* arrow_input = dynamic_cast<ArrowInputFile*>(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);

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -23,6 +23,7 @@
 #include <iterator>
 
 #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<std::shared_ptr<InMemoryCatalog>> InMemoryCatalog::Make(
     const std::string& name, const std::shared_ptr<FileIO>& file_io,
     const std::string& warehouse_location,
     const std::unordered_map<std::string, std::string>& properties) {
+  ICEBERG_PRECHECK(file_io != nullptr, "InMemoryCatalog requires a non-null FileIO");
+  ICEBERG_RETURN_UNEXPECTED(internal::ContentCacheOptions::FromProperties(properties));
   std::shared_ptr<MetricsReporter> reporter;
   auto it = properties.find(std::string(kMetricsReporterImpl));
   if (it != properties.end() && !it->second.empty() &&
@@ -445,7 +448,8 @@ Result<std::shared_ptr<Table>> 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<std::shared_ptr<Table>> InMemoryCatalog::UpdateTable(
@@ -497,7 +501,8 @@ Result<std::shared_ptr<Table>> 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<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
@@ -517,10 +522,10 @@ Result<std::shared_ptr<Transaction>> 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<std::shared_ptr<Table>> 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<std::shared_ptr<Table>> InMemoryCatalog::RegisterTable(
@@ -624,7 +630,7 @@ Result<std::shared_ptr<Table>> 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

src/iceberg/catalog/rest/rest_catalog.cc

@@ -923,12 +923,14 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
   auto table_catalog = std::make_shared<TableScopedCatalog>(
       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<std::shared_ptr<Table>> RestCatalog::MakeTable(
     std::shared_ptr<Catalog> table_catalog, std::shared_ptr<MetricsReporter> reporter,
     const std::unordered_map<std::string, std::string>& table_config,
     std::shared_ptr<auth::AuthSession> table_session, std::unique_ptr<Labels> 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<std::shared_ptr<Table>> 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

src/iceberg/catalog/rest/rest_file_io.cc

@@ -25,6 +25,7 @@
 #include <vector>
 
 #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<std::string, std::string> MergeFileIOProperties(
 }  // namespace
 
 Result<std::unique_ptr<FileIO>> 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<ResolvingFileIO>(config.configs());

src/iceberg/catalog/rest/rest_table.cc

@@ -48,13 +48,19 @@ Result<std::shared_ptr<RestTable>> RestTable::Make(
     std::shared_ptr<Catalog> catalog, std::string full_name,
     std::shared_ptr<MetricsReporter> reporter, RestScanContext rest_context,
     std::unique_ptr<Labels> labels) {
+  ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
   if (metadata == nullptr) {
     return InvalidArgument("Metadata cannot be null");
   }
-  return std::shared_ptr<RestTable>(new RestTable(
+  auto table = std::shared_ptr<RestTable>(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<std::unique_ptr<DataTableScanBuilder>> RestTable::NewScan() const {

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<std::shared_ptr<SqlCatalog>> 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<MetricsReporter> reporter;
@@ -385,7 +387,7 @@ Result<std::shared_ptr<Table>> 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<std::shared_ptr<Table>> SqlCatalog::LoadTable(const TableIdentifier& identifier) {
@@ -424,7 +426,7 @@ Result<std::shared_ptr<Table>> 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<std::shared_ptr<Table>> SqlCatalog::UpdateTable(
@@ -490,7 +492,7 @@ Result<std::shared_ptr<Table>> 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<std::shared_ptr<Transaction>> SqlCatalog::StageCreateTable(
@@ -518,7 +520,8 @@ Result<std::shared_ptr<Transaction>> 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<std::shared_ptr<Table>> 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);
 }
 
 // --------------------------------------------------------------------------