feat: add cross-query metadata and snapshot caching

src/iceberg/CMakeLists.txt

@@ -68,6 +68,7 @@ set(ICEBERG_SOURCES
     manifest/v1_metadata.cc
     manifest/v2_metadata.cc
     manifest/v3_metadata.cc
+    metadata_cache.cc
     metadata_columns.cc
     metrics_config.cc
     metrics/commit_report.cc

src/iceberg/arrow/arrow_io.cc

@@ -508,24 +508,28 @@ Result<std::string> ArrowFileSystemFileIO::ResolvePath(const std::string& file_l
 
 Result<std::shared_ptr<::arrow::io::RandomAccessFile>> OpenArrowInputStream(
     const std::shared_ptr<FileIO>& io, const std::string& path,
-    std::optional<size_t> length) {
+    std::optional<size_t> length, bool cache_content) {
   ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
 
-  if (auto arrow_io = std::dynamic_pointer_cast<ArrowFileSystemFileIO>(io)) {
-    ICEBERG_ASSIGN_OR_RAISE(auto resolved_path, arrow_io->ResolvePath(path));
-    ::arrow::fs::FileInfo file_info(resolved_path, ::arrow::fs::FileType::File);
-    if (length.has_value()) {
-      ICEBERG_ASSIGN_OR_RAISE(auto size, ToInt64Length(*length));
-      file_info.set_size(size);
+  if (!cache_content || !io->MetadataCacheEnabled()) {
+    if (auto arrow_io = std::dynamic_pointer_cast<ArrowFileSystemFileIO>(io)) {
+      ICEBERG_ASSIGN_OR_RAISE(auto resolved_path, arrow_io->ResolvePath(path));
+      ::arrow::fs::FileInfo file_info(resolved_path, ::arrow::fs::FileType::File);
+      if (length.has_value()) {
+        ICEBERG_ASSIGN_OR_RAISE(auto size, ToInt64Length(*length));
+        file_info.set_size(size);
+      }
+      ICEBERG_ARROW_ASSIGN_OR_RETURN(auto input,
+                                     arrow_io->arrow_fs_->OpenInputFile(file_info));
+      return input;
     }
-    ICEBERG_ARROW_ASSIGN_OR_RETURN(auto input,
-                                   arrow_io->arrow_fs_->OpenInputFile(file_info));
-    return input;
   }
 
   int64_t size;
   std::unique_ptr<InputFile> input_file;
-  if (length.has_value()) {
+  if (cache_content) {
+    ICEBERG_ASSIGN_OR_RAISE(input_file, io->NewCachedInputFile(path, length));
+  } else if (length.has_value()) {
     ICEBERG_ASSIGN_OR_RAISE(input_file, io->NewInputFile(path, *length));
   } else {
     ICEBERG_ASSIGN_OR_RAISE(input_file, io->NewInputFile(path));

src/iceberg/arrow/arrow_io_internal.h

@@ -40,7 +40,8 @@ namespace iceberg::arrow {
 /// implement NewInputFile.
 ICEBERG_BUNDLE_EXPORT Result<std::shared_ptr<::arrow::io::RandomAccessFile>>
 OpenArrowInputStream(const std::shared_ptr<FileIO>& io, const std::string& path,
-                     std::optional<size_t> length = std::nullopt);
+                     std::optional<size_t> length = std::nullopt,
+                     bool cache_content = false);
 
 /// \brief Open a FileIO output as an Arrow output stream.
 ///
@@ -87,7 +88,7 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
  private:
   friend Result<std::shared_ptr<::arrow::io::RandomAccessFile>> OpenArrowInputStream(
       const std::shared_ptr<FileIO>& io, const std::string& path,
-      std::optional<size_t> length);
+      std::optional<size_t> length, bool cache_content);
 
   friend Result<std::shared_ptr<::arrow::io::OutputStream>> OpenArrowOutputStream(
       const std::shared_ptr<FileIO>& io, const std::string& path, bool overwrite);

src/iceberg/avro/avro_reader.cc

@@ -54,7 +54,8 @@ namespace {
 Result<std::unique_ptr<AvroInputStream>> CreateInputStream(const ReaderOptions& options,
                                                            int64_t buffer_size) {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto file, arrow::OpenArrowInputStream(options.io, options.path, options.length));
+      auto file, arrow::OpenArrowInputStream(options.io, options.path, options.length,
+                                             options.cache_content));
   return std::make_unique<AvroInputStream>(file, buffer_size);
 }
 

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -24,6 +24,7 @@
 
 #include "iceberg/catalog/catalog_util.h"
 #include "iceberg/file_io.h"
+#include "iceberg/metadata_cache.h"
 #include "iceberg/metrics/metrics_reporters.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
@@ -343,6 +344,10 @@ Result<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");
+  if (properties.contains(std::string(MetadataCacheOptions::kEnabled))) {
+    ICEBERG_RETURN_UNEXPECTED(file_io->ConfigureMetadataCache(properties));
+  }
   std::shared_ptr<MetricsReporter> reporter;
   auto it = properties.find(std::string(kMetricsReporterImpl));
   if (it != properties.end() && !it->second.empty() &&

src/iceberg/catalog/sql/sql_catalog.cc

@@ -26,6 +26,7 @@
 #include "iceberg/catalog/catalog_util.h"
 #include "iceberg/catalog/sql/config.h"
 #include "iceberg/file_io.h"
+#include "iceberg/metadata_cache.h"
 #include "iceberg/metrics/metrics_reporters.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
@@ -148,6 +149,9 @@ Result<std::shared_ptr<SqlCatalog>> SqlCatalog::Make(
   if (file_io == nullptr) {
     return InvalidArgument("SqlCatalog requires a non-null FileIO");
   }
+  if (config.props.contains(std::string(MetadataCacheOptions::kEnabled))) {
+    ICEBERG_RETURN_UNEXPECTED(file_io->ConfigureMetadataCache(config.props));
+  }
   ICEBERG_RETURN_UNEXPECTED(store->Initialize());
 
   std::shared_ptr<MetricsReporter> reporter;

src/iceberg/file_io.cc

@@ -19,9 +19,13 @@
 
 #include "iceberg/file_io.h"
 
+#include <algorithm>
+#include <cstring>
 #include <limits>
 #include <utility>
 
+#include "iceberg/metadata_cache.h"
+#include "iceberg/snapshot.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg {
@@ -40,6 +44,123 @@ Status FinishWithCloseStatus(Status operation_status, Status close_status) {
   return close_status;
 }
 
+Result<std::string> 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<uint64_t>(read_size) >
+      static_cast<uint64_t>(std::numeric_limits<size_t>::max())) {
+    return Invalid("File size {} exceeds size_t max for {}", read_size, file_location);
+  }
+
+  auto size = static_cast<size_t>(read_size);
+  std::string content(size, '\0');
+  ICEBERG_ASSIGN_OR_RAISE(auto stream, input_file.Open());
+  Status read_status = {};
+  if (size > 0) {
+    auto bytes = std::as_writable_bytes(std::span(content.data(), content.size()));
+    read_status = stream->ReadFully(/*position=*/0, bytes);
+  }
+  ICEBERG_RETURN_UNEXPECTED(
+      FinishWithCloseStatus(std::move(read_status), stream->Close()));
+  return content;
+}
+
+class CachedSeekableInputStream : public SeekableInputStream {
+ public:
+  explicit CachedSeekableInputStream(std::shared_ptr<const std::string> content)
+      : content_(std::move(content)) {}
+
+  Result<int64_t> 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<uint64_t>(position) <= content_->size(),
+                     "Position {} exceeds file size {}", position, content_->size());
+    position_ = position;
+    return {};
+  }
+
+  Result<int64_t> Read(std::span<std::byte> out) override {
+    ICEBERG_PRECHECK(!closed_, "Input stream is closed");
+    auto position = static_cast<size_t>(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<int64_t>(bytes_to_read);
+    }
+    return static_cast<int64_t>(bytes_to_read);
+  }
+
+  Status ReadFully(int64_t position, std::span<std::byte> out) override {
+    ICEBERG_PRECHECK(!closed_, "Input stream is closed");
+    ICEBERG_PRECHECK(position >= 0, "Position must not be negative: {}", position);
+    ICEBERG_PRECHECK(static_cast<uint64_t>(position) <= content_->size(),
+                     "Position {} exceeds file size {}", position, content_->size());
+    auto offset = static_cast<size_t>(position);
+    ICEBERG_PRECHECK(out.size() <= content_->size() - offset,
+                     "Read out of bounds: offset {} + length {} exceeds file size {}",
+                     position, out.size(), content_->size());
+    if (!out.empty()) {
+      std::memcpy(out.data(), content_->data() + offset, out.size());
+    }
+    return {};
+  }
+
+  Status Close() override {
+    closed_ = true;
+    return {};
+  }
+
+ private:
+  std::shared_ptr<const std::string> content_;
+  int64_t position_ = 0;
+  bool closed_ = false;
+};
+
+class CachedInputFile : public InputFile {
+ public:
+  CachedInputFile(std::unique_ptr<InputFile> input_file,
+                  std::shared_ptr<MetadataCache> cache, int64_t size)
+      : input_file_(std::move(input_file)),
+        cache_(std::move(cache)),
+        location_(input_file_->location()),
+        size_(size) {}
+
+  std::string_view location() const override { return location_; }
+
+  Result<int64_t> Size() const override {
+    if (auto content = cache_->GetIfPresent(location_)) {
+      return static_cast<int64_t>(content->size());
+    }
+    return size_;
+  }
+
+  Result<std::unique_ptr<SeekableInputStream>> Open() override {
+    auto content =
+        cache_->Get(location_, static_cast<size_t>(size_),
+                    [this]() -> Result<MetadataCache::Content> {
+                      ICEBERG_ASSIGN_OR_RAISE(
+                          auto loaded, ReadInputFile(*input_file_, size_, location_));
+                      return std::make_shared<const std::string>(std::move(loaded));
+                    });
+    if (!content.has_value()) {
+      // Cache loading is an optimization. Match Java's ContentCache by falling back to
+      // the underlying input when a read-ahead attempt fails.
+      return input_file_->Open();
+    }
+    return std::make_unique<CachedSeekableInputStream>(std::move(content).value());
+  }
+
+ private:
+  std::unique_ptr<InputFile> input_file_;
+  std::shared_ptr<MetadataCache> cache_;
+  std::string location_;
+  int64_t size_;
+};
+
 }  // namespace
 
 Result<std::unique_ptr<InputFile>> FileIO::NewInputFile(std::string file_location) {
@@ -69,21 +190,107 @@ Result<std::string> FileIO::ReadFile(const std::string& file_location,
     ICEBERG_ASSIGN_OR_RAISE(input_file, NewInputFile(file_location));
     ICEBERG_ASSIGN_OR_RAISE(read_size, input_file->Size());
   }
-  if (read_size < 0) {
-    return Invalid("Invalid negative file size {} for {}", read_size, file_location);
+  return ReadInputFile(*input_file, read_size, file_location);
+}
+
+Result<std::unique_ptr<InputFile>> FileIO::NewCachedInputFile(
+    std::string file_location, std::optional<size_t> length) {
+  std::unique_ptr<InputFile> input_file;
+  if (length.has_value()) {
+    ICEBERG_ASSIGN_OR_RAISE(input_file, NewInputFile(file_location, *length));
+  } else {
+    ICEBERG_ASSIGN_OR_RAISE(input_file, NewInputFile(file_location));
   }
 
-  auto size = static_cast<size_t>(read_size);
-  std::string content(size, '\0');
-  ICEBERG_ASSIGN_OR_RAISE(auto stream, input_file->Open());
-  Status read_status = {};
-  if (size > 0) {
-    auto bytes = std::as_writable_bytes(std::span(content.data(), content.size()));
-    read_status = stream->ReadFully(/*position=*/0, bytes);
+  auto cache = GetMetadataCache();
+  if (cache == nullptr || !cache->options().enabled) {
+    return input_file;
   }
-  ICEBERG_RETURN_UNEXPECTED(
-      FinishWithCloseStatus(std::move(read_status), stream->Close()));
-  return content;
+
+  if (auto cached = cache->GetIfPresent(file_location)) {
+    return std::make_unique<CachedInputFile>(std::move(input_file), std::move(cache),
+                                             static_cast<int64_t>(cached->size()));
+  }
+
+  int64_t size;
+  if (length.has_value()) {
+    if (*length > static_cast<size_t>(std::numeric_limits<int64_t>::max())) {
+      return InvalidArgument("File length {} exceeds int64_t max", *length);
+    }
+    size = static_cast<int64_t>(*length);
+  } else {
+    ICEBERG_ASSIGN_OR_RAISE(size, input_file->Size());
+  }
+  if (size < 0) {
+    return Invalid("Invalid negative file size {} for {}", size, file_location);
+  }
+  if (std::cmp_greater(size, cache->options().max_content_length)) {
+    return input_file;
+  }
+  return std::make_unique<CachedInputFile>(std::move(input_file), std::move(cache), size);
+}
+
+Result<std::string> FileIO::ReadFileCached(const std::string& file_location,
+                                           std::optional<size_t> length) {
+  auto cache = GetMetadataCache();
+  if (cache == nullptr || !cache->options().enabled) {
+    return ReadFile(file_location, length);
+  }
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto content,
+      cache->Get(file_location, length,
+                 [this, &file_location, length]() -> Result<MetadataCache::Content> {
+                   ICEBERG_ASSIGN_OR_RAISE(auto loaded, ReadFile(file_location, length));
+                   return std::make_shared<const std::string>(std::move(loaded));
+                 }));
+  return *content;
+}
+
+Status FileIO::ConfigureMetadataCache(
+    const std::unordered_map<std::string, std::string>& properties) {
+  ICEBERG_ASSIGN_OR_RAISE(auto options, MetadataCacheOptions::FromProperties(properties));
+  ICEBERG_ASSIGN_OR_RAISE(auto cache, MetadataCache::Make(options));
+  std::lock_guard lock(metadata_cache_state_->mutex);
+  if (metadata_cache_state_->cache == nullptr) {
+    metadata_cache_state_->cache = std::move(cache);
+    return {};
+  }
+  if (metadata_cache_state_->cache->options() == options) {
+    return {};
+  }
+  return InvalidArgument("Metadata cache is already configured with different options");
+}
+
+bool FileIO::MetadataCacheEnabled() const {
+  auto cache = GetMetadataCache();
+  return cache != nullptr && cache->options().enabled;
+}
+
+void FileIO::InvalidateMetadataCache(std::string_view file_location) {
+  auto cache = GetMetadataCache();
+  if (cache != nullptr) {
+    cache->Invalidate(file_location);
+  }
+}
+
+void FileIO::ClearMetadataCache() {
+  auto cache = GetMetadataCache();
+  if (cache != nullptr) {
+    cache->Clear();
+  }
+}
+
+std::shared_ptr<MetadataCache> FileIO::GetMetadataCache() const {
+  std::lock_guard lock(metadata_cache_state_->mutex);
+  return metadata_cache_state_->cache;
+}
+
+std::shared_ptr<internal::SnapshotCacheData> FileIO::GetSnapshotCacheData() const {
+  std::lock_guard lock(metadata_cache_state_->mutex);
+  if (metadata_cache_state_->snapshot_cache == nullptr) {
+    metadata_cache_state_->snapshot_cache = internal::MakeSnapshotCacheData();
+  }
+  return metadata_cache_state_->snapshot_cache;
 }
 
 Status FileIO::WriteFile(const std::string& file_location, std::string_view content) {