Copilot commented on code in PR #873:
URL: https://github.com/apache/iceberg-cpp/pull/873#discussion_r3913433044
##########
src/iceberg/manifest/manifest_group.cc:
##########
@@ -132,6 +132,293 @@ ManifestGroup::~ManifestGroup() = default;
ManifestGroup::ManifestGroup(ManifestGroup&&) noexcept = default;
ManifestGroup& ManifestGroup::operator=(ManifestGroup&&) noexcept = default;
+class ManifestGroup::FilePlanningIterator final
+ : public Iterator<std::shared_ptr<FileScanTask>> {
+ public:
+ static Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>> 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());
+
+ const bool drop_stats = ManifestReader::ShouldDropStats(group->columns_);
+ if (delete_index->has_equality_deletes()) {
+ group->columns_ = ManifestReader::WithStatsColumns(group->columns_);
+ }
+
+ 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_));
+ }
+
+ return std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>(
+ new FilePlanningIterator(std::move(group), std::move(delete_index),
+ std::move(data_file_evaluator), drop_stats));
+ }
+
+ Result<std::optional<std::shared_ptr<FileScanTask>>> Next() 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:
+ FilePlanningIterator(std::unique_ptr<ManifestGroup> group,
+ std::unique_ptr<DeleteFileIndex> delete_index,
+ std::unique_ptr<Evaluator> data_file_evaluator, bool
drop_stats)
+ : group_(std::move(group)),
+ delete_index_(std::move(delete_index)),
+ data_file_evaluator_(std::move(data_file_evaluator)),
+ drop_stats_(drop_stats) {}
+
+ using TaggedEntry = std::pair<int32_t, ManifestEntry>;
+
+ Result<std::optional<TaggedEntry>> NextEntry() {
+ if (!group_->executor_.has_value()) {
+ while (true) {
+ if (!entry_iterator_) {
+ ICEBERG_ASSIGN_OR_RAISE(bool opened, OpenNextManifest());
+ if (!opened) {
+ return std::nullopt;
+ }
+ }
+
+ ICEBERG_ASSIGN_OR_RAISE(auto entry, entry_iterator_->Next());
+ if (!entry.has_value()) {
+ entry_iterator_.reset();
+ continue;
+ }
+ return std::optional<TaggedEntry>{std::in_place, current_spec_id_,
+ std::move(entry).value()};
+ }
+ }
+
+ while (next_batch_entry_ == batch_entries_.size()) {
+ ICEBERG_ASSIGN_OR_RAISE(bool loaded, LoadNextManifestBatch());
+ if (!loaded) {
+ return std::nullopt;
+ }
+ }
+ return
std::optional<TaggedEntry>{std::move(batch_entries_[next_batch_entry_++])};
+ }
+
+ Result<ManifestEvaluator*> GetManifestEvaluator(int32_t spec_id) {
+ auto cached = manifest_evaluators_.find(spec_id);
+ if (cached != manifest_evaluators_.end()) {
+ return cached->second.get();
+ }
+
+ auto spec_iter = group_->specs_by_id_.find(spec_id);
+ ICEBERG_CHECK(spec_iter != group_->specs_by_id_.cend(),
+ "Cannot find partition spec for ID {}", spec_id);
+
+ const auto& spec = spec_iter->second;
+ auto projector =
+ Projections::Inclusive(*spec, *group_->schema_,
group_->case_sensitive_);
+ ICEBERG_ASSIGN_OR_RAISE(auto partition_filter,
+ projector->Project(group_->data_filter_));
+ ICEBERG_ASSIGN_OR_RAISE(partition_filter,
And::Make(std::move(partition_filter),
+
group_->partition_filter_));
+ ICEBERG_ASSIGN_OR_RAISE(auto evaluator,
+ ManifestEvaluator::MakePartitionFilter(
+ std::move(partition_filter), spec,
*group_->schema_,
+ group_->case_sensitive_));
+ auto* result = evaluator.get();
+ manifest_evaluators_.emplace(spec_id, std::move(evaluator));
+ return result;
+ }
+
+ Result<ResidualEvaluator*> GetResidualEvaluator(int32_t spec_id) {
+ auto cached = residual_evaluators_.find(spec_id);
+ if (cached != residual_evaluators_.end()) {
+ return cached->second.get();
+ }
+
+ auto spec_iter = group_->specs_by_id_.find(spec_id);
+ ICEBERG_CHECK(spec_iter != group_->specs_by_id_.cend(),
+ "Cannot find partition spec for ID {}", spec_id);
+
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto evaluator,
+ ResidualEvaluator::Make(
+ (group_->ignore_residuals_ ? True::Instance() :
group_->data_filter_),
+ *spec_iter->second, *group_->schema_, group_->case_sensitive_));
+ auto* result = evaluator.get();
+ residual_evaluators_.emplace(spec_id, std::move(evaluator));
+ return result;
+ }
+
+ Result<bool> ShouldReadManifest(const ManifestFile& manifest) {
+ ICEBERG_ASSIGN_OR_RAISE(auto evaluator,
+ GetManifestEvaluator(manifest.partition_spec_id));
+ ICEBERG_ASSIGN_OR_RAISE(bool should_match, evaluator->Evaluate(manifest));
+ if (!should_match ||
+ (group_->ignore_deleted_ && !manifest.has_added_files() &&
+ !manifest.has_existing_files()) ||
+ (group_->ignore_existing_ && !manifest.has_added_files() &&
+ !manifest.has_deleted_files())) {
+ IncrementSkippedDataManifests();
+ return false;
+ }
+
+ if (group_->scan_metrics_) {
+ group_->scan_metrics_->scanned_data_manifests->Increment(1);
+ }
+ return true;
+ }
+
+ Result<bool> OpenNextManifest() {
+ while (next_manifest_ < group_->data_manifests_.size()) {
+ const auto& manifest = group_->data_manifests_[next_manifest_++];
+ ICEBERG_ASSIGN_OR_RAISE(bool should_read, ShouldReadManifest(manifest));
+ if (!should_read) {
+ continue;
+ }
+
+ ICEBERG_ASSIGN_OR_RAISE(auto reader, group_->MakeReader(manifest));
+ ICEBERG_ASSIGN_OR_RAISE(entry_iterator_, group_->ignore_deleted_
+ ?
reader->LiveEntriesIterator()
+ :
reader->EntriesIterator());
+ current_spec_id_ = manifest.partition_spec_id;
+ return true;
+ }
+ return false;
+ }
+
+ Result<bool> LoadNextManifestBatch() {
+ std::vector<const ManifestFile*> manifests;
+ manifests.reserve(kManifestReadBatchSize);
+ while (next_manifest_ < group_->data_manifests_.size() &&
+ manifests.size() < kManifestReadBatchSize) {
+ const auto& manifest = group_->data_manifests_[next_manifest_++];
+ ICEBERG_ASSIGN_OR_RAISE(bool should_read, ShouldReadManifest(manifest));
+ if (should_read) {
+ manifests.push_back(&manifest);
+ }
+ }
+
+ if (manifests.empty()) {
+ return false;
+ }
+
+ ICEBERG_ASSIGN_OR_RAISE(
+ batch_entries_,
+ ParallelCollect(
+ group_->executor_, manifests,
+ [this](const ManifestFile* manifest) ->
Result<std::vector<TaggedEntry>> {
+ ICEBERG_ASSIGN_OR_RAISE(auto reader,
group_->MakeReader(*manifest));
+ ICEBERG_ASSIGN_OR_RAISE(auto iterator, group_->ignore_deleted_
+ ?
reader->LiveEntriesIterator()
+ :
reader->EntriesIterator());
+ ICEBERG_ASSIGN_OR_RAISE(auto entries, iterator->ToVector());
+
Review Comment:
When an executor is configured, LoadNextManifestBatch() fully materializes
every selected manifest's entries via iterator->ToVector(). This can still
require memory proportional to the largest manifest(s) in the batch,
undermining the "bounded"/streaming planning goal for very large manifests even
though the API is iterator-based.
--
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]
