dongjoon-hyun commented on code in PR #58814:
URL: https://github.com/apache/spark/pull/58814#discussion_r4017125258
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1767,10 +1771,11 @@ case class IdentityReducer(transform:
TransformExpression) extends Reducer[Any,
* the child reports, and a consumer needs no
`GroupPartitionsExec` to
* produce it. Only
`v2BucketingAllowKeysSubsetOfPartitionKeys` projects at
* all, and it reaches `None` two ways: an identity
projection over already
- * grouped and sorted keys rebuilds the same
partitioning, and a narrowing
- * one over a marked claim is refused outright. Only
the first says the
- * child is grouped on the operation keys; the second
leaves a spec that
- * `areKeysCompatible` and `canCreatePartitioning`
both turn away. See
+ * grouped and sorted keys rebuilds the same
partitioning, and a marked
+ * claim is refused outright, since neither narrowing
nor sorting its keys
+ * leaves the routing it promises for its undeclared
rows. Both say the
+ * child is grouped on the operation keys as it
stands, and both leave a
Review Comment:
This holds for the identity projection this PR newly admits, but not for a
marked narrowing layout (partition expressions `[a, b]`, clustering `[a]`).
There the unprojected spec has an empty `keyPositions` entry, so
`canCreatePartitioning` (`keyPositions.forall(_.nonEmpty)`, whose own comment
still says it exists for exactly this case) and `areKeysCompatible` both turn
it away, and `keysSatisfy` does not call that child grouped on the operation
keys. `ShuffleSpecSuite` "a marked narrowing projection yields an unusable
spec" still pins that. The removed sentence was the accurate one for that case.
Could you split the two cases here and in the `createShuffleSpec` comment at
L920 ("The unprojected spec that comes back is still usable")? No behavioural
impact: every caller gates on `satisfies` first, so a narrowing marked spec is
only reachable by calling `createShuffleSpec` directly.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -794,7 +793,47 @@ case class EnsureRequirements(
rightReducers, distributePartitions = applyPartialClustering &&
!replicateRightSide)
}
- if (compatibleAsIs || pushCommonValues) Some(Seq(newLeft, newRight)) else
None
+ // The pairing is only worth committing to if both children still declare
the same aligned key
+ // sequence once the grouping has been pushed into them. They can fail
that. A
+ // `GroupPartitionsExec` gives up its keyed claim when it turns out to
regroup a layout that
+ // pins undeclared rows to `hash(key) % numPartitions` (see
+ // `KeyLayout.mayContainUnknownPartitionKeys`), and only the node knows
the permutation it
+ // performs, so that answer arrives after the pairing was chosen. Asking
before returning is
+ // what keeps the join from skipping both shuffles for a child that no
longer satisfies its
+ // distribution, which is a plan `ValidateRequirements` rejects and every
AQE rule that needs a
+ // valid plan then refuses to touch.
+ //
+ // The check is pairwise, not a per-side `satisfies`. Partially clustered
distribution leaves
+ // both children value-aligned yet not grouped on purpose, so a per-side
gate would refuse that
+ // whole family. What both sides owe each other is the key sequence
`alignToExpectedKeys`
+ // guarantees, each key repeated as many times as the merge expects,
whichever side replicates.
+ // Through `KeyLayout.describesSameKeys`, which carries the reason the key
types are compared as
Review Comment:
nit: this is a sentence fragment. Also, L817 argues against re-asking on the
`compatibleAsIs` path because it "would walk both sides' partition keys", but
`describesSameKeys` on the push path does exactly that walk, so the argument
reads oddly. The stronger reason is the one already given: on that path the
children are the ones the pairing read.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -8247,13 +8247,22 @@ class KeyGroupedPartitioningSuite
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
val df = sql(query)
checkAnswer(df, expected)
- // The regrouped marked side can no longer claim the hash-routing
contract, so the final
- // join re-shuffles it instead of storage-partitioning. Three one-side
shuffles, all keyed
- // with unknown partition keys: rt onto the union, rt2 onto ra2, and
the final join's
- // re-shuffle of the regrouped side. Before the fix the final join
storage-partitioned
- // (only the first two shuffles) and silently lost the id=5 row.
+ // The regrouped marked side can no longer claim the hash-routing
contract. Four one-side
+ // shuffles, all keyed with unknown partition keys: rt onto the union,
rs onto the marked
+ // side once the second join declines, rt2 onto ra2, and the final
join's re-shuffle of the
+ // side that was regrouped. Before SPARK-59050 the final join
storage-partitioned (only the
+ // first, third and fourth shuffles) and silently lost the id=5 row.
Review Comment:
This parenthetical contradicts the sentence it is in. The fourth shuffle is
the final join's own, so it cannot have existed while that join
storage-partitioned, and the second shuffle (`rs`) is the one this PR adds.
Before SPARK-59050 the plan had only the first and third shuffles in this
numbering, which is what the previous wording ("only the first two shuffles" of
three) said.
Also, once the second join declines, its tried `GroupPartitionsExec` is
discarded, so no side reaches the final join regrouped: the fourth shuffle is
the final join's one-side shuffle of `b` onto `r2`'s key order.
Suggestion: `(only the first and third shuffles)`, and `the final join's
one-side shuffle` instead of `the final join's re-shuffle of the side that was
regrouped`.
--
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]
