ulysses-you commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4071638339
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/ValidateRequirementsSuite.scala:
##########
@@ -158,4 +163,213 @@ class ValidateRequirementsSuite extends
SharedSparkSession {
testNestedJoin(Seq((2, 2), (1, 1)), Seq((2, 2)), Seq(5, 5, 5), false)
testNestedJoin(Seq((2, 2), (1, 1)), Seq((2, 5)), Seq(5, 5, 5), false)
}
+
+ test("SPARK-59671: a co-partitioning operator judges keyed children by their
pairing") {
+ // The sides of a storage-partitioned join aligned for skew repeat their
spread keys on
+ // purpose: neither satisfies a clustering on its own, yet the two key
sequences agree
+ // index by index, and the pairing is what the operator reads.
+ val a = AttributeReference("a", IntegerType)()
+ val b = AttributeReference("b", IntegerType)()
+ val rows = Seq(InternalRow(1), InternalRow(1), InternalRow(2))
+ val left = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(a),
rows))
+ val right = DummySparkPlan(outputPartitioning = KeyedPartitioning(Seq(b),
rows))
+ val join = ShuffledHashJoinExec(Seq(a), Seq(b), Inner, BuildLeft, None,
left, right)
+ assert(ValidateRequirements.validate(join),
+ s"aligned but ungrouped keyed sides pair, and that pairing is the
requirement:\n$join")
+
+ // The same key rows in another order do not pair: position by position is
the contract.
+ val off = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(1), InternalRow(2),
InternalRow(1))))
+ assert(!ValidateRequirements.validate(join.copy(right = off)),
+ "the same keys in a different order are not aligned")
+
+ // Nor does a keyed side with one that never pairs: a hashed side pairs
with neither the keys
+ // nor the layout of a keyed one.
+ val hashed = DummySparkPlan(outputPartitioning = HashPartitioning(Seq(b),
3))
+ assert(!ValidateRequirements.validate(join.copy(right = hashed)),
+ "a keyed side does not pair with a hashed one")
+
+ // And a pair each side satisfies on its own is still refused when the
sides do not line up,
+ // which is the operator's requirement: the pairing, not the per-side
answer.
+ val leftKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(a), Seq(InternalRow(1), InternalRow(2),
InternalRow(3))))
+ val rightKeys = DummySparkPlan(outputPartitioning =
+ KeyedPartitioning(Seq(b), Seq(InternalRow(4), InternalRow(5),
InternalRow(6))))
+
assert(leftKeys.outputPartitioning.satisfies(ClusteredDistribution(Seq(a))) &&
+ rightKeys.outputPartitioning.satisfies(ClusteredDistribution(Seq(b))),
+ "test setup: each side answers its own clustering")
+ assert(!ValidateRequirements.validate(ShuffledHashJoinExec(
+ Seq(a), Seq(b), Inner, BuildLeft, None, leftKeys, rightKeys)),
+ "the sides do not line up, and nothing else the operator reads says
otherwise")
+ }
+
+ test("SPARK-59671: a subset-keyed pair is judged on its own layouts") {
+ // With the subset permission on, the planner builds pairs whose sides are
grouped on a subset
Review Comment:
Thanks @cloud-fan! Corrected in ce59a03fc24, in the fixture and in the
helper text: the permission applies where the operation's keys are a subset of
the source's partitioning keys, so a `[a, b]` partitioning may be projected for
an `[a]` operation, and not the other way round.
The fixture now uses that direction: the sides report the projection's own
layout, which passes, and the same pair one step earlier, reporting the
source's keys that no node projected, is refused. The `ShuffleSpecSuite` arm
had the same wording and is corrected with it.
--
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]
