feat: add lazy scan planning iterators

[manuzhang]

What changed

• add a fallible, pull-based Iterator<T> utility
• stream manifest entries and file scan task planning without materializing all results
• preserve scan metrics reporting for completed and partially consumed iterators
• update the demo and add coverage for iterator lifetime and table scan planning

Why

The existing scan planning APIs materialize every manifest entry and file scan task before returning. Large tables can therefore require memory proportional to the full scan plan. The new iterator APIs process one manifest batch at a time and let callers consume tasks on demand while keeping the existing eager APIs compatible.

User impact

Callers can use DataTableScan::PlanFilesIterator() to bound planning memory and stop planning early. Existing PlanFiles() behavior remains available. Streaming planning is pull-based and does not eagerly submit manifest work to the configured planning executor.

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