wgtmac commented on code in PR #1065:
URL: https://github.com/apache/iceberg-cpp/pull/1065#discussion_r4434023020
##########
src/iceberg/data/file_scan_task_reader.cc:
##########
@@ -162,6 +250,45 @@ class FileScanTaskReader::Impl {
new Impl(std::move(options), std::move(field_lookup),
std::move(delete_counter)));
}
+ // Columns that must be added to the read schema so the task's residual
filter
+ // can be evaluated, or nullopt when the residual must not be enforced. An
+ // empty vector means every referenced column is already projected.
+ Result<std::optional<std::vector<SchemaField>>> ResidualReadColumns(
+ const FileScanTask& task) const {
+ if (!apply_residuals_) {
+ return std::nullopt;
+ }
+ const auto& residual = task.residual_filter();
+ if (residual == nullptr || residual->op() == Expression::Operation::kTrue)
{
+ return std::nullopt;
+ }
+
+ // Residuals are unbound and reference the planning scan's schema: bind
them
+ // against the table schema with the scan's case sensitivity.
+ ICEBERG_ASSIGN_OR_RAISE(auto bound,
+ Binder::Bind(*table_schema_, residual,
case_sensitive_));
+ ICEBERG_ASSIGN_OR_RAISE(auto field_id_set,
+ ReferenceVisitor::GetReferencedFieldIds(bound));
Review Comment:
Binding can fold away references, but line 372 binds the original residual
again. With an unprojected required `flag`, `notNull(flag)` drops its ID here
and then fails to bind. Preserve the required references or carry the
simplified expression through; please also cover a partially folded AND.
##########
src/iceberg/data/file_scan_task_reader.cc:
##########
@@ -197,52 +326,80 @@ class FileScanTaskReader::Impl {
return MakeArrowArrayStream(std::move(reader));
}
- const bool has_position_deletes =
- std::any_of(task.delete_files().begin(), task.delete_files().end(),
- [](const std::shared_ptr<DataFile>& f) {
- return f->content == DataFile::Content::kPositionDeletes;
- });
+ // The read schema covers the projection plus any columns needed to enforce
+ // the residual filter; the stream projects back to the caller's schema.
+ std::shared_ptr<Schema> requested_schema = projected_schema_;
+ if (residual_columns.has_value() && !residual_columns->empty()) {
+ std::vector<SchemaField> fields(projected_schema_->fields().begin(),
+ projected_schema_->fields().end());
+ fields.insert(fields.end(), residual_columns->begin(),
residual_columns->end());
+ requested_schema = std::make_shared<Schema>(std::move(fields));
+ }
- ICEBERG_ASSIGN_OR_RAISE(
- auto delete_filter,
- DeleteFilter::Make(data_file->file_path, task.delete_files(),
projected_schema_,
- io_, field_lookup_, has_position_deletes,
delete_counter_));
+ std::unique_ptr<DeleteFilter> delete_filter;
+ std::shared_ptr<Schema> required_schema = requested_schema;
+ if (!task.delete_files().empty()) {
+ const bool has_position_deletes =
+ std::any_of(task.delete_files().begin(), task.delete_files().end(),
+ [](const std::shared_ptr<DataFile>& f) {
+ return f->content ==
DataFile::Content::kPositionDeletes;
+ });
- auto required_schema = delete_filter->RequiredSchema();
- auto project_batch_function =
ProjectionContext::ResolveProjectBatchFunction();
- // ProjectionContext borrows schemas that are kept in
MergeOnReadStreamSource.
- ICEBERG_ASSIGN_OR_RAISE(auto projection_context,
- ProjectionContext::Make(*required_schema,
*projected_schema_,
- project_batch_function));
+ ICEBERG_ASSIGN_OR_RAISE(
+ delete_filter,
+ DeleteFilter::Make(data_file->file_path, task.delete_files(),
requested_schema,
+ io_, field_lookup_, has_position_deletes,
delete_counter_));
+ required_schema = delete_filter->RequiredSchema();
+ }
auto options = MakeReaderOptions(task, io_, required_schema,
task.residual_filter(),
name_mapping_, properties_,
data_file->first_row_id,
data_file->data_sequence_number);
ICEBERG_ASSIGN_OR_RAISE(auto reader,
ReaderFactoryRegistry::Open(data_file->file_format, options));
- auto mor_reader = std::make_unique<MergeOnReadStreamSource>(
- std::move(reader), std::move(delete_filter),
std::move(required_schema),
- projected_schema_, std::move(projection_context));
- return MakeArrowArrayStream(std::move(mor_reader));
+ auto project_batch_function =
ProjectionContext::ResolveProjectBatchFunction();
+ // ProjectionContext borrows schemas that are kept in TaskStreamSource.
+ ICEBERG_ASSIGN_OR_RAISE(auto projection_context,
+ ProjectionContext::Make(*required_schema,
*projected_schema_,
+ project_batch_function));
Review Comment:
This new no-delete residual path still builds projection state from the
logical schema. With `large_list` output, the stream declares `list` while the
all-matching path returns the reader's physical array. Please use the reader's
physical schema, as in #1062, and test both all-matching and filtered residuals.
##########
src/iceberg/data/file_scan_task_reader.cc:
##########
@@ -131,9 +185,43 @@ class MergeOnReadStreamSource {
}
private:
+ // Returns the candidate rows that match the residual filter: either the
+ // previously selected rows, or every row when no filter has removed a row
yet.
+ Result<std::vector<int32_t>> KeepResidualMatches(const ArrowSchema&
input_arrow_schema,
+ const ArrowArray&
input_batch,
+ const std::vector<int32_t>&
selection,
+ bool all_rows_selected) {
+ ICEBERG_PRECHECK(std::in_range<int32_t>(input_batch.length),
+ "Batch length {} exceeds int32_t row index capacity",
+ input_batch.length);
+ ICEBERG_ASSIGN_OR_RAISE(auto row,
+ ArrowArrayStructLike::Make(input_arrow_schema,
input_batch));
+ std::vector<int32_t> kept;
+ auto keep = [&](int64_t index) -> Status {
+ ICEBERG_RETURN_UNEXPECTED(row->Reset(index));
+ ICEBERG_ASSIGN_OR_RAISE(bool matches,
residual_evaluator_->Evaluate(*row));
Review Comment:
Top-level `isNull(s)` and `notNull(s)` are accepted for an optional struct,
but evaluation calls `StructLikeAccessor::GetLiteral`, which rejects
non-primitive types before checking null. Please evaluate null predicates from
the scalar/validity and add a composite-column case.
--
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]
