[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 <naveenp2708@gmail.com>
Signed-off-by: Szehon Ho <szehon.apache@gmail.com>

Author: naveenp2708

sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala

@@ -1410,30 +1410,39 @@ 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')")
 
       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")
+        }
       }
     }
   }