feat: add lazy scan planning streams

[manuzhang]

Scan planning currently materializes manifest entries and file scan tasks before returning. This change adds fallible, single-pass streams so callers can consume data-manifest entries and scan tasks incrementally or stop early.

Builds on the generic pull utility introduced in #905, renaming it to Stream<T> consistently with the scan APIs. The API changes below are intentional; this PR does not preserve source compatibility through adapters or aliases.

What changed

• Add DataTableScan::PlanFilesStream() and consuming ManifestGroup::PlanFilesStream() APIs. Their eager PlanFiles() methods collect the stream into a vector.
• Make ManifestReader::EntriesStream() and LiveEntriesStream() the implementation extension points. Non-virtual Entries() and LiveEntries() collect these streams for eager callers.
• Separate stream interfaces from ownership: ManifestEntryStream = Stream<ManifestEntry> and FileScanTaskStream = Stream<std::shared_ptr<FileScanTask>>, with corresponding *Ptr aliases for std::unique_ptr.
• Next() returns Result<std::optional<T>>: an error indicates failure, an empty optional indicates exhaustion, and a value is the next item. Errors and exhaustion are terminal; ToVector() collects the remaining items.
• Preserve scan metrics for completed and partially consumed streams. Keep partition values for delete matching and residual evaluation even with a narrow manifest projection; drop unrequested statistics only after equality-delete matching.

Ownership and memory behavior

Streams own their reader/planning resources and may outlive the ManifestReader, ManifestGroup, or DataTableScan that created them. A configured executor is borrowed and must remain alive until the planning stream is destroyed.

Data-manifest entries are read in bounded batches. Serial planning opens one manifest at a time; executor-backed planning opens at most 32 matching manifest streams per batch and consumes their entries incrementally. Destroying a partially consumed stream releases its resources.

This does not make every part of planning lazy: snapshot/manifest-list metadata and the delete-file index remain materialized, and delete manifests are read eagerly before the planning stream is returned.

Breaking changes and migration

Generic stream implementations

Replace #include "iceberg/util/iterator.h" with #include "iceberg/util/stream.h", and rename Iterator<T> references/base classes to Stream<T>. Keep the NextImpl() override and fallible return type. The old header/type and compatibility alias are not retained.

ManifestGroup callers

ManifestGroup::PlanFiles() is now rvalue-qualified and consumes its group, as does PlanFilesStream(). For a group returned by ManifestGroup::Make():

// Before:
auto tasks = group->PlanFiles();

// After, eager consumption:
auto tasks = std::move(*group).PlanFiles();

// Or, incremental consumption (instead of the eager call):
auto tasks_stream = std::move(*group).PlanFilesStream();

Do not reuse the consumed group; construct and configure a new one to plan again. For an object rather than a pointer, use std::move(group). DataTableScan::PlanFiles() keeps its existing caller syntax and eager result type.

Custom ManifestReader implementations

Replace overrides of Entries() and LiveEntries() with these required overrides:

Result<ManifestEntryStreamPtr> EntriesStream() override;
Result<ManifestEntryStreamPtr> LiveEntriesStream() override;

Implement the entry-producing logic in self-contained streams derived from ManifestEntryStream, with NextImpl() returning Result<std::optional<ManifestEntry>>. Transfer or share all resources needed for consumption; returned streams must remain valid after reader destruction. LiveEntriesStream() must exclude deleted entries. Existing eager callers can continue calling Entries() and LiveEntries(), which now collect these streams. There is no eager-to-stream fallback or separate streaming-capability mixin.

Validation

• Reproduced the V2/V3 equality-delete CI failures locally before the fix, and added coverage for the same projection issue with position deletes. Both regressions pass after the fix, including serial/executor-backed planning and eager/streaming consumption for equality deletes.
• Built manifest_test, scan_test, and util_test with CMake on macOS (Debug, static bundle; REST/Hive/SQL catalog disabled).
• ctest --test-dir build --output-on-failure -R '^(manifest_test|scan_test|util_test)$': all three suites passed.
• clang-format --dry-run --Werror for changed C++ files and git diff --check passed.

[Copilot]

Pull request overview

This PR introduces a fallible, pull-based Iterator<T> abstraction and uses it to add streaming (lazy) scan planning APIs so manifest reading and file task planning can be consumed incrementally instead of fully materialized, while preserving scan metrics reporting for fully and partially consumed plans.

Changes:

• Add a generic Iterator<T> interface (Next() + ToVector()) for fallible, lazily produced values.
• Implement streaming manifest-entry reading and streaming file-scan-task planning via new *Iterator() APIs on ManifestReader, ManifestGroup, and DataTableScan (keeping existing eager APIs intact).
• Update examples and add tests covering iterator lifetime/resource ownership and lazy planning behavior.

Reviewed changes

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

Show a summary per file

File	Description
src/iceberg/util/meson.build	Installs the new iterator.h header in Meson builds.
src/iceberg/util/iterator.h	Adds the fallible pull-based Iterator<T> interface and a ToVector() helper.
src/iceberg/type_fwd.h	Forward-declares Iterator<T> for use in public APIs.
src/iceberg/test/table_scan_test.cc	Adds coverage for PlanFilesIterator() behavior and iterator lifetime beyond the scan object.
src/iceberg/test/manifest_reader_test.cc	Adds coverage that manifest entry iterators own reader resources and can outlive the reader.
src/iceberg/table_scan.h	Adds DataTableScan::PlanFilesIterator() public API.
src/iceberg/table_scan.cc	Implements lazy scan planning and metrics reporting for partially consumed iterators.
src/iceberg/manifest/manifest_reader.h	Adds EntriesIterator() / LiveEntriesIterator() streaming APIs (defaulting to eager adaptation).
src/iceberg/manifest/manifest_reader.cc	Implements streaming manifest entry iteration and refactors eager reads to build on iterators.
src/iceberg/manifest/manifest_reader_internal.h	Updates internal reader interface/state to support iterator-owned resources.
src/iceberg/manifest/manifest_group.h	Adds ManifestGroup::PlanFilesIterator() streaming planning API.
src/iceberg/manifest/manifest_group.cc	Implements pull-based task planning iterator over manifest batches and entries.
example/demo_example.cc	Demonstrates consuming scan tasks via PlanFilesIterator() using Next().

[Copilot]

Pull request overview

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

Suppressed comments (1)

src/iceberg/util/iterator.h:62

• Iterator::ToVector() unconditionally moves the element into the output vector. This fails to compile for copy-only T (copy-constructible but not move-constructible). Using std::move_if_noexcept(*value) keeps move semantics for moveable types while falling back to copy when move is unavailable/undesirable.

      values.push_back(std::move(value).value());

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

Suppressed comments (2)

src/iceberg/manifest/manifest_group.cc:474

• PlanFilesIterator() moves from *this, which silently consumes the ManifestGroup instance while leaving the original object in a moved-from state. To make this consumption explicit and prevent accidental reuse, consider making the method rvalue-qualified (e.g., PlanFilesIterator() &&) and/or changing the API to require ownership (e.g., a static/free function taking std::unique_ptr<ManifestGroup>).

Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>>
ManifestGroup::PlanFilesIterator() {
  auto group = std::make_unique<ManifestGroup>(std::move(*this));
  return FilePlanningIterator::Make(std::move(group));
}

src/iceberg/table_scan.cc:134

• If iterator_->Next() returns an error, Finalize() is not called here, so metrics/reporting will only happen when the wrapper iterator is destroyed. If the caller propagates the error but retains the iterator object for longer than expected, scan reporting may be significantly delayed. Consider finalizing on error as well (best-effort), so reporting occurs promptly when planning fails.

  Result<std::optional<std::shared_ptr<FileScanTask>>> Next() override {
    auto start = std::chrono::steady_clock::now();
    auto result = iterator_->Next();
    planning_duration_ += std::chrono::duration_cast<std::chrono::nanoseconds>(
        std::chrono::steady_clock::now() - start);
    if (result.has_value() && !result.value().has_value()) {
      Finalize();
    }
    return result;
  }

[Copilot]

Pull request overview

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

Suppressed comments (4)

src/iceberg/manifest/manifest_reader.cc:724

• ArrowSchema is a C struct without real move semantics; passing it “by value” and then using std::move(arrow_schema) is still a shallow copy, which makes correct ownership transfer depend on subtle guard/release ordering. To make this robust, explicitly transfer ownership at the call site (e.g., pass std::exchange(arrow_schema, ArrowSchema{}) into the iterator so the local is cleared), or wrap ArrowSchema in a move-only RAII type and move that into ManifestEntryIteratorImpl.

  ManifestEntryIteratorImpl(std::unique_ptr<Reader> reader,
                            std::shared_ptr<Schema> file_schema, ArrowSchema arrow_schema,
                            std::shared_ptr<InheritableMetadata> inheritable_metadata,
                            std::optional<int64_t> first_row_id, bool is_committed,
                            bool only_live, std::unique_ptr<Evaluator> evaluator,
                            std::unique_ptr<InclusiveMetricsEvaluator> metrics_evaluator,
                            std::shared_ptr<PartitionSet> partition_set,
                            std::shared_ptr<Counter> skip_counter, bool drop_stats)
      : reader_(std::move(reader)),
        file_schema_(std::move(file_schema)),
        arrow_schema_(std::exchange(arrow_schema, ArrowSchema{})),
        arrow_schema_guard_(&arrow_schema_),
        inheritable_metadata_(std::move(inheritable_metadata)),

src/iceberg/manifest/manifest_group.cc:480

• This implementation “consumes” *this via move, but the method is callable on an lvalue object that remains in a moved-from (unspecified) state; subsequent calls on the same ManifestGroup instance become easy to misuse. Consider making this API rvalue-qualified (e.g., PlanFilesIterator() &&) so callers must explicitly std::move(group) and the type system reinforces the “consumes configuration” contract; alternatively, set an internal consumed flag and fail fast on subsequent method calls.

Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>>
ManifestGroup::PlanFilesIterator() {
  auto group = std::make_unique<ManifestGroup>(std::move(*this));
  return FilePlanningIterator::Make(std::move(group));
}

src/iceberg/table_scan.cc:783

• Prefer std::make_unique<ReportingFileTaskIterator>(...) over constructing a std::unique_ptr from new directly; it’s safer (exception-proof with respect to intermediate allocations/argument evaluation) and consistent with modern C++ ownership patterns.

  return std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>(
      new ReportingFileTaskIterator(std::move(iterator), std::move(scan_metrics),
                                    planning_duration, context_.metrics_reporter,
                                    std::move(report).value()));

src/iceberg/table_scan.cc:123

• The PR description calls out metrics reporting for “completed and partially consumed iterators,” but the added tests shown only cover full consumption (ToVector()) and end-of-iteration reporting. Add a test that consumes only the first planned task (or none), destroys the iterator, and asserts that a MetricsReporter mock/stub is invoked exactly once with partial metrics recorded (and that it doesn’t require the DataTableScan to outlive the iterator).

  ~ReportingFileTaskIterator() override { Finalize(); }

[Copilot]

Pull request overview

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

Suppressed comments (1)

src/iceberg/table_scan.cc:145

• ReportingFileTaskIterator currently finalizes (and emits a metrics report) from its destructor even if iteration terminates due to an error from the underlying iterator. This differs from PlanFiles(), which only reports on successful planning, and can produce misleading “successful” scan reports for failed planning. Consider suppressing reporting after the first Next() error (while still allowing reporting when the iterator is simply dropped early by the caller).

  Result<std::optional<std::shared_ptr<FileScanTask>>> Next() override {
    auto start = std::chrono::steady_clock::now();
    auto result = iterator_->Next();
    planning_duration_ += std::chrono::duration_cast<std::chrono::nanoseconds>(
        std::chrono::steady_clock::now() - start);

[Copilot]

Pull request overview

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

Suppressed comments (3)

src/iceberg/manifest/manifest_group.cc:480

• PlanFilesIterator() moves from *this, leaving the original ManifestGroup in a moved-from (unspecified) state, but the API does not enforce that the object is consumed. This makes it easy for callers to accidentally keep using the same instance after calling PlanFilesIterator(). Consider making this API rvalue-qualified (e.g., PlanFilesIterator() &&) so consumption is explicit, or alternatively return an iterator that holds shared ownership of a stable planning state without invalidating the original object.

Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>>
ManifestGroup::PlanFilesIterator() {
  auto group = std::make_unique<ManifestGroup>(std::move(*this));
  return FilePlanningIterator::Make(std::move(group));
}

src/iceberg/manifest/manifest_reader.cc:829

• These helpers depend on RTTI via dynamic_cast, which adds an RTTI requirement and a runtime type check on every call. A more robust approach is to make the iterator APIs virtual on ManifestReader with default implementations that adapt from Entries()/LiveEntries(); implementations that can stream override them. This removes the RTTI dependency and keeps dispatch purely virtual.

Result<std::unique_ptr<Iterator<ManifestEntry>>> ManifestReader::EntriesIterator() {
  if (auto* iterable = dynamic_cast<SupportsManifestEntryIteration*>(this)) {
    return iterable->EntriesIterator();
  }
  ICEBERG_ASSIGN_OR_RAISE(auto entries, Entries());
  return std::make_unique<VectorIterator<ManifestEntry>>(std::move(entries));
}

Result<std::unique_ptr<Iterator<ManifestEntry>>> ManifestReader::LiveEntriesIterator() {
  if (auto* iterable = dynamic_cast<SupportsManifestEntryIteration*>(this)) {
    return iterable->LiveEntriesIterator();
  }
  ICEBERG_ASSIGN_OR_RAISE(auto entries, LiveEntries());
  return std::make_unique<VectorIterator<ManifestEntry>>(std::move(entries));
}

src/iceberg/manifest/manifest_reader.h:159

• The streaming planning code path destroys ManifestReader instances immediately after obtaining an entry iterator (e.g., in ManifestGroup::FilePlanningIterator::OpenNextManifest()), which implicitly requires that the returned iterator fully owns all resources needed for continued iteration. This lifetime/ownership requirement should be documented explicitly here (and/or on ManifestReader::{EntriesIterator,LiveEntriesIterator}) so third-party implementations of SupportsManifestEntryIteration don't accidentally return iterators that reference this and become dangling after the reader is destroyed.

/// \brief Optional mix-in for ManifestReader implementations that support lazy entry
/// iteration.
class ICEBERG_EXPORT SupportsManifestEntryIteration {
 public:
  virtual ~SupportsManifestEntryIteration() = default;

  /// \brief Lazily read manifest entries.
  virtual Result<std::unique_ptr<Iterator<ManifestEntry>>> EntriesIterator() = 0;

  /// \brief Lazily read only live (non-deleted) manifest entries.
  virtual Result<std::unique_ptr<Iterator<ManifestEntry>>> LiveEntriesIterator() = 0;
};

[Copilot]

🟡 Changes recommended

The executor-enabled iterator planning path currently materializes full manifest entry vectors per manifest batch, which can negate the intended bounded-memory behavior for very large manifests.

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

Review details

• Files reviewed: 18/18 changed files
• Comments generated: 1
• Review effort level: Lite

[Copilot]

🔵 Needs a closer look

The PR description documents a non-existent API name (PlanFilesIterator()), which should be corrected to match the shipped PlanFilesStream() API to avoid user confusion.

Review details

Suppressed comments (1)

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

src/iceberg/table_scan.h:472

• The PR description says callers can use DataTableScan::PlanFilesIterator(), but the public API added here is PlanFilesStream(). Please update the PR description (or add an alias) so the documented API name matches what is actually shipped.

• Files reviewed: 20/20 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🔵 Needs a closer look

It introduces new public streaming APIs and significantly refactors scan planning/manifest reading behavior, warranting final human review for correctness and compatibility.

Review details

Suppressed comments (1)

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

src/iceberg/table_scan.h:472

• PR description/user-impact section refers to DataTableScan::PlanFilesIterator(), but the public API added here is PlanFilesStream(). This mismatch can confuse API consumers and reviewers; please update the PR description (or add/rename an API entry point) so the documented name matches the shipped header.

• Files reviewed: 20/20 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🟡 Changes recommended

manifest_reader.h introduces a Stream<ManifestEntry> alias while ManifestEntry is only forward-declared there, which can cause compilation failures due to Stream<T> using std::optional<T>/std::vector<T> in its interface.

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

Review details

Suppressed comments (1)

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

src/iceberg/table_scan.h:479

• The PR description says callers can use DataTableScan::PlanFilesIterator(), but the API added in this change is PlanFilesStream(). Please either update the PR description/user-impact text to match the shipped API name, or rename the method for consistency before merging.

• Files reviewed: 18/18 changed files
• Comments generated: 1
• Review effort level: Lite

[Copilot]

🟡 Changes recommended

Two critical issues remain in executor lifetime handling and filtered-batch stream termination.

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

Review details

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

[Copilot]

🔵 Needs a closer look

The iterator API compatibility issue and executor batch early-termination issue remain unresolved.

Review details

Suppressed comments (2)

src/iceberg/manifest/manifest_group.cc:364

• When an executor is configured, this treats a batch with no matching manifests as end-of-stream. If the first 32 manifests are filtered out but a later manifest matches, NextEntry() returns nullopt and never examines that later manifest, so PlanFilesStream() silently drops valid tasks. Continue loading batches until one yields a stream or next_manifest_ reaches the end.

    if (manifests.empty()) {
      return false;
    }

src/iceberg/util/stream.h:45

• This replaces the public Iterator<T>/iterator.h API introduced by merged PR #905 with Stream<T>/stream.h, so clients that adopted #905 now fail to compile even though the scan-planning change is otherwise additive. Please retain a compatibility header and type (or explicitly version/document this breaking rename) before removing the installed API.

• Files reviewed: 18/18 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🟡 Changes recommended

Three critical compatibility blockers remain involving eager PlanFiles(), ManifestReader extension points, and the public Iterator interface.

Get a fresh assessment by requesting another Copilot review.

Review details

Suppressed comments (1)

src/iceberg/util/stream.h:45

• PR #905 already exposed and installed Iterator<T> through iceberg/util/iterator.h. Replacing it with Stream<T> and removing the old header breaks downstream includes and type names, even though this PR is stacked on that public API. Keep a compatibility header/type alias (or provide an explicit deprecation/migration path) instead of deleting the recently merged interface.

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

[Copilot]

🔵 Needs a closer look

A moderate unresolved stream-termination bug can silently drop later matching manifests.

Review details

Suppressed comments (1)

src/iceberg/manifest/manifest_group.cc:363

• When the next batch contains only manifests rejected by ShouldReadManifest, this returns false even if next_manifest_ still points at later manifests. The stream is then marked exhausted and silently drops every matching manifest after that skipped batch (for example, 32 filtered manifests followed by a matching one). Continue loading batches until either a non-empty batch is found or all manifests are consumed.

    if (manifests.empty()) {
      return false;
    }

• Files reviewed: 18/18 changed files
• Comments generated: 0 new
• Review effort level: Lite