feat(inspect): implement streaming SnapshotsTable scans

src/iceberg/arrow_row_builder.cc

@@ -75,6 +75,8 @@ ArrowRowBuilder::~ArrowRowBuilder() {
 
 int64_t ArrowRowBuilder::num_columns() const { return array_.n_children; }
 
+int64_t ArrowRowBuilder::num_rows() const { return array_.length; }
+
 ArrowArray* ArrowRowBuilder::column(int64_t index) {
   if (index < 0 || index >= array_.n_children) {
     return nullptr;

src/iceberg/arrow_row_builder_internal.h

@@ -85,6 +85,9 @@ class ICEBERG_EXPORT ArrowRowBuilder {
   /// \brief The number of top-level columns in the batch.
   int64_t num_columns() const;
 
+  /// \brief The number of completed rows in the batch.
+  int64_t num_rows() const;
+
   /// \brief Access the nanoarrow child builder for a top-level column.
   ///
   /// \param index Zero-based column index. Returns nullptr if out of range.

src/iceberg/inspect/history_table.cc

@@ -26,37 +26,32 @@
 #include "iceberg/schema.h"
 #include "iceberg/schema_field.h"
 #include "iceberg/table.h"
-#include "iceberg/table_identifier.h"
 #include "iceberg/type.h"
+#include "iceberg/util/macros.h"
 
 namespace iceberg {
-namespace {
 
-std::shared_ptr<Schema> MakeHistoryTableSchema() {
-  return std::make_shared<Schema>(std::vector<SchemaField>{
+HistoryTable::HistoryTable(std::shared_ptr<Table> table)
+    : MetadataTable(std::move(table)) {}
+
+HistoryTable::~HistoryTable() = default;
+
+const std::shared_ptr<Schema>& HistoryTable::schema() const {
+  static const auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
       SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()),
       SchemaField::MakeRequired(2, "snapshot_id", int64()),
       SchemaField::MakeOptional(3, "parent_id", int64()),
       SchemaField::MakeRequired(4, "is_current_ancestor", boolean())});
+  return schema;
 }
 
-TableIdentifier MakeHistoryTableName(const TableIdentifier& source_name) {
-  return TableIdentifier{.ns = source_name.ns, .name = source_name.name + ".history"};
-}
-
-}  // namespace
-
-HistoryTable::HistoryTable(std::shared_ptr<Table> table)
-    : MetadataTable(table, MakeHistoryTableName(table->name()),
-                    MakeHistoryTableSchema()) {}
-
-HistoryTable::~HistoryTable() = default;
-
 Result<std::unique_ptr<HistoryTable>> HistoryTable::Make(std::shared_ptr<Table> table) {
-  if (table == nullptr) [[unlikely]] {
-    return InvalidArgument("Table cannot be null");
-  }
+  ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
   return std::unique_ptr<HistoryTable>(new HistoryTable(std::move(table)));
 }
 
+Result<ArrowArrayStream> HistoryTable::Scan() {
+  return NotSupported("Scan is not supported for the history table");
+}
+
 }  // namespace iceberg

src/iceberg/inspect/history_table.h

@@ -40,6 +40,10 @@ class ICEBERG_EXPORT HistoryTable : public MetadataTable {
 
   Kind kind() const noexcept override { return Kind::kHistory; }
 
+  const std::shared_ptr<Schema>& schema() const override;
+
+  Result<ArrowArrayStream> Scan() override;
+
  private:
   explicit HistoryTable(std::shared_ptr<Table> table);
 };

src/iceberg/inspect/metadata_table.cc

@@ -22,33 +22,33 @@
 #include <memory>
 #include <utility>
 
-#include "iceberg/inspect/history_table.h"
-#include "iceberg/inspect/snapshots_table.h"
-
 namespace iceberg {
 
-MetadataTable::MetadataTable(std::shared_ptr<Table> source_table,
-                             TableIdentifier identifier, std::shared_ptr<Schema> schema)
-    : identifier_(std::move(identifier)),
-      schema_(std::move(schema)),
-      source_table_(std::move(source_table)) {}
+MetadataTable::MetadataTable(std::shared_ptr<Table> source_table)
+    : source_table_(std::move(source_table)) {}
 
 MetadataTable::~MetadataTable() = default;
 
-Result<std::unique_ptr<MetadataTable>> MetadataTable::Make(std::shared_ptr<Table> table,
-                                                           Kind kind) {
-  if (table == nullptr) [[unlikely]] {
-    return InvalidArgument("Table cannot be null");
-  }
+bool MetadataTable::supports_time_travel() const noexcept { return false; }
+
+const std::shared_ptr<Table>& MetadataTable::source_table() const {
+  return source_table_;
+}
+
+TimeTravelMetadataTable::TimeTravelMetadataTable(std::shared_ptr<Table> source_table)
+    : MetadataTable(std::move(source_table)) {}
 
-  switch (kind) {
-    case Kind::kSnapshots:
-      return SnapshotsTable::Make(table);
-    case Kind::kHistory:
-      return HistoryTable::Make(table);
-  }
+TimeTravelMetadataTable::~TimeTravelMetadataTable() = default;
+
+bool TimeTravelMetadataTable::supports_time_travel() const noexcept { return true; }
+
+Result<ArrowArrayStream> TimeTravelMetadataTable::Scan() {
+  return ScanSnapshot(SnapshotSelection{});
+}
 
-  return NotSupported("Unsupported metadata table type");
+Result<ArrowArrayStream> TimeTravelMetadataTable::Scan(
+    const SnapshotSelection& snapshot_selection) {
+  return ScanSnapshot(snapshot_selection);
 }
 
 }  // namespace iceberg

src/iceberg/inspect/metadata_table.h

@@ -20,46 +20,108 @@
 #pragma once
 
 /// \file iceberg/inspect/metadata_table.h
-/// \brief Define base APIs for metadata tables.
+/// \brief Base APIs for inspecting Iceberg metadata tables.
 
+#include <concepts>
 #include <memory>
+#include <string>
+#include <utility>
+#include <variant>
 
+#include "iceberg/arrow_c_data.h"
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
-#include "iceberg/table_identifier.h"
 #include "iceberg/type_fwd.h"
+#include "iceberg/util/timepoint.h"
 
 namespace iceberg {
 
-/// \brief Base class for Iceberg metadata tables.
+/// \brief Base interface for an Iceberg metadata table.
 class ICEBERG_EXPORT MetadataTable {
  public:
+  /// \brief Supported metadata table kinds.
   enum class Kind {
     kSnapshots,
     kHistory,
   };
 
-  static Result<std::unique_ptr<MetadataTable>> Make(std::shared_ptr<Table> table,
-                                                     Kind kind);
+  /// \brief Maximum number of rows emitted in each Arrow batch.
+  static constexpr int64_t kBatchSize = 1024;
+
+  /// \brief Create a metadata table of the requested concrete type.
+  ///
+  /// \tparam MetadataTableType Concrete class derived from MetadataTable.
+  /// \param table Source table whose metadata will be exposed.
+  /// \return The constructed metadata table, or an error.
+  template <typename MetadataTableType>
+    requires std::derived_from<MetadataTableType, MetadataTable>
+  static Result<std::unique_ptr<MetadataTableType>> Make(std::shared_ptr<Table> table) {
+    return MetadataTableType::Make(std::move(table));
+  }
 
   virtual ~MetadataTable();
 
+  /// \brief Return this metadata table's kind.
   virtual Kind kind() const noexcept = 0;
 
-  const TableIdentifier& name() const { return identifier_; }
+  /// \brief Return the schema of rows emitted by scans.
+  virtual const std::shared_ptr<Schema>& schema() const = 0;
+
+  /// \brief Return the source table whose metadata is exposed.
+  const std::shared_ptr<Table>& source_table() const;
 
-  const std::shared_ptr<Schema>& schema() const { return schema_; }
+  /// \brief Return whether this metadata table supports time travel.
+  virtual bool supports_time_travel() const noexcept;
 
-  const std::shared_ptr<Table>& source_table() const { return source_table_; }
+  /// \brief Scan the metadata table without time travel.
+  ///
+  /// The caller owns the returned stream and must release it with
+  /// ArrowArrayStreamRelease.
+  virtual Result<ArrowArrayStream> Scan() = 0;
 
  protected:
-  explicit MetadataTable(std::shared_ptr<Table> source_table, TableIdentifier identifier,
-                         std::shared_ptr<Schema> schema);
+  explicit MetadataTable(std::shared_ptr<Table> source_table);
 
  private:
-  TableIdentifier identifier_;
-  std::shared_ptr<Schema> schema_;
   std::shared_ptr<Table> source_table_;
 };
 
+/// \brief Snapshot selection parameters for a time-travel scan.
+struct SnapshotSelection {
+  /// \brief Select the current snapshot, a snapshot ID, or an as-of timestamp.
+  ///
+  /// std::monostate selects the current snapshot.
+  std::variant<std::monostate, int64_t, TimePointMs> snapshot;
+
+  /// \brief Resolve the snapshot relative to this branch or tag.
+  ///
+  /// An empty string uses the main branch.
+  std::string ref_name;
+};
+
+/// \brief Base interface for metadata tables that support time travel.
+class ICEBERG_EXPORT TimeTravelMetadataTable : public MetadataTable {
+ public:
+  ~TimeTravelMetadataTable() override;
+
+  /// \brief Return true because this interface supports time travel.
+  bool supports_time_travel() const noexcept final;
+
+  /// \brief Scan using the current snapshot on the main branch.
+  Result<ArrowArrayStream> Scan() final;
+
+  /// \brief Scan using the requested snapshot selection.
+  ///
+  /// \param snapshot_selection Snapshot ID, timestamp, and optional ref selection.
+  /// \return An Arrow stream containing the metadata table rows, or an error.
+  Result<ArrowArrayStream> Scan(const SnapshotSelection& snapshot_selection);
+
+ protected:
+  explicit TimeTravelMetadataTable(std::shared_ptr<Table> source_table);
+
+  /// \brief Implement a scan for the requested snapshot selection.
+  virtual Result<ArrowArrayStream> ScanSnapshot(
+      const SnapshotSelection& snapshot_selection) = 0;
+};
+
 }  // namespace iceberg