mbutrovich commented on code in PR #2671:
URL: https://github.com/apache/iceberg-rust/pull/2671#discussion_r3691811085
##########
crates/integrations/datafusion/src/physical_plan/scan.rs:
##########
@@ -138,18 +137,60 @@ impl ExecutionPlan for IcebergTableScan {
fn execute(
&self,
- _partition: usize,
+ partition: usize,
_context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
- let fut = get_batch_stream(
- self.table.clone(),
- self.snapshot_id,
- self.projection.clone(),
- self.predicates.clone(),
- );
- let stream = futures::stream::once(fut).try_flatten();
-
- // Apply limit if specified
+ let stream: Pin<Box<dyn Stream<Item = DFResult<RecordBatch>> + Send>>
= match &self
+ .file_task_groups
+ {
+ Some(file_task_groups) => {
+ let Some(file_task_group) =
file_task_groups.get(partition).cloned() else {
+ return
Err(datafusion::common::DataFusionError::Internal(format!(
+ "IcebergTableScan partition {partition} does not
exist; scan has {} partitions",
+ file_task_groups.len()
+ )));
+ };
+
+ let tasks: FileScanTaskStream = Box::pin(futures::stream::iter(
+ (0..file_task_group.len()).map(move |idx|
Ok(file_task_group[idx].clone())),
+ ));
+ let stream = build_table_scan(&self.table, &self.scan_config)?
+ .arrow_reader_builder()
+ // Eager planning lets DataFusion drive scan concurrency
via output
+ // partitions. Match DataFusion's FileStream model, where
each
+ // output partition owns one ScanState; keep one data file
in
+ // flight per output partition here.
+ //
https://github.com/apache/datafusion/blob/ad8e7b7f2babe3fcddc3a4f9b5cd1ac0d1b16ad9/datafusion/datasource/src/file_stream/scan_state.rs#L42-L43
+ .with_data_file_concurrency_limit(1)
+ .build()
+ // TODO: Avoid cloning FileScanTasks here once ArrowReader
can accept shared tasks.
+ .read(tasks)
Review Comment:
The eager arm rebuilds a `TableScan` via `build_table_scan` on every
`execute(partition)` call just to read `.arrow_reader_builder()` off it.
`build_table_scan` (`scan_planning.rs:133-148`) does a snapshot lookup,
per-column schema validation, field id resolution, and `PlanContext`
construction. `arrow_reader_builder()`
(`crates/iceberg/src/scan/mod.rs:484-496`) does not read any of that: it only
copies `file_io`, `runtime`, `concurrency_limit_data_files`,
`row_group_filtering_enabled`, `row_selection_enabled`, and `batch_size` off
the `TableScan`. So this runs once per output partition per scan, and none of
the snapshot/schema/predicate work it does is used by the one thing this call
site wants from it.
This isn't an argument for dropping `TableScan` from this path. Routing both
the eager arm and `to_arrow()` through the same `arrow_reader_builder()` is
what closed the "two independently-hardcoded default sources with nothing
pinning their equivalence" gap from the earlier round on this PR
(https://github.com/apache/iceberg-rust/pull/2671#discussion_r3605675108).
Building `ArrowReaderBuilder::new(...)` directly here instead would reopen that
gap.
`plan_file_task_groups` (`scan_planning.rs:80-113`) already builds a
`TableScan` once, to call `.plan_files()`. Could that same instance, or the
`ArrowReaderBuilder` it produces, be threaded through to `IcebergTableScan` and
reused by every `execute(partition)` call, instead of rebuilding it per
partition here?
--
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]
