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

[Copilot]

Pull request overview

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

Suppressed comments (1)

Previously missed (1) — in code that hasn't changed since the last review.

src/iceberg/inspect/snapshots_table.cc:75

• AppendSnapshot() unconditionally copies snapshot.summary just to drop the operation key. For tables with many snapshots or large summaries, this per-row unordered_map copy can become a noticeable CPU/memory cost during streaming scans.

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

[manuzhang]

@WZhuo Can you check latest review comments? I think it's getting close to merge and I have a follow-up PR #876 that depends on this.

[manuzhang]

It would be better to decouple with Arrow types.

[WZhuo]

Agreed. Decoupling the metadata table API from Arrow C types would be cleaner. This likely deserves a metadata-specific batch reader abstraction rather than reusing the file-oriented Reader interface. Since that is a broader API change, I would prefer to address it in a follow-up PR and keep ArrowArrayStream for this PR.

[manuzhang]

Well, this defines the basic interface for all metadata tables. I think it will be better to get it right first, maybe in a separate PR. @wgtmac What do you think?

[WZhuo]

@WZhuo Can you check latest review comments? I think it's getting close to merge and I have a follow-up PR #876 that depends on this.

Ok, I'll fix the comments

[Copilot]

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Pull request overview

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

[Copilot]

🟡 Changes recommended

The snapshots summary filtering behavior and its test expectations conflict with the PR’s stated contract (null vs empty summary after removing operation).

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

• Files reviewed: 15/15 changed files
• Comments generated: 2
• Review effort level: Lite