Copilot commented on code in PR #873:
URL: https://github.com/apache/iceberg-cpp/pull/873#discussion_r3985353805
##########
src/iceberg/manifest/manifest_group.cc:
##########
@@ -131,6 +131,310 @@ ManifestGroup::~ManifestGroup() = default;
ManifestGroup::ManifestGroup(ManifestGroup&&) noexcept = default;
ManifestGroup& ManifestGroup::operator=(ManifestGroup&&) noexcept = default;
+class ManifestGroup::FilePlanningStream final : public FileScanTaskStream {
+ public:
+ static Result<FileScanTaskStreamPtr> Make(std::unique_ptr<ManifestGroup>
group) {
+ ICEBERG_RETURN_UNEXPECTED(group->CheckErrors());
+
+ group->delete_index_builder_.WithScanMetrics(group->scan_metrics_);
+ ICEBERG_ASSIGN_OR_RAISE(auto delete_index,
group->delete_index_builder_.Build());
+
+ auto stats_projection =
+ group->PrepareStatsProjection(delete_index->has_equality_deletes());
+
+ std::unique_ptr<Evaluator> data_file_evaluator;
+ if (group->file_filter_ &&
+ group->file_filter_->op() != Expression::Operation::kTrue) {
+ ICEBERG_ASSIGN_OR_RAISE(
+ data_file_evaluator,
+ Evaluator::Make(*DataFileFilterSchema(), group->file_filter_,
+ group->case_sensitive_));
+ }
+ const bool drop_stats = stats_projection.drop_stats;
+
+ return FileScanTaskStreamPtr(new FilePlanningStream(
+ std::move(group), std::move(delete_index),
std::move(data_file_evaluator),
+ std::move(stats_projection.columns), drop_stats));
+ }
+
+ Result<std::optional<std::shared_ptr<FileScanTask>>> NextImpl() override {
+ while (true) {
+ ICEBERG_ASSIGN_OR_RAISE(auto entry, NextEntry());
+ if (!entry.has_value()) {
+ return std::nullopt;
+ }
+
+ auto [spec_id, value] = std::move(entry).value();
+ if (group_->ignore_existing_ && value.status ==
ManifestStatus::kExisting) {
+ IncrementSkippedDataFiles();
+ continue;
+ }
+
+ ICEBERG_DCHECK(value.data_file != nullptr, "Data file cannot be null");
+ if (data_file_evaluator_) {
+ DataFileStructLike data_file(*value.data_file);
+ ICEBERG_ASSIGN_OR_RAISE(bool should_match,
+ data_file_evaluator_->Evaluate(data_file));
+ if (!should_match) {
+ IncrementSkippedDataFiles();
+ continue;
+ }
+ }
+
+ if (!group_->manifest_entry_predicate_(value)) {
+ IncrementSkippedDataFiles();
+ continue;
+ }
+
+ ICEBERG_ASSIGN_OR_RAISE(auto delete_files,
delete_index_->ForEntry(value));
+
+ // Equality-delete matching uses data-file statistics. Drop unrequested
stats only
+ // after the delete index has finished matching this entry.
+ if (drop_stats_) {
+ ContentFileUtil::DropAllStats(*value.data_file);
+ } else if (!group_->columns_to_keep_stats_.empty()) {
+ ContentFileUtil::DropUnselectedStats(*value.data_file,
+ group_->columns_to_keep_stats_);
+ }
+
+ UpdateResultMetrics(*value.data_file, delete_files);
+
+ ICEBERG_ASSIGN_OR_RAISE(auto residuals, GetResidualEvaluator(spec_id));
+ ICEBERG_ASSIGN_OR_RAISE(auto residual,
+
residuals->ResidualFor(value.data_file->partition));
+
+ return
std::optional<std::shared_ptr<FileScanTask>>{std::make_shared<FileScanTask>(
+ std::move(value.data_file), std::move(delete_files),
std::move(residual))};
+ }
+ }
+
+ private:
+ FilePlanningStream(std::unique_ptr<ManifestGroup> group,
+ std::unique_ptr<DeleteFileIndex> delete_index,
+ std::unique_ptr<Evaluator> data_file_evaluator,
+ std::vector<std::string> columns, bool drop_stats)
+ : group_(std::move(group)),
+ delete_index_(std::move(delete_index)),
+ data_file_evaluator_(std::move(data_file_evaluator)),
+ columns_(std::move(columns)),
+ drop_stats_(drop_stats) {}
+
+ using TaggedEntry = std::pair<int32_t, ManifestEntry>;
+ using TaggedStream = std::pair<int32_t, ManifestEntryStreamPtr>;
+
+ Result<std::optional<TaggedEntry>> NextEntry() {
+ if (!group_->executor_.has_value()) {
+ while (true) {
+ if (!entry_stream_) {
+ ICEBERG_ASSIGN_OR_RAISE(bool opened, OpenNextManifest());
+ if (!opened) {
+ return std::nullopt;
+ }
+ }
+
+ ICEBERG_ASSIGN_OR_RAISE(auto entry, entry_stream_->Next());
+ if (!entry.has_value()) {
+ entry_stream_.reset();
+ continue;
+ }
+ return std::optional<TaggedEntry>{std::in_place, current_spec_id_,
+ std::move(entry).value()};
+ }
+ }
+
+ while (true) {
+ if (next_batch_stream_ == batch_streams_.size()) {
+ ICEBERG_ASSIGN_OR_RAISE(bool loaded, LoadNextManifestBatch());
+ if (!loaded) {
+ return std::nullopt;
+ }
+ }
Review Comment:
When executor planning is enabled, `LoadNextManifestBatch()` returns `false`
when the current 32-manifest window contains no eligible manifests. This branch
treats that as end-of-stream, so if the first 32 manifests are filtered out but
a later manifest matches, the stream silently returns no tasks. Continue
loading batches until an eligible manifest is found or `next_manifest_` reaches
the end.
##########
src/iceberg/manifest/manifest_group.h:
##########
@@ -136,6 +137,18 @@ 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 stream 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
streams for
+ /// manifests in each batch are opened in parallel, while entries are
consumed one
+ /// manifest at a time. Delete manifests are still read eagerly when
creating the
+ /// stream because delete files must be indexed before data-file planning
can begin.
+ /// Creating the stream consumes this group's configuration, so this method
may only
+ /// be called on an rvalue.
Review Comment:
`PlanWith()` stores only a non-owning `std::reference_wrapper<Executor>`,
and the moved group retains that reference for later `Next()` calls. The stream
is documented as outliving this group/scan, but consuming it after a locally
scoped executor is destroyed will dereference a dangling reference. Either
retain an owned executor lifetime or document/enforce that a configured
executor must remain alive until the stream is destroyed.
--
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]
