Skip to content

Commit 6988eb0

Browse files
authored
feat: add lazy scan planning streams (#873)
Scan planning currently materializes manifest entries and file scan tasks before returning. This change adds fallible, single-pass streams so callers can consume entries and tasks incrementally or stop early. - Add `DataTableScan::PlanFilesStream()` and `ManifestGroup::PlanFilesStream()`. The eager `PlanFiles()` methods collect the stream into a vector. - Add `ManifestReader::EntriesStream()` and `LiveEntriesStream()` as the reader implementation extension points. `Entries()` and `LiveEntries()` collect these streams for eager callers. - Rename the pull utility from `Iterator<T>` to `Stream<T>`, with `ManifestEntryStream` and `FileScanTaskStream` aliases and corresponding owning `*Ptr` aliases. `Next()` returns `Result<std::optional<T>>`; errors and exhaustion are terminal, and `ToVector()` collects the remaining values. - Distinguish explicit empty selections from default projections: `Select({})` selects no user columns, while `Select({"*"})` or omitting `Select()` selects all columns. Table projections still include fields required by filters, and an explicit empty selection conflicts with an explicit projection schema. - Preserve partition values needed for delete matching and residual evaluation under narrow projections. Read statistics needed for equality-delete matching, then drop statistics the caller did not request. - Report scan metrics on exhaustion or early stream destruction, without emitting a successful report after a planning error. Streams own their reader or planning resources and may outlive the object that created them. A configured executor is borrowed and must remain alive until the planning stream is destroyed. Serial planning opens one data manifest at a time; executor-backed planning opens at most 32 matching manifest streams per batch and consumes their entries incrementally. `DataTableScan` uses the configured executor when there is more than one data manifest or more than one delete manifest. Snapshot and manifest-list metadata remain materialized, and delete manifests are read eagerly to build the delete-file index before returning the planning stream. API changes: include `iceberg/util/stream.h` and use `Stream<T>` in place of `Iterator<T>`. `ManifestGroup::PlanFiles()` and `PlanFilesStream()` consume the group and require an rvalue, such as `std::move(*group).PlanFilesStream()`. Custom manifest readers implement `EntriesStream()` and `LiveEntriesStream()` and return streams that remain valid after reader destruction. `DataTableScan::PlanFiles()` retains its existing caller syntax and eager result type. Validation: built `manifest_test` and `scan_test` with CMake and ran `ctest --test-dir build --output-on-failure -R '^(manifest_test|scan_test)$'`; both suites passed. Coverage includes empty and wildcard projections, equality and position deletes, partition isolation, executor-backed planning, stream lifetimes, and scan reporting. `git diff --check` passed.
1 parent 6361791 commit 6988eb0

18 files changed

Lines changed: 1227 additions & 314 deletions

‎example/demo_example.cc‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -79,16 +79,26 @@ int main(int argc, char** argv) {
7979
}
8080

8181
auto scan = std::move(scan_result.value());
82-
auto plan_result = scan->PlanFiles();
82+
auto plan_result = scan->PlanFilesStream();
8383
if (!plan_result.has_value()) {
8484
std::cerr << "Failed to plan files: " << plan_result.error().message << std::endl;
8585
return 1;
8686
}
8787

8888
std::cout << "Scan tasks: " << std::endl;
8989
auto scan_tasks = std::move(plan_result.value());
90-
for (const auto& scan_task : scan_tasks) {
91-
std::cout << " - " << scan_task->data_file()->file_path << std::endl;
90+
while (true) {
91+
auto task_result = scan_tasks->Next();
92+
if (!task_result.has_value()) {
93+
std::cerr << "Failed to plan next file: " << task_result.error().message
94+
<< std::endl;
95+
return 1;
96+
}
97+
if (!task_result.value().has_value()) {
98+
break;
99+
}
100+
std::cout << " - " << task_result.value().value()->data_file()->file_path
101+
<< std::endl;
92102
}
93103

94104
return 0;

0 commit comments

Comments
 (0)