[SPARK-58138][SQL] Preserve child output partitioning through BIN BY#57275
[SPARK-58138][SQL] Preserve child output partitioning through BIN BY#57275vranes wants to merge 1 commit into
Conversation
e2844ca to
0e93786
Compare
cloud-fan
left a comment
cloud-fan
left a comment
There was a problem hiding this comment.
0 blocking, 0 non-blocking, 0 nits.
Clean. BinByExec reuses PartitioningPreservingUnaryExecNode with the same outputExpressions = output idiom as ProjectExec and the aggregate execs, so the pass-through/scaled distinction falls out of the trait's output-set filter rather than needing special-casing.
No findings.
Verification
Traced the load-bearing correctness claim: a preserved HashPartitioning still holds across BIN BY's 1:N row multiplication. BIN BY runs under mapPartitionsInternal (never moves rows across partitions), and pass-through column values are identical across an input row's emitted sub-rows, so co-location on a pass-through key is preserved. The scaled-DISTRIBUTE case is correctly dropped: BinByResolution.scaledDistributeAttributes mints a fresh ExprId, so the child partitioning keyed on the original (pre-scale) column is absent from the output set and filtered out. The mixed case drops whole. All three are covered by the added tests, which reproduce the fresh-ExprId minting faithfully and assert on partitioning metadata directly.
### What changes were proposed in this pull request? This PR makes `BinByExec` extend `PartitioningPreservingUnaryExecNode` so it reports an `outputPartitioning` derived from its child. A child partitioning is kept only when its keys are in the operator's output: a partitioning on a pass-through column survives, while one on a scaled `DISTRIBUTE` column is dropped (that column has a fresh `ExprId` absent from the output). Keeping the pass-through case is safe because `BIN BY` transforms each partition independently and pass-through columns hold the same value across an input row's sub-rows. ### Why are the changes needed? A downstream operator clustered on a pass-through column that the child already partitions on would otherwise get an unnecessary shuffle. ### Does this PR introduce _any_ user-facing change? No. `BIN BY` is gated off by default (`spark.sql.binByRelationOperator.enabled`). ### How was this patch tested? `ProjectedOrderingAndPartitioningSuite`: a pass-through key is preserved as `HashPartitioning`, a scaled `DISTRIBUTE` key is dropped to `UnknownPartitioning`, and a mixed key is dropped whole. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Anthropic) Closes #57275 from vranes/bin-by-output-partitioning. Authored-by: Nikolina Vraneš <nikolina.vranes@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> (cherry picked from commit 5294834) Signed-off-by: Wenchen Fan <wenchen@databricks.com>
2 participants
What changes were proposed in this pull request?
This PR makes
BinByExecextendPartitioningPreservingUnaryExecNodeso it reports anoutputPartitioningderived from its child.A child partitioning is kept only when its keys are in the operator's output: a partitioning on a pass-through column survives, while one on a scaled
DISTRIBUTEcolumn is dropped (that column has a freshExprIdabsent from the output). Keeping the pass-through case is safe becauseBIN BYtransforms each partition independently and pass-through columns hold the same value across an input row's sub-rows.Why are the changes needed?
A downstream operator clustered on a pass-through column that the child already partitions on would otherwise get an unnecessary shuffle.
Does this PR introduce any user-facing change?
No.
BIN BYis gated off by default (spark.sql.binByRelationOperator.enabled).How was this patch tested?
ProjectedOrderingAndPartitioningSuite: a pass-through key is preserved asHashPartitioning, a scaledDISTRIBUTEkey is dropped toUnknownPartitioning, and a mixed key is dropped whole.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic)