cloud-fan commented on code in PR #58942:
URL: https://github.com/apache/spark/pull/58942#discussion_r4068736081
##########
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:
**Nit (P3):** This describes the subset in the opposite direction from the
configuration contract. `allowKeysSubsetOfPartitionKeys` permits the operation
keys to be a subset of the source partition keys, so an `[a, b]` partitioning
may be projected for an `[a]` operation. It does not make the planner build an
`[a]` partitioning for an `[a, b]` operation; with
`requireAllClusterKeysForCoPartition` at its default, that shape is rejected.
Please update this fixture comment and the matching helper text to use the
operation-subset-of-source direction.
--
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]
