Repository navigation
fix(scheduler): prune scan files by statistics and balance file groups - #2558
Conversation
DataFusion builds a file scan's file groups before any predicate is applied, so a selective filter on a column the files are clustered by leaves a few groups with all the surviving files and the rest empty. Add the BalanceFileGroups rule, which prunes files using the statistics collected during planning and deals the survivors round-robin over the scan's groups. It runs after planning in the static planner and during plan preparation in AQE, and is controlled by ballista.optimizer.balance_scan_file_groups.
github-actions
Bot
added
the
documentation
A single file with missing or inexact statistics made the batched pruning call return nothing for the column, so no file was pruned at all. Evaluate each file on its own and keep any file that cannot be judged.
The rule now reads ballista.optimizer.balance_scan_file_groups from the ConfigOptions it is given, so neither planner gates it. Also avoid cloning files that are about to be discarded and tidy the tests.
milenkovicm
left a comment
milenkovicm
left a comment
There was a problem hiding this comment.
LGTM, thanks @andygrove
villebro
left a comment
villebro
left a comment
There was a problem hiding this comment.
Looks great! As a long term fix, should we consider moving the pruning/rebalancing logic to DataFusion? It seems DF could benefit from this optimization independently of Ballista.
| /// Files without statistics are always kept. The number of file groups never | ||
| /// grows, and shrinks only when fewer files survive than there were groups. | ||
| #[derive(Debug, Default)] | ||
| pub struct BalanceFileGroups; |
There was a problem hiding this comment.
nit: As pruning is a big part of this, should this be called BalanceAndPruneFileGroups?
There was a problem hiding this comment.
Good call, renamed it to BalanceAndPruneFileGroups in 168b80b.
|
Thanks @villebro, good call. I filed apache/datafusion#26130 to prune files by statistics in DataFusion mostly hides this skew in a single process because idle partitions steal files from their siblings. We can't rely on that in Ballista since each partition runs as its own task. If the upstream change lands, DataFusion's existing file splitting would spread the remaining files and we could drop this rule. Until then this rule covers it on our side. |
…roups The rule prunes scan files by statistics as well as rebalancing them, so name it after both, as suggested in review.
Oh interesting, I wasn't aware of this file stealing detail. TIL 👍 |
|
FWIW, with MPTs, we get whatever functionality they add for free. |
Which issue does this PR close?
Closes #2551.
Rationale for this change
DataFusion builds a file scan's file groups before any predicate is applied: files are sorted by path and cut into contiguous runs. Pruning from Parquet statistics only happens later, inside each task. When the files are clustered by the filtered column (for example Hive-style date directories filtered by a date range), every file that survives pruning sits in a few neighbouring groups. Only the tasks that own those groups have rows to read, the other tasks open their files and skip them, and the stage takes as long as the busiest few tasks.
What changes are included in this PR?
BalanceAndPruneFileGroupsphysical optimizer rule (ballista/scheduler/src/physical_optimizer/file_group_balance.rs). It evaluates the scan's predicate against the file statistics collected during planning, drops files that cannot match, and deals the remaining files round-robin over the scan's file groups.FilterPushdown(so the predicate is on the scan and DataFusion's file scan repartitioning sees the pruned files). Physical plans submitted directly by a client are not rewritten.ballista.optimizer.balance_scan_file_groups(defaulttrue) to turn it off, with the generatedconfigs.mdupdated.Tests: unit tests for the rule, including a case that plans a real
ListingTableover Parquet files clustered by the filtered column.Are there any user-facing changes?
A new config option,
ballista.optimizer.balance_scan_file_groups. Scans with selective predicates on clustered columns may now plan fewer, better balanced file groups. No public API changes.