kamcheungting-db commented on code in PR #1065:
URL: https://github.com/apache/iceberg-cpp/pull/1065#discussion_r4510452108
##########
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:
Done.
New version: Projection uses the reader’s physical Arrow schema.
I have ddded large_list coverage for both all-matching and filtered
residuals, with and without deletes and column pruning.
--
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]
