Repository navigation
Byte-balanced file-scan scaling can create severe per-partition file-count skew #759
Description
Description
The default FileScanConfig scaler distributes scan work by bytes while consuming files in order. When many small files occur together, one output partition can receive hundreds of file entries even though its assigned bytes are similar to those of other partitions.
Each scan partition processes its file entries sequentially. If sibling file-stream work stealing is disabled—as it must be in our distributed-task setup to keep tasks within their assigned files—the file-heavy partition becomes a long tail while its siblings finish early.
Evidence
On a Parquet scan with 21,888 source files, the scaler produced 40 tasks with eight scan partitions each. One task’s partitions contained:
[559, 58, 68, 56, 65, 90, 55, 65] file-range entries
In a three-run, interleaved scan probe of that task alone:
| Layout | Median runtime |
|---|---|
| Original byte-balanced groups | 454.3s |
| Same file entries and byte ranges redistributed evenly across its eight partitions | 117.2s |
| Original groups with sibling work stealing enabled | 108.2s |
Redistribution changed neither the task’s input nor its output: all runs read the same 10,509,432 rows and produced the same seven aggregate sums. Removing full-span range annotations while leaving the skew in place did not recover performance (467.0s median). The 559-entry partition took approximately 454s despite reading fewer rows than any of its siblings.
We observed an associated end-to-end TPC-DS Q80 regression after upgrading to datafusion-distributed 3.0.0. A local whole-file, round-robin scaling workaround recovered that query’s performance.
Expected behavior
For scans without an ordering or partitioning constraint that prevents redistribution, scaling should account for per-file overhead as well as bytes, so one sequential scan partition does not receive vastly more files than its peers. Existing partitioning and ordering guarantees must be preserved.
Related work
#310 covers other correctness and parallelism concerns with the default estimator; #449 covers unused scan metadata. PR #496 introduced the byte-based scaling involved here. None appears to track this file-count long tail specifically.
Activity
gabotechs commentedon Oct 2, 2026
🤔 Interesting, I don't think we've ever tested this project reading so many parquet files.
This project uses upstream's
FileGroupPartitionerstruct for re-partitioning files, which I think it's just byte-based without taking into account number of files.I'd be curious in how did you manage to solve this in your system, did you implemented your own
ScaleUpLeafNodeHandlerwith a custom re-distribution algorithm? Maybe this is even worth raising in upstreamdatafusion.sesteves commentedon Oct 2, 2026
Yes. We added a custom
ScaleUpLeafNodeEventhandler for eligible set-op scans (inUNION,INTERSECT, orEXCEPTqueries). It spreads complete files across tasks and their parallel scan streams while preserving the scan partition count. Separately, redistributing the same input across one task’s streams cut median scan time from 454s to 117s.I agree DataFusion’s
FileGroupPartitionercould more generally account for file count alongside bytes while preserving ordering and partitioning guarantees.gabotechs commentedon Oct 2, 2026
I'm confused by this though, in 2.0.0 we were also relying on
FileGroupPartitioner, essentially with the same repartitioning algorithm than in 3.0.0:https://github.com/datafusion-contrib/datafusion-distributed/blob/v2.0.0/src/distributed_planner/task_estimator.rs#L288
So I'd not have expected this to be a regression introduced from 2.0.0 to 3.0.0
sesteves commentedon Oct 2, 2026
We were upgrading from a pin just before v2.0.0 (
b4856ef), which still used whole-file round-robin rebalancing. The byte-basedFileGroupPartitionerscaling came in with #496, three commits later, so this isn't a 2.0.0 → 3.0.0 regression. It's the change from #496 that we picked up as part of the 3.0.0 upgrade.gabotechs commentedon Oct 3, 2026
That makes sense. Adding a general-purpose rule that weights both byte-size and file-count might be non trivial to do... I honestly do not have a recommendation for something that works fine in both cases.
For now maintaining your own file-count based
ScaleUpLeafNodeHandleris probably the best choice.