t3hw commented on code in PR #2342:
URL: https://github.com/apache/iceberg-rust/pull/2342#discussion_r3106860089
##########
crates/iceberg/src/arrow/reader.rs:
##########
@@ -281,16 +282,25 @@ impl ArrowReader {
tasks
.map_ok(move |task| {
let file_io = file_io.clone();
+ let delete_file_loader =
self.delete_file_loader.clone();
+
+ async move {
+ let stream = Self::process_file_scan_task(
+ task,
+ batch_size,
+ file_io,
+ delete_file_loader,
+ row_group_filtering_enabled,
+ row_selection_enabled,
+ parquet_read_options,
+ )
+ .await?;
- Self::process_file_scan_task(
- task,
- batch_size,
- file_io,
- self.delete_file_loader.clone(),
- row_group_filtering_enabled,
- row_selection_enabled,
- parquet_read_options,
- )
+ // Spawn stream consumption onto a separate tokio
task so that
+ // CPU-heavy Parquet decompression runs on the
tokio thread pool
+ // rather than being serialized on the polling
thread.
+ Ok(Self::spawn_record_batch_stream(stream, 1))
Review Comment:
Hi @vustef -thanks for the pointer, I read #1684 and Liu's comment carefully
before drafting this.
I think Liu's architecture model is right: step 2 (parallel execution of
FileScanTasks) belongs to the engine. That's exactly what Comet does in #2020 -
one DF SessionContext per Spark Task, ArrowReader at
concurrency==1, cross-task parallelism owned by Spark.
My PR keeps that whole path intact: when concurrency_limit_data_files == 1,
the #2020 fast path (and_then + try_flatten) is used verbatim - no spawn, no
channel, no reordering. Callers that own their own parallelism (Comet, Spark,
or any future DF executor that shards FileScanTasks across partitions) don't
see a behavior change.
The change applies only on the multi-concurrency branch, which today is
hit by two groups:
1. Callers using the raw `ArrowReader` / `table.scan().to_arrow()` path
directly (my use case: streaming RecordBatches into a sink via the core crate,
no DataFusion). For these callers, the existing `try_buffer_unordered(N)`
already gives I/O concurrency but inner-stream polling is single-threaded, so
Parquet decompression is serial - the spawn fixes that.
2. The `iceberg-datafusion` integration, which today hard-codes
`Partitioning::UnknownPartitioning(1)` in `IcebergTableScan` (see the TODO:
"to be replaced once we support output-partitioning"). Until that TODO lands,
DF sees a single-partition scan and can't shard FileScanTasks across
partitions per Liu's model - so the multi-concurrency branch is the only place
CPU parallelism can actually happen today. When output-partitioning lands,
engines will set concurrency==1 themselves and hit the #2020 fast path, just
like Comet - at which point the spawn path is cleanly dormant rather than
harmful.
Happy to split the tests into a follow-up or add a config flag to make the
spawn strictly opt-in, if either would help it land.
--
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]
