feat(inspect): implement streaming SnapshotsTable scans

[WZhuo]

Summary

Implement streaming scans for SnapshotsTable and refine the metadata-table scan APIs so snapshot metadata can be consumed safely in bounded Arrow batches.

Changes

Metadata table APIs

• Add a typed MetadataTable::Make<T>() factory that preserves the concrete metadata-table type.
• Keep the base MetadataTable interface limited to non-time-travel Scan() calls.
• Add TimeTravelMetadataTable as the capability-specific interface for scans using SnapshotSelection.
• Model snapshot ID and as-of timestamp as mutually exclusive values with std::variant<std::monostate, int64_t, TimePointMs>.
• Return ArrowArrayStream from scan APIs and document stream ownership.
• Provide shared metadata-table schema and source-table accessors.

Snapshots table

• Implement SnapshotsTable::Scan() using a stateful Arrow stream.
• Convert the Iceberg schema to ArrowSchema once when creating the stream and release owned resources when the stream closes.
• Emit at most MetadataTable::kBatchSize rows per batch.
• Skip null snapshot entries safely.
• Materialize committed_at, snapshot_id, parent_id, operation, manifest_list, and summary using ArrowRowBuilder.
• Remove SnapshotSummaryFields::kOperation from the summary map and emit null when the remaining summary is empty.

Arrow row builder

• Add ArrowRowBuilder::num_rows() so streaming producers can enforce batch-size limits without maintaining duplicate row counters.

Tests

• Add shared metadata-table fixtures and Arrow stream import helpers.
• Cover snapshots schema and row values, summary filtering, null snapshots, empty summaries, empty tables, and multi-batch scans.
• Cover metadata-table construction, time-travel capability reporting, and history-table schema behavior.

Testing

cmake --build build --target metadata_table_test -j2
build/src/iceberg/test/metadata_table_test

[Copilot]

Pull request overview

Implements initial metadata-table scanning support by adding a Scan() API to MetadataTable (with snapshot-selection parameters for future time-travel) and providing a concrete SnapshotsTable::Scan() implementation that materializes snapshot rows into Arrow arrays. The PR also restructures/extends the metadata-table test suite to validate schemas and snapshot scanning behavior.

Changes:

• Added SnapshotSelection and a virtual MetadataTable::Scan() API (with a convenience overload) plus supports_time_travel().
• Implemented SnapshotsTable::Scan() to emit snapshot rows (6 columns) via ArrowRowBuilder.
• Added/expanded tests and wired new test sources into the metadata-table test target.

Reviewed changes

Copilot reviewed 9 out of 9 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
src/iceberg/inspect/metadata_table.h	Adds SnapshotSelection, Scan() API, and supports_time_travel() declaration/docs.
src/iceberg/inspect/metadata_table.cc	Implements default Scan() + supports_time_travel() and wires factory for kinds.
src/iceberg/inspect/snapshots_table.h	Declares SnapshotsTable::Scan() override.
src/iceberg/inspect/snapshots_table.cc	Implements snapshot scanning into Arrow via ArrowRowBuilder.
src/iceberg/test/metadata_table_test.cc	Simplifies base setup and adds SupportsTimeTravel test.
src/iceberg/test/metadata_table_test_base.h	New shared fixture/helpers for metadata table tests.
src/iceberg/test/snapshots_table_test.cc	New tests validating snapshots table construction/schema/scan output.
src/iceberg/test/history_table_test.cc	New schema test for history table.
src/iceberg/test/CMakeLists.txt	Adds new test sources to the metadata-table test target.

[Copilot]

Pull request overview

Copilot reviewed 9 out of 9 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (1)

src/iceberg/inspect/metadata_table.cc:38

• supports_time_travel() is currently hard-coded to return false. That matches today’s two Kind values, but it’s easy to forget to update once additional metadata table kinds are added, and it doesn’t reflect the docstring/PR description that this is kind-driven. Consider switching on kind() and making the non-exhaustive case unreachable to keep future additions honest.

bool MetadataTable::supports_time_travel() const noexcept { return false; }

[Copilot]

Pull request overview

Copilot reviewed 9 out of 9 changed files in this pull request and generated 1 comment.

[Copilot]

Pull request overview

Copilot reviewed 9 out of 9 changed files in this pull request and generated 3 comments.

[Copilot]

Pull request overview

Copilot reviewed 9 out of 9 changed files in this pull request and generated 1 comment.

[Copilot]

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated 1 comment.

Suppressed comments (3)

src/iceberg/inspect/metadata_table.h:80

• The PR description says Scan() returns Result<ArrowArray> with a default NotSupported implementation plus a zero-arg overload, but the code introduces Result<ArrowArrayStream> as a pure virtual method on MetadataTable (and adds a separate TimeTravelMetadataTable). Either the PR description should be updated to match the implemented API, or the API should be adjusted to match the stated intent (including explicitly providing a default NotSupported if that’s still desired).

  /// \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;

src/iceberg/inspect/snapshots_table.cc:73

• This copies snapshot.summary for every scanned snapshot solely to remove operation. For large summaries or large snapshot counts, this adds avoidable per-row allocations and hashing work. Prefer avoiding the full copy (e.g., detect the operation-only case without copying, or build a filtered map only when needed, or extend the map-appending helper to accept a predicate/skip key).

  auto summary = snapshot.summary;
  summary.erase(SnapshotSummaryFields::kOperation);
  if (summary.empty()) {
    ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(5)));
  } else {
    ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), summary));
  }

src/iceberg/test/snapshots_table_test.cc:112

• These constants rely on integer-literal type promotion rules. For clarity and to avoid accidental narrowing/overflow on some toolchains, consider making the intent explicit with int64_t/LL literals (e.g., 1234567890000LL * 1000).

  EXPECT_EQ(committed_at->Value(0), 1234567890000 * 1000);
  EXPECT_EQ(committed_at->Value(1), 9876543210000 * 1000);

[Copilot]

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated 2 comments.

Suppressed comments (3)

src/iceberg/inspect/snapshots_table.cc:73

• This makes a full copy of snapshot.summary for every row just to drop the operation entry, which can be costly for large scans. A more efficient approach is to build the Arrow map by iterating snapshot.summary and appending only entries whose key != kOperation, avoiding the intermediate copy (or only copying when the operation key is actually present).

  auto summary = snapshot.summary;
  summary.erase(SnapshotSummaryFields::kOperation);
  if (summary.empty()) {
    ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(5)));
  } else {
    ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), summary));
  }

src/iceberg/inspect/metadata_table.h:60

• The public factory now only supports compile-time selection (Make<ConcreteType>(...)). This makes it harder/impossible for callers who only have a MetadataTable::Kind at runtime to construct a table. Consider keeping (or reintroducing) a Make(table, Kind) overload as a thin wrapper around the concrete Make() methods to preserve the runtime-factory use case.

  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));
  }

src/iceberg/inspect/metadata_table.h:80

• The PR description says Scan() returns Result<ArrowArray>, but the implementation exposes Result<ArrowArrayStream>. Please update the PR description (or the API) so the documented behavior matches the code.

  /// \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;

[Copilot]

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated no new comments.

Suppressed comments (5)

src/iceberg/test/metadata_table_test_base.h:88

• On the ImportRecordBatchReader failure path, the passed-in ArrowArrayStream is not released, which can leak stream-owned resources (and the new scan API explicitly documents that callers own the stream). Since ReadAllBatches takes ownership via rvalue and returns early here, it should call ArrowArrayStreamRelease(&stream) (or equivalent helper) before returning the error.

  static Result<std::vector<std::shared_ptr<::arrow::RecordBatch>>> ReadAllBatches(
      ArrowArrayStream&& stream) {
    auto reader_result = ::arrow::ImportRecordBatchReader(&stream);
    if (!reader_result.ok()) {
      return InvalidArrowData(reader_result.status().ToString());
    }

src/iceberg/inspect/snapshots_table.cc:102

• Next() dereferences table_ unconditionally. After Close() runs, table_ is reset, and any subsequent Next() call would crash. Even if Arrow consumers are expected not to call get_next after release, adding an explicit guard (e.g., return an error or end-of-stream when table_ == nullptr) makes the stream more robust and prevents hard-to-diagnose crashes in misuse scenarios.

  Result<std::optional<ArrowArray>> Next() {
    const auto& snapshots = table_->snapshots();
    if (next_snapshot_ == snapshots.size()) {
      return std::nullopt;
    }

src/iceberg/inspect/snapshots_table.cc:73

• This copies the entire snapshot.summary map for every emitted row, which adds allocation/copy overhead during scans (especially with large summaries and across batches). A more efficient approach is to avoid the full copy by iterating the original map and appending only non-kOperation entries (or building a filtered view/temporary only when the operation key is present).

  auto summary = snapshot.summary;
  summary.erase(SnapshotSummaryFields::kOperation);
  if (summary.empty()) {
    ICEBERG_RETURN_UNEXPECTED(AppendNull(builder.column(5)));
  } else {
    ICEBERG_RETURN_UNEXPECTED(AppendStringMap(builder.column(5), summary));
  }

src/iceberg/test/metadata_table_test_base.h:109

• Tests hardcode the "operation" summary key as a string literal here, while other new tests use SnapshotSummaryFields::kOperation. Using the shared constant consistently would reduce duplication and prevent future drift if the canonical key changes.

        {"operation", "append"},

src/iceberg/inspect/history_table.cc:55

• MetadataTable now requires Scan() as the primary non-time-travel entry point, but HistoryTable is constructible and yet always returns NotSupported for Scan(). This makes the interface surprising for consumers (a table type that cannot be scanned). Consider either implementing a scan for HistoryTable, moving it behind a capability-specific interface (similar to TimeTravelMetadataTable), or making the non-scannable type unconstructible/hidden until scan support exists.

Result<ArrowArrayStream> HistoryTable::Scan() {
  return NotSupported("Scan is not supported for the history table");
}

[wgtmac]

Is it better to remove this extra layer of TimeTravelMetadataTable? We can check supports_time_travel() and by default Scan(const SnapshotSelection& snapshot_selection) can just return NotSupported.

[wgtmac]

ArrowArrayStreamRelease is not declared by this repository's Arrow C Data API; callers must invoke the stream's release callback. Please correct the ownership documentation.

[wgtmac]

This stream re-reads the mutable Table on every batch. If Table::Refresh() runs between get_next calls, later batches can come from a different snapshot list, skipping or truncating rows. Please capture the metadata or snapshot list when Scan() creates the stream.

[wgtmac]

HistoryTable remains constructible, but the new common Scan() API always returns NotSupported here. Java HistoryTable scans table.history() into rows. Please implement a streaming history scan, or keep non-scannable tables out of this interface.

[wgtmac]

std::ignore is used here without a direct include. Please add or replace std::ignore so this file does not rely on transitive includes.