feat(inspect): add shared snapshot metadata scan support

src/iceberg/CMakeLists.txt

@@ -63,6 +63,7 @@ set(ICEBERG_SOURCES
     inspect/metadata_table.cc
     inspect/metadata_scan_task.cc
     inspect/metadata_scan_task_reader.cc
+    inspect/metadata_table_util_internal.cc
     inspect/refs_table.cc
     inspect/snapshots_table.cc
     inheritable_metadata.cc

src/iceberg/inspect/history_table.cc

@@ -77,7 +77,7 @@ HistoryTable::HistoryTable(std::shared_ptr<const Table> table)
 
 HistoryTable::~HistoryTable() = default;
 
-const std::shared_ptr<Schema>& HistoryTable::schema() const { return CanonicalSchema(); }
+std::shared_ptr<Schema> HistoryTable::schema() const { return CanonicalSchema(); }
 
 const std::shared_ptr<Schema>& HistoryTable::CanonicalSchema() {
   static const auto schema = std::make_shared<Schema>(std::vector<SchemaField>{

src/iceberg/inspect/history_table.h

@@ -46,7 +46,7 @@ class ICEBERG_EXPORT HistoryTable : public MetadataTable {
 
   MetadataTableKind kind() const noexcept override { return MetadataTableKind::kHistory; }
 
-  const std::shared_ptr<Schema>& schema() const override;
+  std::shared_ptr<Schema> schema() const override;
 
  protected:
   Result<std::unique_ptr<MetadataScan>> MakeScan(

src/iceberg/inspect/metadata_scan.cc

@@ -107,20 +107,26 @@ class TaskArrowStreamSource {
 
 }  // namespace
 
-MetadataScanBuilder::MetadataScanBuilder(const MetadataTable& table)
-    : table_(&table),
-      context_{.kind = table.kind(),
-               .table_metadata = table.source_table()->metadata(),
-               .schema = table.schema(),
-               .projected_schema = table.schema()} {}
+MetadataScanBuilder::MetadataScanBuilder(const MetadataTable& table,
+                                         internal::MetadataScanContext context)
+    : table_(&table), context_(std::move(context)) {}
 
 MetadataScanBuilder::~MetadataScanBuilder() = default;
 
 Result<std::unique_ptr<MetadataScanBuilder>> MetadataScanBuilder::Make(
     const MetadataTable& table) {
   ICEBERG_PRECHECK(table.source_table() != nullptr, "Source table cannot be null");
-  ICEBERG_PRECHECK(table.schema() != nullptr, "Metadata table schema cannot be null");
-  return std::unique_ptr<MetadataScanBuilder>(new MetadataScanBuilder(table));
+  auto metadata = table.source_table()->metadata();
+  ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
+  ICEBERG_ASSIGN_OR_RAISE(auto schema, table.SchemaForScan(*metadata));
+  ICEBERG_PRECHECK(schema != nullptr, "Metadata table schema cannot be null");
+  internal::MetadataScanContext context{.kind = table.kind(),
+                                        .table_metadata = std::move(metadata),
+                                        .schema = schema,
+                                        .projected_schema = std::move(schema),
+                                        .io = table.source_table()->io()};
+  return std::unique_ptr<MetadataScanBuilder>(
+      new MetadataScanBuilder(table, std::move(context)));
 }
 
 MetadataScanBuilder& MetadataScanBuilder::Select(

src/iceberg/inspect/metadata_scan.h

@@ -52,6 +52,7 @@ struct MetadataScanContext {
   std::optional<int64_t> snapshot_id;
   std::string ref_name;
   std::optional<int64_t> as_of_time_in_millis;
+  std::shared_ptr<FileIO> io;
 };
 
 }  // namespace internal
@@ -97,7 +98,7 @@ class ICEBERG_EXPORT MetadataScanBuilder : public ErrorCollector {
   Result<std::unique_ptr<MetadataScan>> Build();
 
  private:
-  explicit MetadataScanBuilder(const MetadataTable& table);
+  MetadataScanBuilder(const MetadataTable& table, internal::MetadataScanContext context);
 
   const MetadataTable* table_;
   internal::MetadataScanContext context_;

src/iceberg/inspect/metadata_scan_task.h

@@ -37,6 +37,8 @@ class ICEBERG_EXPORT MetadataScanTask {
   enum class TaskKind {
     /// StaticMetadataTask reads in-memory table metadata (snapshots, history, refs).
     kStatic,
+    /// SnapshotMetadataTask reads snapshot manifests and their file entries.
+    kSnapshot,
   };
 
   virtual ~MetadataScanTask() = default;

src/iceberg/inspect/metadata_scan_task_reader.cc

@@ -25,6 +25,7 @@
 #include "iceberg/inspect/history_table.h"
 #include "iceberg/inspect/metadata_scan_task.h"
 #include "iceberg/inspect/metadata_table.h"
+#include "iceberg/inspect/metadata_table_util_internal.h"
 #include "iceberg/inspect/refs_table.h"
 #include "iceberg/inspect/snapshots_table.h"
 #include "iceberg/util/checked_cast.h"
@@ -47,6 +48,11 @@ Result<std::unique_ptr<MetadataScanTaskReader>> MetadataScanTaskReader::Make(
 Result<ArrowArrayStream> MetadataScanTaskReader::Open(
     const MetadataScanTask& task) const {
   switch (task.task_kind()) {
+    case MetadataScanTask::TaskKind::kSnapshot: {
+      const auto& snapshot_task =
+          internal::checked_cast<const internal::SnapshotMetadataTask&>(task);
+      return snapshot_task.Open(options_.batch_size);
+    }
     case MetadataScanTask::TaskKind::kStatic: {
       const auto& static_task = internal::checked_cast<const StaticMetadataTask&>(task);
       switch (static_task.table_kind()) {

src/iceberg/inspect/metadata_scan_util_internal.h

@@ -40,6 +40,7 @@
 #include "iceberg/schema_internal.h"
 #include "iceberg/type.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/stream.h"
 
 namespace iceberg::internal {
 
@@ -181,4 +182,79 @@ Result<ArrowArrayStream> MakeMetadataBatchStream(
                                batch_size, std::forward<GetRow>(get_row)));
 }
 
+/// Stream manifest-backed rows with the same batch-size and ownership contract.
+template <typename Row, typename AppendRow>
+class MetadataRowsSource : public Stream<ArrowArray> {
+ public:
+  MetadataRowsSource(std::shared_ptr<iceberg::Schema> schema,
+                     std::unique_ptr<Stream<Row>> rows, int64_t batch_size,
+                     AppendRow append_row)
+      : schema_(std::move(schema)),
+        rows_(std::move(rows)),
+        batch_size_(batch_size),
+        append_row_(std::move(append_row)) {}
+
+  Status Close() {
+    rows_.reset();
+    return {};
+  }
+
+  Result<ArrowSchema> Schema() {
+    ArrowSchema schema{};
+    ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*schema_, &schema));
+    return schema;
+  }
+
+ protected:
+  Result<std::optional<ArrowArray>> NextImpl() override {
+    ICEBERG_PRECHECK(rows_ != nullptr, "Cannot read from a closed metadata stream");
+    ICEBERG_ASSIGN_OR_RAISE(auto row, rows_->Next());
+    if (!row) {
+      return std::nullopt;
+    }
+    ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(*schema_));
+    ICEBERG_RETURN_UNEXPECTED(append_row_(builder, *row));
+    while (builder.num_rows() < batch_size_) {
+      ICEBERG_ASSIGN_OR_RAISE(auto next_row, rows_->Next());
+      if (!next_row) {
+        break;
+      }
+      ICEBERG_RETURN_UNEXPECTED(append_row_(builder, *next_row));
+    }
+    ICEBERG_ASSIGN_OR_RAISE(auto batch, std::move(builder).Finish());
+    return batch;
+  }
+
+ private:
+  const std::shared_ptr<iceberg::Schema> schema_;
+  std::unique_ptr<Stream<Row>> rows_;
+  const int64_t batch_size_;
+  AppendRow append_row_;
+};
+
+template <typename Row, typename AppendRow>
+Result<ArrowArrayStream> MakeMetadataBatchStream(const std::shared_ptr<Schema>& schema,
+                                                 std::unique_ptr<Stream<Row>> rows,
+                                                 int64_t batch_size,
+                                                 AppendRow append_row) {
+  ICEBERG_PRECHECK(batch_size > 0, "Batch size must be positive");
+  ICEBERG_PRECHECK(schema != nullptr, "Metadata stream schema cannot be null");
+  ICEBERG_PRECHECK(rows != nullptr, "Metadata row source cannot be null");
+  if constexpr (requires { static_cast<bool>(append_row); }) {
+    ICEBERG_PRECHECK(static_cast<bool>(append_row),
+                     "Metadata row callback cannot be empty");
+  }
+  return MakeArrowArrayStream(std::make_unique<MetadataRowsSource<Row, AppendRow>>(
+      schema, std::move(rows), batch_size, std::move(append_row)));
+}
+
+template <typename Row, typename AppendRow>
+Result<ArrowArrayStream> MakeMetadataBatchStream(const std::shared_ptr<Schema>& schema,
+                                                 std::vector<Row> rows,
+                                                 int64_t batch_size,
+                                                 AppendRow append_row) {
+  return MakeMetadataBatchStream(schema, Stream<Row>::FromVector(std::move(rows)),
+                                 batch_size, std::move(append_row));
+}
+
 }  // namespace iceberg::internal

src/iceberg/inspect/metadata_table.cc

@@ -53,6 +53,11 @@ const std::shared_ptr<const Table>& MetadataTable::source_table() const {
   return source_table_;
 }
 
+Result<std::shared_ptr<Schema>> MetadataTable::SchemaForScan(
+    const TableMetadata& metadata) const {
+  return schema();
+}
+
 Result<std::unique_ptr<MetadataScanBuilder>> MetadataTable::NewScan() const {
   return MetadataScanBuilder::Make(*this);
 }

src/iceberg/inspect/metadata_table.h

@@ -78,8 +78,8 @@ class ICEBERG_EXPORT MetadataTable {
   /// \brief Return this metadata table's kind.
   virtual Kind kind() const noexcept = 0;
 
-  /// \brief Return the complete canonical schema of this metadata table.
-  virtual const std::shared_ptr<Schema>& schema() const = 0;
+  /// \brief Return shared ownership of this metadata table's canonical schema.
+  virtual std::shared_ptr<Schema> schema() const = 0;
 
   /// \brief Return the source table whose metadata is exposed.
   const std::shared_ptr<const Table>& source_table() const;
@@ -93,6 +93,10 @@ class ICEBERG_EXPORT MetadataTable {
  protected:
   explicit MetadataTable(std::shared_ptr<const Table> source_table);
 
+  /// \brief Derive a scan schema from the captured source metadata.
+  virtual Result<std::shared_ptr<Schema>> SchemaForScan(
+      const TableMetadata& metadata) const;
+
   /// \brief Plan a scan using the captured metadata, projection, and snapshot selection.
   virtual Result<std::unique_ptr<MetadataScan>> MakeScan(
       internal::MetadataScanContext context) const = 0;