This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new f520bfcf9022 [SPARK-59688][SQL][FOLLOWUP] Cover a left outer join in
the coinciding-partition-keys test
f520bfcf9022 is described below
commit f520bfcf90226a06cc75d625c07d25742328b090
Author: naveenp2708 <[email protected]>
AuthorDate: Fri Oct 2 13:31:40 2026 -0700
[SPARK-59688][SQL][FOLLOWUP] Cover a left outer join in the
coinciding-partition-keys test
**What changes were proposed in this pull request?**
A follow-up test for SPARK-59688. It parameterizes the existing end-to-end
test over an inner and a LEFT OUTER join, on the same identity(id) vs
flip_low_bit(id) setup. Test only, no production change.
**Why are the changes needed?**
The end-to-end test in SPARK-59688 covers an inner join, where the wrong
result is missing rows (0 of 2). An outer join hits the same coinciding-keys
bug with a different symptom: it keeps the left rows and null-extends them,
returning (0, 'a', null) instead of (0, 'a', 'x'). Every left row here has a
match, so both join types share the expected rows, the plan assertions, and the
test body. The assertions fail on a revert of the fix and pass with it.
**Does this PR introduce any user-facing change?**
No. Test only.
**How was this patch tested?**
The parameterized test. It fails on a revert of the partitioning.scala fix
and passes with it. KeyGroupedPartitioningSuite runs green (195 tests).
**Was this patch authored or co-authored using generative AI tooling?**
No
Closes #59129 from naveenp2708/SPARK-59688-outer-join-test.
Authored-by: naveenp2708 <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit 099c3ee2e2e71ef8c826ee06d35081cb6d82b29c)
Signed-off-by: Szehon Ho <[email protected]>
---
.../connector/KeyGroupedPartitioningSuite.scala | 41 +++++++++++++---------
1 file changed, 25 insertions(+), 16 deletions(-)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
index 488f6c6a2d40..03b493b96727 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
@@ -1410,6 +1410,9 @@ class KeyGroupedPartitioningSuite
// report the same partition key list, [0, 1] of LongType, while the
rows behind a key differ:
// the identity side's key 0 holds id 0, the transform side's holds id
1. The identity side's
// raw keys have to be reduced onto `flip_low_bit` before the partitions
can be paired up.
+ // Read as they stand, an inner join drops the mispaired rows (0 of 2)
and a left outer join
+ // null-extends them; the reduce fixes both. Every left row here has a
match, so the two join
+ // types expect the same rows.
val cols = Array(Column.create("id", LongType), Column.create("data",
StringType))
createTable("t1", cols, Array(identity("id")))
sql("INSERT INTO testcat.ns.t1 VALUES (0, 'a'), (1, 'b')")
@@ -1417,23 +1420,29 @@ class KeyGroupedPartitioningSuite
createTable("t2", cols, Array(Expressions.apply("flip_low_bit",
Expressions.column("id"))))
sql("INSERT INTO testcat.ns.t2 VALUES (0, 'x'), (1, 'y')")
- val df = sql(
- "SELECT t1.id, t1.data, t2.data FROM testcat.ns.t1 JOIN testcat.ns.t2
ON t1.id = t2.id")
+ val expected = Seq(Row(0L, "a", "x"), Row(1L, "b", "y"))
+ Seq("JOIN", "LEFT OUTER JOIN").foreach { joinType =>
+ val df = sql(
+ s"SELECT t1.id, t1.data, t2.data FROM testcat.ns.t1 $joinType
testcat.ns.t2 " +
+ "ON t1.id = t2.id")
- withSQLConf(SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key ->
"true") {
- checkAnswer(df, Seq(Row(0L, "a", "x"), Row(1L, "b", "y")))
- val plan = stripAQEPlan(df.queryExecution.executedPlan)
- assert(collectShuffles(plan).isEmpty, "storage-partitioned join should
not shuffle")
- val groupPartitions = collectGroupPartitions(plan)
- assert(groupPartitions.size == 2,
- "both sides should be regrouped onto the merged partition keys")
- assert(groupPartitions.count(_.reducers.exists(_.exists(_.isDefined)))
== 1,
- "and exactly one side should reduce, the identity one onto
`flip_low_bit`")
- // The join subtree, not the whole plan: `ValidateRequirements` walks
children and a query
- // stage is a leaf, so validating an AQE plan checks nothing.
- val joins = collect(plan) { case smj: SortMergeJoinExec => smj }
- assert(joins.size == 1, s"test setup: one join to validate:\n$plan")
- assert(ValidateRequirements.validate(joins.head), "the plan that
leaves must hold up")
+ withSQLConf(SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key ->
"true") {
+ checkAnswer(df, expected)
+ val plan = stripAQEPlan(df.queryExecution.executedPlan)
+ assert(collectShuffles(plan).isEmpty,
+ s"$joinType: storage-partitioned join should not shuffle")
+ val groupPartitions = collectGroupPartitions(plan)
+ assert(groupPartitions.size == 2,
+ s"$joinType: both sides should be regrouped onto the merged
partition keys")
+
assert(groupPartitions.count(_.reducers.exists(_.exists(_.isDefined))) == 1,
+ s"$joinType: exactly one side should reduce, the identity one onto
`flip_low_bit`")
+ // The join subtree, not the whole plan: `ValidateRequirements`
walks children and a
+ // query stage is a leaf, so validating an AQE plan checks nothing.
+ val joins = collect(plan) { case smj: SortMergeJoinExec => smj }
+ assert(joins.size == 1, s"$joinType: test setup: one join to
validate:\n$plan")
+ assert(ValidateRequirements.validate(joins.head),
+ s"$joinType: the plan that leaves must hold up")
+ }
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
