Preserve hash join (BHJ and SHJ) stream side ordering

Author: c21

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashJoin.scala

@@ -54,6 +54,8 @@ trait HashJoin extends BaseJoinExec {
 
   override def outputPartitioning: Partitioning = streamedPlan.outputPartitioning
 
+  override def outputOrdering: Seq[SortOrder] = streamedPlan.outputOrdering
+
   protected lazy val (buildPlan, streamedPlan) = buildSide match {
     case BuildLeft => (left, right)
     case BuildRight => (right, left)

sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala

@@ -1104,4 +1104,42 @@ class JoinSuite extends QueryTest with SharedSparkSession with AdaptiveSparkPlan
       })
     }
   }
+
+  test("SPARK-32383: Preserve hash join (BHJ and SHJ) stream side ordering") {
+    val df1 = spark.range(100).select($"id".as("k1"))
+    val df2 = spark.range(100).select($"id".as("k2"))
+    val df3 = spark.range(3).select($"id".as("k3"))
+    val df4 = spark.range(100).select($"id".as("k4"))
+
+    // Test broadcast hash join
+    withSQLConf(
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "50") {
+      val plan = df1.join(df2, $"k1" === $"k2")
+        .join(df3, $"k1" === $"k3")
+        .join(df4, $"k1" === $"k4")
+        .queryExecution
+        .executedPlan
+      assert(plan.collect { case _: SortMergeJoinExec => true }.size === 2)
+      assert(plan.collect { case _: BroadcastHashJoinExec => true }.size === 1)
+      // No extra sort before last sort merge join
+      assert(plan.collect { case _: SortExec => true }.size === 3)
+    }
+
+    // Test shuffled hash join
+    withSQLConf(
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "50",
+      SQLConf.SHUFFLE_PARTITIONS.key -> "2",
+      SQLConf.PREFER_SORTMERGEJOIN.key -> "false") {
+      val df3 = spark.range(10).select($"id".as("k3"))
+      val plan = df1.join(df2, $"k1" === $"k2")
+        .join(df3, $"k1" === $"k3")
+        .join(df4, $"k1" === $"k4")
+        .queryExecution
+        .executedPlan
+      assert(plan.collect { case _: SortMergeJoinExec => true }.size === 2)
+      assert(plan.collect { case _: ShuffledHashJoinExec => true }.size === 1)
+      // No extra sort before last sort merge join
+      assert(plan.collect { case _: SortExec => true }.size === 3)
+    }
+  }
 }