Repository navigation
fix: keep aggregates above undecided joins partial and final - #2806
Merged
Merged
Conversation
An undecided AQE join (DynamicJoinSelectionExec) reported the partitioning of the join it was planned as. A CollectLeft join passes its right input's partitioning through, so a join planned with a one-partition table such as nation on the right reported one partition, although the join it resolves into may build nation and output the other side's partitions. Meanwhile DataFusion's CombinePartialFinal- Aggregate saw an aggregate over one partition and merged it into a Single aggregate, which later replans cannot split. Once the join resolved, the aggregate's input was coalesced and exchanged below it, so every task of the consuming stage read and aggregated the whole input: TPC-H q11's scalar sum read all 32M rows in each of 16 tasks at SF1000 with 256 partitions. Report at least as many partitions as the widest input until the join is decided.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
No issue. Split out of #2792.
Rationale for this change
Until AQE decides an undecided join (
DynamicJoinSelectionExec), the join reports the output partitioning of the join it was planned as. ACollectLefthash join passes its right input's partitioning through. So a join that was planned with a one-partition table on the right, such asnation, reports one output partition, even though the join it resolves into may build that small table and output the other side's partitions.DataFusion's
CombinePartialFinalAggregateruns during each replan. It sees an aggregate over what claims to be one partition and merges thePartialandFinalpair into a singleSingleaggregate, which later replans cannot split back apart. Once the join resolves, the aggregate's input is coalesced and exchanged below it, and every task of the consuming stage reads and aggregates the whole input.In TPC-H q11 this hits the scalar
sumsubquery. At SF1000 with 256 partitions, each of 16 tasks read all 32M input rows, and the query took 14.7 s. With this change it takes 5.4 s (Spark + Comet on the same cluster: 43.7 s). The behavior is onmaintoday.What changes are included in this PR?
DynamicJoinSelectionExecreportsUnknownPartitioningwith the larger of its inputs' partition counts when the planned join reports fewer, both when it is built (from_hash_join,from_sort_join) and inwith_new_children. Partial and final aggregates above it therefore stay split. Once the join is decided, the real join's properties take over.undecided_join_reports_at_least_its_widest_input: aCollectLeftjoin with a one-partition right input must report the wider left input's partition count, including afterreplace_children.Are there any user-facing changes?
No. Only plans with an aggregate above an undecided join over a one-partition input change: the aggregate stays partial and final instead of becoming single.