feat(inspect): implement SnapshotsTable scanning

- Add Scan() virtual method and Scan() convenience overload to MetadataTable
- Add SnapshotSelection struct for time-travel snapshot resolution
- Add supports_time_travel() concrete method driven by kind()
- Implement SnapshotsTable::Scan() to materialize snapshot rows via ArrowRowBuilder

Author: WZhuo

src/iceberg/inspect/metadata_table.cc

@@ -35,6 +35,23 @@ MetadataTable::MetadataTable(std::shared_ptr<Table> source_table,
 
 MetadataTable::~MetadataTable() = default;
 
+bool MetadataTable::supports_time_travel() const noexcept {
+  // Time travel is supported for tables that read from a single snapshot's
+  // manifests. Tables that scan all snapshots or return in-memory history do
+  // not.
+  switch (kind()) {
+    case Kind::kSnapshots:
+    case Kind::kHistory:
+      return false;
+  }
+  return false;
+}
+
+Result<ArrowArray> MetadataTable::Scan(
+    std::optional<SnapshotSelection> /*snapshot_selection*/) {
+  return NotSupported("Scan is not supported for this metadata table type");
+}
+
 Result<std::unique_ptr<MetadataTable>> MetadataTable::Make(std::shared_ptr<Table> table,
                                                            Kind kind) {
   if (table == nullptr) [[unlikely]] {

src/iceberg/inspect/metadata_table.h

@@ -20,14 +20,28 @@
 #pragma once
 
 #include <memory>
+#include <optional>
+#include <string>
 
+#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 Parameters for snapshot selection (time travel).
+struct SnapshotSelection {
+  /// \brief The snapshot ID to read.
+  std::optional<int64_t> snapshot_id;
+  /// \brief Read the snapshot that was current at this timestamp.
+  std::optional<TimePointMs> as_of_timestamp;
+  /// \brief Read the snapshot referenced by this named ref (branch or tag).
+  std::optional<std::string> ref_name;
+};
+
 /// \brief Base class for Iceberg metadata tables.
 class ICEBERG_EXPORT MetadataTable {
  public:
@@ -43,6 +57,28 @@ class ICEBERG_EXPORT MetadataTable {
 
   virtual Kind kind() const noexcept = 0;
 
+  /// \brief Whether this metadata table supports time-travel queries.
+  ///
+  /// Time travel is supported for tables that read from a single snapshot's
+  /// manifests (e.g., Entries, Files, Manifests, Partitions). Tables that
+  /// scan all snapshots (All*) or return in-memory history (Snapshots,
+  /// History, Refs) do not support time travel.
+  bool supports_time_travel() const noexcept;
+
+  /// \brief Scan the metadata table using the current snapshot.
+  ///
+  /// Convenience overload — delegates to Scan(std::nullopt).
+  Result<ArrowArray> Scan() { return Scan(std::nullopt); }
+
+  /// \brief Scan the metadata table and return all rows as an Arrow struct array.
+  ///
+  /// The returned ArrowArray is a struct array where each element is one row.
+  /// The caller takes ownership and must call ArrowArrayRelease when done.
+  ///
+  /// The default implementation returns NotSupported. Subclasses override this
+  /// to materialize their data.
+  virtual Result<ArrowArray> Scan(std::optional<SnapshotSelection> snapshot_selection);
+
   const TableIdentifier& name() const { return identifier_; }
 
   const std::shared_ptr<Schema>& schema() const { return schema_; }

src/iceberg/inspect/snapshots_table.cc

@@ -19,15 +19,18 @@
 
 #include "iceberg/inspect/snapshots_table.h"
 
+#include <chrono>
 #include <memory>
 #include <utility>
 #include <vector>
 
+#include "iceberg/arrow_row_builder_internal.h"
 #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 {
@@ -65,4 +68,46 @@ Result<std::unique_ptr<SnapshotsTable>> SnapshotsTable::Make(
   return std::unique_ptr<SnapshotsTable>(new SnapshotsTable(std::move(table)));
 }
 
+Result<ArrowArray> SnapshotsTable::Scan(
+    std::optional<SnapshotSelection> /*snapshot_selection*/) {
+  ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(*schema()));
+
+  for (const auto& snapshot : source_table()->snapshots()) {
+    // column 0: committed_at (timestamptz → int64 micros)
+    ICEBERG_RETURN_UNEXPECTED(AppendInt(
+        builder.column(0), std::chrono::duration_cast<std::chrono::microseconds>(
+                               snapshot->timestamp_ms.time_since_epoch())
+                               .count()));
+
+    // column 1: snapshot_id (long)
+    ICEBERG_RETURN_UNEXPECTED(AppendInt(builder.column(1), snapshot->snapshot_id));
+
+    // column 2: parent_id (long, optional)
+    if (snapshot->parent_snapshot_id.has_value()) {
+      ICEBERG_RETURN_UNEXPECTED(
+          AppendInt(builder.column(2), *snapshot->parent_snapshot_id));
+    } else {
+      ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(2)));
+    }
+
+    // column 3: operation (string, optional)
+    auto op = snapshot->Operation();
+    if (op.has_value()) {
+      ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(3), *op));
+    } else {
+      ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(3)));
+    }
+
+    // column 4: manifest_list (string, optional)
+    ICEBERG_RETURN_UNEXPECTED(AppendString(builder.column(4), snapshot->manifest_list));
+
+    // column 5: summary (map<string,string>)
+    ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), snapshot->summary));
+
+    ICEBERG_RETURN_UNEXPECTED(builder.FinishRow());
+  }
+
+  return std::move(builder).Finish();
+}
+
 }  // namespace iceberg

src/iceberg/inspect/snapshots_table.h

@@ -37,6 +37,13 @@ class ICEBERG_EXPORT SnapshotsTable : public MetadataTable {
 
   Kind kind() const noexcept override { return Kind::kSnapshots; }
 
+  /// \brief Scan all snapshots as rows.
+  ///
+  /// The snapshots table always returns every known snapshot, so the
+  /// snapshot_selection parameter is ignored.
+  Result<ArrowArray> Scan(
+      std::optional<SnapshotSelection> /*snapshot_selection*/) override;
+
  private:
   explicit SnapshotsTable(std::shared_ptr<Table> table);
 };

src/iceberg/test/CMakeLists.txt

@@ -188,7 +188,12 @@ if(ICEBERG_BUILD_BUNDLE)
 
   add_iceberg_test(catalog_test USE_BUNDLE SOURCES in_memory_catalog_test.cc)
 
-  add_iceberg_test(metadata_table_test USE_BUNDLE SOURCES metadata_table_test.cc)
+  add_iceberg_test(metadata_table_test
+                   USE_BUNDLE
+                   SOURCES
+                   history_table_test.cc
+                   metadata_table_test.cc
+                   snapshots_table_test.cc)
 
   add_iceberg_test(eval_expr_test
                    USE_BUNDLE

src/iceberg/test/history_table_test.cc

@@ -0,0 +1,54 @@
+/*
+ * 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.
+ */
+
+/// \file history_table_test.cc
+/// Unit tests for HistoryTable.
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include "iceberg/inspect/metadata_table.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/test/matchers.h"
+#include "iceberg/test/metadata_table_test_base.h"
+#include "iceberg/type.h"
+
+namespace iceberg {
+namespace {
+
+std::shared_ptr<Schema> MakeHistorySchema() {
+  return 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())});
+}
+
+}  // namespace
+
+class HistoryTableTest : public MetadataTableTestBase {};
+
+TEST_F(HistoryTableTest, SchemaMatchesIcebergSchema) {
+  ICEBERG_UNWRAP_OR_FAIL(auto history_table,
+                         MetadataTable::Make(table_, MetadataTable::Kind::kHistory));
+  EXPECT_TRUE(*history_table->schema() == *MakeHistorySchema());
+}
+
+}  // namespace iceberg

src/iceberg/test/metadata_table_test.cc

@@ -33,88 +33,40 @@
 #include "iceberg/type.h"
 
 namespace iceberg {
-namespace {
-
-std::shared_ptr<Schema> MakeSnapshotsSchema() {
-  return std::make_shared<Schema>(std::vector<SchemaField>{
-      SchemaField::MakeRequired(1, "committed_at", timestamp_tz()),
-      SchemaField::MakeRequired(2, "snapshot_id", int64()),
-      SchemaField::MakeOptional(3, "parent_id", int64()),
-      SchemaField::MakeOptional(4, "operation", string()),
-      SchemaField::MakeOptional(5, "manifest_list", string()),
-      SchemaField::MakeOptional(
-          6, "summary",
-          std::make_shared<MapType>(SchemaField::MakeRequired(7, "key", string()),
-                                    SchemaField::MakeRequired(8, "value", string())))});
-}
-
-std::shared_ptr<Schema> MakeHistorySchema() {
-  return 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())});
-}
-
-}  // namespace
 
 class MetadataTableTest : public ::testing::Test {
  protected:
   void SetUp() override {
-    io_ = std::make_shared<MockFileIO>();
-    catalog_ = std::make_shared<MockCatalog>();
-
     auto schema = std::make_shared<Schema>(
         std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int64()),
                                  SchemaField::MakeOptional(2, "name", string())},
         1);
-    metadata_ = std::make_shared<TableMetadata>(
+    auto metadata = std::make_shared<TableMetadata>(
         TableMetadata{.format_version = 2, .schemas = {schema}, .current_schema_id = 1});
 
-    TableIdentifier source_ident{.ns = Namespace{.levels = {"db"}},
-                                 .name = "source_table"};
-    auto source_table_result =
-        Table::Make(source_ident, metadata_, "s3://bucket/meta.json", io_, catalog_);
-    EXPECT_THAT(source_table_result, IsOk());
-    source_table_ = *source_table_result;
-
-    auto snapshots_table_result =
-        MetadataTable::Make(source_table_, MetadataTable::Kind::kSnapshots);
-    EXPECT_THAT(snapshots_table_result, IsOk());
-    snapshots_table_ = std::move(*snapshots_table_result);
+    TableIdentifier ident{.ns = Namespace{.levels = {"db"}}, .name = "source_table"};
+    ICEBERG_UNWRAP_OR_FAIL(table_, Table::Make(ident, metadata, "s3://bucket/meta.json",
+                                               std::make_shared<MockFileIO>(),
+                                               std::make_shared<MockCatalog>()));
   }
 
-  std::shared_ptr<MockFileIO> io_;
-  std::shared_ptr<MockCatalog> catalog_;
-  std::shared_ptr<TableMetadata> metadata_;
-  std::shared_ptr<Table> source_table_;
-  std::unique_ptr<MetadataTable> snapshots_table_;
+  std::shared_ptr<Table> table_;
 };
 
-TEST_F(MetadataTableTest, Constructor) {
-  EXPECT_EQ(snapshots_table_->kind(), MetadataTable::Kind::kSnapshots);
-  EXPECT_EQ(snapshots_table_->source_table(), source_table_);
-  EXPECT_EQ(snapshots_table_->name().name, "source_table.snapshots");
-  EXPECT_EQ(snapshots_table_->name().ns.levels, (std::vector<std::string>{"db"}));
-  EXPECT_NE(snapshots_table_->schema(), nullptr);
-}
-
-TEST_F(MetadataTableTest, SnapshotsSchemaMatchesIcebergSchema) {
-  EXPECT_TRUE(*snapshots_table_->schema() == *MakeSnapshotsSchema());
-}
-
-TEST_F(MetadataTableTest, HistorySchemaMatchesIcebergSchema) {
-  auto history_table_result =
-      MetadataTable::Make(source_table_, MetadataTable::Kind::kHistory);
-  ASSERT_THAT(history_table_result, IsOk());
-
-  EXPECT_TRUE(*(*history_table_result)->schema() == *MakeHistorySchema());
-}
-
 TEST_F(MetadataTableTest, FactoryRejectsNullSourceTable) {
   auto result = MetadataTable::Make(nullptr, MetadataTable::Kind::kSnapshots);
   EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument));
   EXPECT_THAT(result, HasErrorMessage("Table cannot be null"));
 }
 
+TEST_F(MetadataTableTest, SupportsTimeTravel) {
+  ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
+                         MetadataTable::Make(table_, MetadataTable::Kind::kSnapshots));
+  EXPECT_FALSE(snapshots_table->supports_time_travel());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto history_table,
+                         MetadataTable::Make(table_, MetadataTable::Kind::kHistory));
+  EXPECT_FALSE(history_table->supports_time_travel());
+}
+
 }  // namespace iceberg