Copilot commented on code in PR #952:
URL: https://github.com/apache/iceberg-cpp/pull/952#discussion_r4061090098
##########
src/iceberg/manifest/manifest_group.cc:
##########
@@ -296,67 +376,23 @@ class ManifestGroup::FilePlanningStream final : public
FileScanTaskStream {
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_));
+ ICEBERG_ASSIGN_OR_RAISE(auto evaluator,
context_.MakeResidualEvaluator(spec_id));
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));
- const bool has_non_deleted_files =
- manifest.has_added_files() || manifest.has_existing_files();
- const bool has_non_existing_files =
- manifest.has_added_files() || manifest.has_deleted_files();
- const bool has_only_ignored_files =
- (group_->ignore_deleted_ && !has_non_deleted_files) ||
- (group_->ignore_existing_ && !has_non_existing_files);
- if (!should_match || has_only_ignored_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,
columns_));
- ICEBERG_ASSIGN_OR_RAISE(entry_stream_, group_->ignore_deleted_
- ? reader->LiveEntriesStream()
- : reader->EntriesStream());
- current_spec_id_ = manifest.partition_spec_id;
- return true;
- }
- return false;
- }
-
Result<bool> LoadNextManifestBatch() {
+ const size_t batch_size = group_->executor_.has_value() ?
kManifestReadBatchSize : 1;
std::vector<const ManifestFile*> manifests;
- manifests.reserve(kManifestReadBatchSize);
+ manifests.reserve(batch_size);
while (next_manifest_ < group_->data_manifests_.size() &&
- manifests.size() < kManifestReadBatchSize) {
+ manifests.size() < batch_size) {
Review Comment:
`NextEntry()` treats a `false` return from `LoadNextManifestBatch()` as
end-of-stream, but this loader can return `false` when the current batch
contains only filtered manifests while `next_manifest_` still has input. With
`batch_size == 1`, any skipped manifest before a matching one now truncates
serial planning; an all-skipped executor batch has the same problem. Continue
loading batches until one is non-empty or all manifests have been examined, and
add a regression case with a skipped manifest before a match.
--
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]
