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.
…one (apache#7424) * fix: report SinglePartition from a native Iceberg scan that replaces one On Spark 3.4, BatchScanExec reports SinglePartition for a scan of one input partition, and Spark plans no exchange above it. The native Iceberg scan reported UnknownPartitioning(0). Since the aggregates declare their required distribution (apache#7231), EnsureRequirements, which AQE runs again over a DPP subquery's converted plan, inserted an exchange inside a native block that was already serialized, and the native scan lost its tasks. Report SinglePartition when the replaced scan does. * test: run the SPARK-34637 Iceberg DPP test on Spark 3.4 again (cherry picked from commit 891008e)
Which issue does this PR close?
Backport of #7231 and its follow-up #7424 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.Once the aggregates declare their distribution, Spark 3.4 breaks
CometIcebergNativeSuite"AQE DPP - V2 broadcast query stage creation order (SPARK-34637)" (#7422). Spark 3.4 plans no exchange above a one-split scan, but the native Iceberg scan reportedUnknownPartitioning(0), so AQE's secondEnsureRequirementspass inserted an exchange inside an already serialized native block. #7424 fixes that by reportingSinglePartitionwhen the replaced scan does, so it goes in with this change.What changes are included in this PR?
Two cherry-picks, both with
git cherry-pick -xand no conflicts:CometDelegatesRequiredChildDistributionand mixes it into the Comet aggregate, window, window group limit and join operators. The diff against the source commit is identical apart from line numbers and blob hashes.SinglePartitionfromCometIcebergNativeScanExecwhen the replaced scan does. Only the main-source change applies. The test change in fix: report SinglePartition from a native Iceberg scan that replaces one #7424 removes a Spark 3.4 skip thatbranch-1.2never had, and the test here already has the guard that fix: report SinglePartition from a native Iceberg scan that replaces one #7424 leaves in place.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. The #7424 commit compiles (./mvnw test-compile); the Spark 3.4 run of the SPARK-34637 test is covered by this PR'sSpark 3.4 [scans]job. I did not rerun the wider Comet suites that the source PRs ran.