wgtmac commented on code in PR #873:
URL: https://github.com/apache/iceberg-cpp/pull/873#discussion_r3965842054
##########
src/iceberg/table_scan.cc:
##########
@@ -661,6 +720,69 @@ Result<std::vector<std::shared_ptr<FileScanTask>>>
DataTableScan::PlanFiles() co
return tasks;
}
+Result<FileScanTaskIterator> DataTableScan::PlanFilesIterator() const {
+ ICEBERG_ASSIGN_OR_RAISE(auto snapshot, this->snapshot());
+ if (!snapshot) {
+ return std::make_unique<EmptyIterator<std::shared_ptr<FileScanTask>>>();
+ }
+
+ std::shared_ptr<ScanMetrics> scan_metrics;
+ std::optional<std::chrono::steady_clock::time_point> planning_start;
+ if (context_.metrics_reporter) {
+ auto metrics_context = MetricsContext::Default();
+ scan_metrics = ScanMetrics::Make(*metrics_context);
+ planning_start = std::chrono::steady_clock::now();
+ }
+
+ TableMetadataCache metadata_cache(metadata_.get());
+ ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id,
metadata_cache.GetPartitionSpecsById());
+
+ SnapshotCache snapshot_cache(snapshot.get());
+ ICEBERG_ASSIGN_OR_RAISE(auto data_manifests,
snapshot_cache.DataManifests(io_));
+ ICEBERG_ASSIGN_OR_RAISE(auto delete_manifests,
snapshot_cache.DeleteManifests(io_));
+
+ if (scan_metrics) {
+ scan_metrics->total_data_manifests->Increment(
+ static_cast<int64_t>(data_manifests.size()));
+ scan_metrics->total_delete_manifests->Increment(
+ static_cast<int64_t>(delete_manifests.size()));
+ }
+
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto manifest_group,
+ ManifestGroup::Make(io_, schema_, specs_by_id,
+ {data_manifests.begin(), data_manifests.end()},
Review Comment:
These vectors are already materialized, but {begin(), end()} copies every
ManifestFile before the lazy iterator is returned. Can we move them into
ManifestGroup::Make() to avoid doubling peak metadata memory?
##########
src/iceberg/manifest/manifest_group.cc:
##########
@@ -373,6 +679,18 @@ Result<std::unique_ptr<ManifestReader>>
ManifestGroup::MakeReader(
return reader;
}
+bool ManifestGroup::PrepareStatsProjection(bool has_equality_deletes) {
+ // The caller's projection records whether stats were requested.
Equality-delete
+ // matching may add stats temporarily, but they should still be dropped from
the
+ // result when the original projection did not request them. Keeping this
decision
+ // here ensures eager and iterator planning use identical semantics.
+ const bool drop_stats = ManifestReader::ShouldDropStats(columns_);
+ if (has_equality_deletes) {
+ columns_ = ManifestReader::WithStatsColumns(columns_);
Review Comment:
PrepareStatsProjection() mutates columns_ in place. Reusing the same
ManifestGroup for a second plan can change whether stats are retained. Should
this temporary projection stay local to each plan?
##########
src/iceberg/manifest/manifest_reader.h:
##########
@@ -42,13 +43,23 @@ class ICEBERG_EXPORT ManifestReader {
virtual ~ManifestReader() = default;
/// \brief Read all manifest entries in the manifest file.
- ///
- /// TODO(gangwu): provide a lazy-evaluated iterator interface for better
performance.
virtual Result<std::vector<ManifestEntry>> Entries() = 0;
/// \brief Read only live (non-deleted) manifest entries.
virtual Result<std::vector<ManifestEntry>> LiveEntries() = 0;
+ /// \brief Lazily read manifest entries.
+ ///
+ /// Implementations using SupportsManifestEntryIteration stream entries
lazily. Other
+ /// implementations are adapted from Entries() for compatibility.
+ Result<std::unique_ptr<Iterator<ManifestEntry>>> EntriesIterator();
Review Comment:
Could we rename the new pull APIs before making them public?
These APIs return a fallible, single-pass source:
`Next()` returns `Result<std::optional<T>>`.
This is different from a standard C++ iterator or range.
Suggested names:
- `PlanFilesIterator()` → `PlanFilesStream()`
- `EntriesIterator()` → `EntriesStream()`
- `LiveEntriesIterator()` → `LiveEntriesStream()`
- `FileScanTaskIterator` → `FileScanTaskStream`
Example:
```cpp
auto stream = scan->PlanFilesStream();
auto next = stream->Next();
```
This keeps `PlanFiles()` for eager planning and makes the streaming API
explicit.
##########
src/iceberg/manifest/manifest_group.h:
##########
@@ -136,6 +137,16 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector
{
/// \brief Plan scan tasks for all matching data files.
Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles();
+ /// \brief Lazily plan scan tasks for matching data files.
+ ///
+ /// The returned iterator owns the planning state and may outlive this
ManifestGroup.
+ /// It reads one bounded manifest batch at a time instead of materializing
all manifest
+ /// entries and scan tasks. When PlanWith() configures an executor, entry
iterators for
+ /// manifests in each batch are opened in parallel, while entries are
consumed one
+ /// manifest at a time. Creating the iterator consumes this group's
configuration, so
Review Comment:
This iterator is lazy for data manifests, but Build() still reads all delete
manifests before the first Next(). Could we state that here so “stop planning
early” is not read as avoiding delete-manifest I/O?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
