Repository navigation
Conversation
…n operators (apache#7231) * fix: declare required child distribution on aggregate, window and join operators AQE's OptimizeSkewInRebalancePartitions splits a skewed shuffle partition unless an operator above needs ClusteredDistribution, which ValidateRequirements checks. Comet's aggregate, window and shuffled join operators declared no requirement, so a REBALANCE(k) followed by GROUP BY k or PARTITION BY k returned one row per piece of a skewed key. Return the original Spark operator's requiredChildDistribution. * refactor: share the required-distribution override and cover window group limit Move the five identical requiredChildDistribution overrides into the CometDelegatesRequiredChildDistribution trait and mix it into CometWindowGroupLimitExec as well, whose Final mode needs its partition spec clustered. Extend the test with a window group limit and a project/filter below the aggregate, and drop its temp view afterwards. * test: isolate and assert each join in the skewed rebalance test Drop the aggregate above the join, which rejected the split on its own, and join the skewed side to a one-row-per-key side so the output stays small. Force the shuffled hash join explicitly and assert that the Comet join operator is in the plan. Each case fails when only that join's distribution delegation is removed. (cherry picked from commit ce3e665)
comphead
left a comment
comphead
left a comment
There was a problem hiding this comment.
Thanks @andygrove. PR Build (Linux) / Spark 3.4, JDK 17, Scala 2.12 [scans] fails here and passes on the other branch-1.2 backports opened today (#7489, #7509, #7512, #7513, #7514, #7515). I cannot open the job log (403), but I expect it is the failure #7231 caused on main: CometIcebergNativeSuite "AQE DPP - V2 broadcast query stage creation order (SPARK-34637)" fails with Column index 1 is out of bound (#7422). Spark 3.4 plans no exchange above a one-split scan, the final CometHashAggregate now asks for ClusteredDistribution, and the native Iceberg scan reports UnknownPartitioning(0), so AQE's second EnsureRequirements pass inserts an exchange inside an already serialized native block. branch-1.2 has that test without the skip #7423 added on main, and its scan still reports UnknownPartitioning(0).
The fix on main is #7424 (approved, set to merge once its checks pass). Could this backport wait for it and carry it as a second cherry-pick, or take the #7423 skip in the meantime? The Iceberg 1.11 shard-1 failure is a Gradle distribution download timeout, so a re-run should clear that one.
Which issue does this PR close?
Backport of #7231 to
branch-1.2. Closes #7208 on this branch.Rationale for this change
With AQE on,
REBALANCE(k)followed byGROUP BY kreturns duplicate groups when the data is skewed, and aPARTITION BY kwindow restartsrow_number()inside a key (#7208, a correctness bug). Comet's aggregate, window and join operators declared norequiredChildDistribution, so Spark'sOptimizeSkewInRebalancePartitionscould split a skewed partition under them.branch-1.2has the same operators and the same gap. See #7231 for the full analysis.What changes are included in this PR?
Cherry-pick of ce3e665 (
git cherry-pick -x) with no conflicts and no adaptations. The diff against the source commit is identical apart from line numbers and blob hashes. It addsCometDelegatesRequiredChildDistributionand mixes it into the Comet aggregate, window, window group limit and join operators.Other release branches:
branch-1.2is the newest release branch, so there is no newer one to port to. Whetherbranch-1.1needs a backport is not part of this PR.How are these changes tested?
On this branch, with JDK 17 and a native library built from
branch-1.2: the newCometExecSuitetest (AQE does not split a skewed rebalance partition under operators that need clustering) passes. I confirmed it fails with the four main-source files reverted tobranch-1.2's version, so the branch has the bug and the test catches it. I did not rerun the wider Comet suites that the source PR ran.