DataFusion 54: DataSourceExec shared work-queue makes each Ballista scan task read the whole table (wrong results + ~Nx slowdown)

[andygrove]

Describe the bug

After upgrading to DataFusion 54 (PR #1906), distributed TPC-H execution is dramatically slower. CI's TPC-H SF10 job went from ~12 min on main to ~52 min and then failed with No space left on device. Example per-query (CI, SF10, prefer_hash_join=false):

query	main (DF53)	DF54 branch
q1	2.95 s	40 s
q2	3.38 s	600 s
q3	5.48 s	143 s
q5	15.8 s	died (disk full)

This reproduces locally and is well-characterized below. Single-node DataFusion 54 is fast, so the regression is in Ballista's distributed execution path.

To Reproduce

SF10 parquet, 1 scheduler + 1 executor (-c 8), release build:

tpch benchmark ballista --query 1 --partitions 16 --iterations 1 -c datafusion.optimizer.prefer_hash_join=false

The core finding: tasks queued behind the first wave run ~8× slower

With an executor of N task slots, the first N tasks (wave 1) run fast; every subsequent wave runs ~8× slower for identical work.

executor slots	partitions	waves	q1 time
8	8	1	0.50 s
8	16	2	4.2 s
8	24	3	8.0 s
16	16	1	0.30 s
16	32	2	4.0 s

• Single-node DataFusion 54 (same data, 16 target partitions): 0.21 s.
• DataFusion 53 (same Ballista commit), distributed, q1 @16part/8slots: 0.56 s (no wave penalty).
• It resets per query and is purely timing/wave-dependent, not partition-index-dependent (with 16 slots, partitions 8–15 run in wave 1 and are fast).

Per-task timing (q1 @16part/8slots): wave-1 tasks ~0.45 s each; wave-2 tasks start immediately after wave 1 (no dispatch gap) but take ~3.6 s each of compute time.

What the slow tasks are doing

A CPU sample during the slow wave shows all 8 worker threads busy (not parked) in:
execute_query_stage → GroupedHashAggregateStream → ProjectionStream → FilterExecStream → CooperativeStream → FileStream → ParquetRecordBatchReader → snappy/RLE decode.

So later-wave tasks are fully parallel and CPU-bound, but the same parquet decode is ~8× slower than in wave 1 for the same rows/task. The merged scan stage metrics also report very low scan_efficiency_ratio (~6%) and inflated output_rows/files_opened that scale with partition count (may be partly a metric-merge artifact, but the wall-clock slowdown is real).

Ruled out

All tested with config propagation working (after merging the fix for client datafusion.* overrides):

• Stripping CooperativeExec from the executed plan — no effect.
• tokio version — unchanged (1.52.3) between DF53 and DF54 branches.
• Executor worker-thread priority (the DedicatedExecutor low priority) — no effect.
• Worker thread count (×8) — no effect.
• Sort-based vs hash-based shuffle — identical.
• datafusion.optimizer.repartition_file_scans=false — no effect (didn't change the plan).
• datafusion.optimizer.repartition_file_min_size huge (whole-file groups, no byte-range splitting) — made it much worse (51 s), so byte-range splitting is helping, not hurting.
• datafusion.execution.parquet.pushdown_filters=true — no effect.
• Sharing one RuntimeEnv across tasks (so the DF54 FileMetadataCache persists instead of being fresh per task) — no effect.

Caveat: a test bypassing the DedicatedExecutor was inadvertently applied to the pull/poll path while the cluster ran PushStaged, so "runtime bypass" is still inconclusive.

Question / context

Has anyone seen DataFusion 54 parquet/scan behavior where tasks run after an initial batch decode the same data significantly slower on a shared process? (We hit DF54 parquet footer/metadata-caching changes when upgrading Comet.) Any pointers on what changed in 54's parquet scan or runtime that could cause later, fully-parallel, CPU-bound scan tasks to be ~8× slower would help. Pinning this likely needs a DF53↔54 scan-level comparison or CPU-counter profiling.

Blocks #1906.

[andygrove]

Update — root cause is data amplification, not slow decode.

Measuring the partial-aggregate input rows (reduction_factor metric, which is reliable, unlike the merged scan output_rows):

q1 partitions	slots	input rows to partial aggregate
8	8 (all wave 1)	59.1 M (≈ table, correct)
9	8 (8 wave-1 + 1 wave-2)	118.3 M

Going from 8→9 partitions added one task that runs in the second wave, and total processed rows jumped by ~59 M — i.e. that single second-wave task scanned the entire table (~59 M rows) instead of its ~1/9 file-range partition.

So: tasks that run after the first wave scan the whole dataset instead of their assigned file range. Key properties:

• It is wave-/slot-dependent, not partition-index-dependent: with -c16, all 16 partitions of q1 --partitions 16 run in one wave and the query is fast (0.3 s, ~59 M rows total); with -c8 the same plan runs in two waves and the second-wave tasks each re-scan the whole table (4.2 s). Same serialized plan, same file groups — only the number of concurrent slots (waves) differs.
• It resets per query (the first wave is always correct/fast), so it is tied to per-stage execution state, not process-global accumulation.
• CPU profile of a single slow task is dominated by aggregate accumulate + group-by-on-string-columns + snappy decode — consistent with simply processing ~8× the rows.

This explains the earlier symptoms (≈N× scan inefficiency, disk-full on join queries from shuffling N× data). It looks like later-wave tasks fail to restrict the parquet scan to their file_group's byte range and read all row groups — possibly related to the DF54 parquet footer/metadata handling (we hit metadata-cache changes upgrading Comet). Repro: tpch benchmark ballista --query 1 --partitions 9 --iterations 1 -c datafusion.optimizer.prefer_hash_join=false on an 8-slot executor.

[andygrove]

Root cause found — and it's a correctness bug, not (just) a decode slowdown

The earlier "same rows/task but ~8× slower decode" reading was a misdiagnosis. The real cause is that each scan task reads the entire table instead of its own file group, so later-wave tasks do ~N× the work and produce ~N× inflated results. The slowdown is a side effect of the over-read.

What actually happens

DataFusion 54 changed DataSourceExec to hand file groups to partition streams from a shared work-queue (open_with_args + a per-execution SharedWorkSource; create_sibling_state in FileScanConfig). The queue is only divided across partitions when all partitions of one plan instance are polled concurrently.

Ballista executes one partition per task on its own decoded plan instance (ShuffleWriterExec/SortShuffleWriterExec calls child.execute(input_partition) for a single partition). A single partition polled in isolation drains the entire work-queue → it scans the whole table.

• Wave 1 is dispatched as one LaunchMultiTask batch sharing one decoded plan instance; its partitions run concurrently and cooperatively split the file groups → table read once → fast and correct.
• Each later task is dispatched individually as a slot frees → its own plan instance → drains all file groups alone → whole-table scan → ~N× CPU and N× rows.

Correctness impact

q1 SF10, 16 partitions, prefer_hash_join=false:

executor slots	waves	count_order (A/F)	correct?
16	1	14,804,077	✅
8	2	118,432,616	❌ (exactly 8×)

Total rows counted on -c8 = 473,140,872 = 8 × the table. Averages still match (8×sum / 8×count cancels), which is why it can pass spot checks; the benchmark harness only validates results at SF1.

Minimal standalone reproducer (pure DataFusion 54, no Ballista)

Register a multi-file parquet table with target_partitions = N, build a physical plan, then:

// TEST 1: poll all N partitions concurrently on ONE instance
//   -> each reads ~1/N of the rows; sum == table row count.   (correct)
// TEST 2: poll a single partition alone on its own instance
//   -> that one partition reads the ENTIRE table.             (the bug)
let stream = plan.execute(p, ctx.task_ctx())?;   // for a lone p, reads everything

Observed on 16 files (TPC-H lineitem SF10): TEST 1 → 16/16 partitions each ~3.7M rows, sum 59,142,609; TEST 2 → execute(0), execute(8), execute(15) each return 59,142,609 (the whole table).

This is intrinsic to FileScanConfig under DF54 (scheduling_type() is hardcoded Cooperative; repartition_file_scans=false does not change it). It worked on DF53, where DataSourceExec::execute(p) read file group p statically.

Fix that works

In the executor, restrict each task's DataSourceExec to the file group for its partition_id before execution (other group slots emptied, partition count preserved), so the lone execute(partition_id) reads only its group. With this change, q1 -c8 is correct (A/F 14,804,077) and back to ~0.50 s, and q1–q14 all return canonical TPC-H row counts. Guard: skip when partition_id >= file_groups.len() (an operator between the scan and the stage output changed the partition count). PR to follow.