feat: allow Partitioning::Range to satisfy window Distribution::KeyPartitioned requirements#23416
feat: allow Partitioning::Range to satisfy window Distribution::KeyPartitioned requirements#23416mithuncy wants to merge 4 commits into
Conversation
github-actions
Bot
added
physical-expr
|
This is in my queue, just waiting for #23184 to land 🙇 |
mithuncy
force-pushed
the
issue-23289-window-range-on-23184
branch
from
3a4b586 to
f125c39
Compare
github-actions
Bot
removed
physical-expr
mithuncy
force-pushed
the
issue-23289-window-range-on-23184
branch
from
f125c39 to
2995df1
Compare
…rtitioned requirements Windows got a hash repartition even when the input was already range partitioned on the PARTITION BY keys. Opt WindowAggExec and BoundedWindowAggExec into range satisfaction for KeyPartitioned, the same way AggregateExec does. Incompatible keys still fall back to hash repartitioning, and windows without PARTITION BY keep requiring a single partition. Covered by enforce_distribution and sanity_checker tests plus a window section in range_partitioning.slt. Closes apache#23289
mithuncy
force-pushed
the
issue-23289-window-range-on-23184
branch
from
2995df1 to
a065d66
Compare
| ---- | ||
| physical_plan | ||
| 01)ProjectionExec: expr=[non_range_key@0 as non_range_key, sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW] | ||
| 02)--BoundedWindowAggExec: wdw=[sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "sum(range_partitioned.value) PARTITION BY [range_partitioned.non_range_key] ORDER BY [range_partitioned.value ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable Int64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted] |
There was a problem hiding this comment.
what do you think about doing one negative case for WindowAggExec
There was a problem hiding this comment.
added as TEST 22 right after the bounded negative
| set datafusion.optimizer.preserve_file_partitions = 0; | ||
|
|
||
|
|
||
| ########## |
There was a problem hiding this comment.
I think the coverage is great here, I do want to avoid this file from getting too noisy with checking file preserving and subset satisfaction for each operator though. What if we did something like test:
- Bounded window accepts exact Range(a) (test 19)
- General window accepts exact Range(a) (test 20)
- Range(a) does not satisfy PARTITION BY b, and a hash is added for bounded and unbounded (test 21 and new test)
- Safe range subset, Range(a) satisfies PARTITION BY (a, b) when subset is enabled (test 22)
- Subset disabled, the same query gets a hash exchange (test 24)
- No PARTITION BY, gets one logical window (test 28)
Then we should add a matrix test for these configurations as a unit / integration test inenforce_distribution.rsto ensure we are inserting repartitions as expeced according to these. I think this is fitting as those configs are tightly coupled with that file. I know @2010YOUY01 had some good guidance on test, would love your two cents here too 👍
There was a problem hiding this comment.
this is a non-blocker for this PR though, something I am taking note of
There was a problem hiding this comment.
Agreed on keeping this file lean. Will do either in this PR or as an immediate follow-up — whichever you and @2010YOUY01 prefer once he's had a look.
There was a problem hiding this comment.
I think lets keep fil lean in this PR and make follow up for the enforce_distribution.rs matrix tests 😄 . Lmk thoughts @2010YOUY01
| if self.partition_keys().is_empty() { | ||
| debug!("No partition defined for BoundedWindowAggExec!!!"); | ||
| vec![Distribution::SinglePartition] | ||
| crate::InputDistributionRequirements::new(vec![Distribution::SinglePartition]) |
There was a problem hiding this comment.
nit: can we also import InputDistributionRequirements
There was a problem hiding this comment.
Done — imported InputDistributionRequirements in both window execs
gene-bordegaray
left a comment
gene-bordegaray
left a comment
There was a problem hiding this comment.
looks good, thank you 😄
cc: @gabotechs
| set datafusion.optimizer.preserve_file_partitions = 0; | ||
|
|
||
|
|
||
| ########## |
There was a problem hiding this comment.
I think lets keep fil lean in this PR and make follow up for the enforce_distribution.rs matrix tests 😄 . Lmk thoughts @2010YOUY01
2 participants
Which issue does this PR close?
Partitioning::Rangeto satisfy windowDistribution::KeyPartitionedrequirements #23289.Rebased on main now that #23184 has merged. Overlaps with #23355, which also opts the window execs into range satisfaction as part of its PartitionedTopK work — happy to rebase whichever lands second.
What changes are included in this PR?
In one commit:
WindowAggExecandBoundedWindowAggExecwith partition keys into range satisfaction viaInputDistributionRequirements::allow_range_satisfaction_for_key_partitioning, mirroringAggregateExec. Compatible range-partitioned inputs then satisfy the window key requirement without a hash repartition; subset satisfaction and the hash fallback for incompatible keys come from the existing satisfaction machinery. Windows withoutPARTITION BYkeep requiring a single partition.Are these changes tested?
Yes: new
slttests inrange_partitioning.slt(exact and subset reuse, rehash on incompatible keys,subset_repartition_threshold/preserve_file_partitions/target_partitionsbehavior,WindowAggExecvia an unbounded frame, and no-PARTITION BY), plus plan-shape tests inenforce_distribution.rsand acceptance/rejection tests insanity_checker.rs.Are there any user-facing changes?
No: only plan changes.