Copilot commented on code in PR #873:
URL: https://github.com/apache/iceberg-cpp/pull/873#discussion_r3916269729
##########
src/iceberg/table_scan.cc:
##########
@@ -661,6 +721,70 @@ Result<std::vector<std::shared_ptr<FileScanTask>>>
DataTableScan::PlanFiles() co
return tasks;
}
+Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>>
+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()},
+ {delete_manifests.begin(), delete_manifests.end()}));
Review Comment:
This constructs new vectors from `data_manifests` / `delete_manifests`
iterators, which copies all `ManifestFile` items even though the source
containers are already `std::vector`s. That extra allocation/copy partially
undermines the goal of reducing planning memory overhead for large tables.
Consider passing the vectors by move (e.g., `std::move(data_manifests)` /
`std::move(delete_manifests)`) if `ManifestGroup::Make` can accept rvalues, or
adding an overload that takes `std::vector<ManifestFile>` by value so callers
can move into it without rebuilding.
##########
src/iceberg/manifest/manifest_group.cc:
##########
@@ -132,6 +132,305 @@ 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 =
+ 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_));
+ }
+
+ 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>;
+ using TaggedIterator = std::pair<int32_t,
std::unique_ptr<Iterator<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 (true) {
+ if (next_batch_iterator_ == batch_iterators_.size()) {
+ ICEBERG_ASSIGN_OR_RAISE(bool loaded, LoadNextManifestBatch());
+ if (!loaded) {
+ return std::nullopt;
+ }
+ }
+
+ auto& [spec_id, iterator] = batch_iterators_[next_batch_iterator_];
+ ICEBERG_ASSIGN_OR_RAISE(auto entry, iterator->Next());
+ if (!entry.has_value()) {
+ iterator.reset();
+ ++next_batch_iterator_;
+ continue;
+ }
+ return std::optional<TaggedEntry>{std::in_place, spec_id,
std::move(entry).value()};
+ }
+ }
+
+ 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;
+ }
+
+ // Open the readers concurrently, but keep their iterators instead of
collecting
+ // entries here. This preserves bounded memory for large manifests while
retaining
+ // parallel manifest initialization when an executor is configured.
+ ICEBERG_ASSIGN_OR_RAISE(
+ batch_iterators_,
+ ParallelCollect(
+ group_->executor_, manifests,
+ [this](const ManifestFile* manifest) ->
Result<std::vector<TaggedIterator>> {
+ ICEBERG_ASSIGN_OR_RAISE(auto reader,
group_->MakeReader(*manifest));
+ ICEBERG_ASSIGN_OR_RAISE(auto iterator, group_->ignore_deleted_
+ ?
reader->LiveEntriesIterator()
+ :
reader->EntriesIterator());
+
+ std::vector<TaggedIterator> tagged_iterators;
+ tagged_iterators.emplace_back(manifest->partition_spec_id,
+ std::move(iterator));
+ return tagged_iterators;
+ }));
+ next_batch_iterator_ = 0;
+ return true;
+ }
+
+ void IncrementSkippedDataManifests() {
+ if (group_->scan_metrics_) {
+ group_->scan_metrics_->skipped_data_manifests->Increment(1);
+ }
+ }
+
+ void IncrementSkippedDataFiles() {
+ if (group_->scan_metrics_) {
+ group_->scan_metrics_->skipped_data_files->Increment(1);
+ }
+ }
+
+ void UpdateResultMetrics(const DataFile& data_file,
+ const std::vector<std::shared_ptr<DataFile>>&
delete_files) {
+ if (!group_->scan_metrics_) {
+ return;
+ }
+
+ group_->scan_metrics_->total_file_size_in_bytes->Increment(
+ ContentFileUtil::ContentSizeInBytes(data_file));
+ group_->scan_metrics_->result_data_files->Increment(1);
+ group_->scan_metrics_->result_delete_files->Increment(
+ static_cast<int64_t>(delete_files.size()));
+ int64_t deletes_size = 0;
+ for (const auto& delete_file : delete_files) {
+ deletes_size += ContentFileUtil::ContentSizeInBytes(*delete_file);
+ }
+
group_->scan_metrics_->total_delete_file_size_in_bytes->Increment(deletes_size);
+ }
+
+ std::unique_ptr<ManifestGroup> group_;
+ std::unique_ptr<DeleteFileIndex> delete_index_;
+ std::unique_ptr<Evaluator> data_file_evaluator_;
+ std::unordered_map<int32_t, std::unique_ptr<ManifestEvaluator>>
manifest_evaluators_;
+ std::unordered_map<int32_t, std::shared_ptr<ResidualEvaluator>>
residual_evaluators_;
+ std::unique_ptr<Iterator<ManifestEntry>> entry_iterator_;
+ std::vector<TaggedIterator> batch_iterators_;
+ size_t next_manifest_ = 0;
+ size_t next_batch_iterator_ = 0;
+ int32_t current_spec_id_ = 0;
+ bool drop_stats_;
+
+ // Limit the number of manifest readers and iterators retained by
executor-backed
+ // planning. The executor still controls actual task concurrency, while this
fixed
+ // cap prevents resource use from scaling with the total manifest count.
Entries
+ // within each manifest remain streamed, so this does not cap manifest size.
+ static constexpr size_t kManifestReadBatchSize = 32;
Review Comment:
The manifest batch size is a hard-coded constant in the iterator
implementation. Since this directly affects the peak number of
concurrently-open readers/iterators (resource usage) and may need tuning for
different environments, consider making it configurable (e.g., via
`ManifestGroup` options / context) or at least hoisting it to a shared
configuration point so it can be adjusted without editing the implementation.
##########
src/iceberg/manifest/manifest_reader.cc:
##########
@@ -894,74 +1047,27 @@ Result<std::vector<ManifestEntry>>
ManifestReaderImpl::ReadEntries(bool only_liv
ICEBERG_RETURN_UNEXPECTED(OpenReader(std::move(projected_data_file_schema)));
ICEBERG_DCHECK(file_reader_ != nullptr, "File reader should be initialized");
- std::vector<ManifestEntry> manifest_entries;
ICEBERG_ASSIGN_OR_RAISE(auto arrow_schema, file_reader_->Schema());
internal::ArrowSchemaGuard schema_guard(&arrow_schema);
// Get evaluators if needed
- Evaluator* evaluator = nullptr;
- InclusiveMetricsEvaluator* metrics_evaluator = nullptr;
+ std::unique_ptr<Evaluator> evaluator;
+ std::unique_ptr<InclusiveMetricsEvaluator> metrics_evaluator;
if (HasPartitionFilter() || HasRowFilter()) {
- ICEBERG_ASSIGN_OR_RAISE(evaluator, GetEvaluator());
+ ICEBERG_ASSIGN_OR_RAISE(evaluator, TakeEvaluator());
}
if (HasRowFilter()) {
- ICEBERG_ASSIGN_OR_RAISE(metrics_evaluator, GetMetricsEvaluator());
+ ICEBERG_ASSIGN_OR_RAISE(metrics_evaluator, TakeMetricsEvaluator());
}
bool drop_stats = drop_stats_ && ShouldDropStats(columns_);
-
- while (true) {
- ICEBERG_ASSIGN_OR_RAISE(auto result, file_reader_->Next());
- if (!result.has_value()) {
- break; // EOF
- }
-
- internal::ArrowArrayGuard array_guard(&result.value());
- ICEBERG_ASSIGN_OR_RAISE(
- auto entries, ParseManifestEntry(&arrow_schema, &result.value(),
*file_schema_,
- first_row_id_, is_committed_));
-
- for (auto& entry : entries) {
- ICEBERG_RETURN_UNEXPECTED(inheritable_metadata_->Apply(entry));
-
- if (only_live && !entry.IsAlive()) {
- continue;
- }
-
- if (needs_filtering) {
- ICEBERG_DCHECK(entry.data_file != nullptr, "Data file cannot be null");
- if (evaluator) {
- ICEBERG_ASSIGN_OR_RAISE(bool partition_match,
-
evaluator->Evaluate(entry.data_file->partition));
- if (!partition_match) {
- if (skip_counter_) skip_counter_->Increment(1);
- continue;
- }
- }
- if (metrics_evaluator) {
- ICEBERG_ASSIGN_OR_RAISE(bool metrics_match,
-
metrics_evaluator->Evaluate(*entry.data_file));
- if (!metrics_match) {
- if (skip_counter_) skip_counter_->Increment(1);
- continue;
- }
- }
- ICEBERG_ASSIGN_OR_RAISE(bool in_partition_set,
InPartitionSet(*entry.data_file));
- if (!in_partition_set) {
- if (skip_counter_) skip_counter_->Increment(1);
- continue;
- }
- }
-
- if (drop_stats) {
- ContentFileUtil::DropAllStats(*entry.data_file);
- }
-
- manifest_entries.push_back(std::move(entry));
- }
- }
-
- return manifest_entries;
+ auto iterator = std::unique_ptr<Iterator<ManifestEntry>>(new
ManifestEntryIteratorImpl(
+ std::move(file_reader_), file_schema_, std::move(arrow_schema),
+ inheritable_metadata_, first_row_id_, is_committed_, only_live,
+ std::move(evaluator), std::move(metrics_evaluator), partition_set_,
skip_counter_,
+ drop_stats));
Review Comment:
This uses `new` directly to construct the iterator implementation. While
it's immediately wrapped, using
`std::make_unique<ManifestEntryIteratorImpl>(...)` improves readability and
reduces the chance of accidental raw-pointer handling in future edits. The
existing `schema_guard.Release()` pattern can remain the same, but the iterator
construction can be modernized to avoid explicit `new`.
--
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]
