peter-toth commented on code in PR #59863:
URL: https://github.com/apache/spark/pull/59863#discussion_r4441033307
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -114,9 +114,10 @@ case class GroupPartitionsExec(
* `plannedPartitioning` rebuilds its keyed members on top of the grouping.
A rule that hands this
* node a child reporting something else invalidates both, so the keyed
claim is dropped and only
* the physical output count is reported. `ValidateRequirements` then finds
the parent's
- * distribution unsatisfied and `AdaptiveSparkPlanExec` reverts the rewrite,
which is what an
- * `AQEShuffleReadExec` landing over a keyed shuffle stage relies on. Only
an `AQEShuffleReadRule`
- * is validated that way, so `checkChildStillMatches` covers the rest at
execution.
+ * distribution unsatisfied and `AdaptiveSparkPlanExec` reverts the rewrite.
A coalesced or
+ * skewed `AQEShuffleReadExec` over a keyed stage also reports
`UnknownPartitioning` itself, so
+ * validation can reject it even without this node. Only an
`AQEShuffleReadRule` is validated
+ * that way, so `checkChildStillMatches` covers the rest at execution.
Review Comment:
**Finding 13.** The skewed half of this doesn't hold:
- A skew split comes from `OptimizeSkewedJoin`, which is not an
`AQEShuffleReadRule`.
- It marks the join `isSkewJoin = true`, and the join then requires
`UnspecifiedDistribution` (`ShuffledJoin.scala:57-61`).
- So the rule's own validation (`OptimizeSkewedJoin.scala:259-263`) accepts
the skew read. That is correct, since the split still pairs partition i with
partition i.
Also, "`checkChildStillMatches` covers the rest" now follows "even without
this node", but a side without the node has no execution-time check. For
example:
```suggestion
* distribution unsatisfied and `AdaptiveSparkPlanExec` reverts the
rewrite. A coalesced
* `AQEShuffleReadExec` over a keyed stage also reports
`UnknownPartitioning` itself, so
* validation can reject it even without this node. Only an
`AQEShuffleReadRule` is validated
* that way, so `checkChildStillMatches` covers the rest at execution. A
join side whose identity
* node `EnsureRequirements` skipped has no such check. It relies on its
shuffle stages' fixed
* partitioning instead.
```
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -5416,6 +5553,79 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-60618: partially clustered join keeps its replicate-side choice
when the " +
+ "alignment is the identity") {
+ // The smaller side by statistics is replicated: `pc_small` holds five
splits in 200 bytes,
+ // `pc_large` three splits in 216. Both are bucketed, whose transform
derives no ordering, so
+ // the join's sort has to be added. The larger side ends up with one split
per bucket, so the
+ // alignment on it is the identity and the pass-through node has to stay:
a re-run peels a
+ // local sort only through a `GroupPartitionsExec`, and reading the
statistics from the scans
+ // again is what keeps the choice. Without the node the counts (five
against three) pick the
+ // other side.
Review Comment:
**Finding 12.** Since 5346f9f the skip needs every leaf to be a shuffle
stage. Both sides here are scans, so the leaf check alone keeps both nodes.
This test no longer reaches the replicate-side gate this comment describes. It
passes on master, and also with `!replicateSideChosen` removed from
`canSkipIdentityGroup`.
No planned tree reaches that gate now. With partial clustering on,
`KeyedShuffleSpec.canCreatePartitioning` (`partitioning.scala:2549`) plans no
keyed shuffle. So no side with only stage leaves gets to the push branch. I ran
the skip test's `items`/`purchases` shape with partial clustering on. Inner,
left, right and full outer joins all plan two hash shuffles and no
`GroupPartitionsExec`. Only the hand-built stage pair in
`EnsureRequirementsSuite` reaches the gate.
Keeping the gate as a guard against a later change there is fine. Could its
comment at `EnsureRequirements.scala:934-935` say that no planned tree reaches
it today? For this test, either:
- drop it, since the SPARK-58996 test above (L5513) already covers a re-run
that keeps its replicate-side choice, or
- say what keeps the nodes now:
```suggestion
// The smaller side by statistics is replicated: `pc_small` holds five
splits in 200 bytes,
// `pc_large` three splits in 216. Both are bucketed, whose transform
derives no ordering, so
// the join's sort has to be added. The larger side ends up with one
split per bucket, so the
// alignment on it is the identity. Its leaf is a scan, so the node
stays, and a re-run peels
// the local sort through it and reads the statistics from the scans
again. Without the node
// the counts (five against three) would pick the other side.
```
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1152,12 +1152,149 @@ class KeyGroupedPartitioningSuite
testWithCustomersAndOrders(customers_partitions, orders_partitions, 2, 0)
}
+ test("SPARK-60618: scan rebuilt with different partition keys after
alignment") {
Review Comment:
**Finding 14.** Nit: this test and the skip test below sit inside the
`partitioned join:` group that runs `testWithCustomersAndOrders`. This one
checks the same guard as "SPARK-59252: a planned scan keeps the partitioning it
was planned with" (L2231), so it fits next to that one. The skip test is a
one-side keyed shuffle case, so it reads better next to the `SPARK-41471:
shuffle one side:` tests (L3088 onward).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
