[VL] Fix GlutenRemoveRedundantSortsSuite in Spark 4.x

gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -733,7 +733,9 @@ class VeloxTestSettings extends BackendTestSettings {
   // TODO: 4.x enableSuite[GlutenProjectedOrderingAndPartitioningSuite]  // 6 failures
   enableSuite[GlutenQueryPlanningTrackerEndToEndSuite]
   // TODO: 4.x enableSuite[GlutenRemoveRedundantProjectsSuite]  // 14 failures
-  // TODO: 4.x enableSuite[GlutenRemoveRedundantSortsSuite]  // 1 failure
+  enableSuite[GlutenRemoveRedundantSortsSuite]
+    // Rewrite as it check spark SortExec.
+    .includeAllGlutenTests()
   enableSuite[GlutenRowToColumnConverterSuite]
   enableSuite[GlutenSQLExecutionSuite]
   enableSuite[GlutenSQLFunctionSuite]

gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantSortsSuite.scala

@@ -16,6 +16,157 @@
  */
 package org.apache.spark.sql.execution
 
-import org.apache.spark.sql.GlutenSQLTestsTrait
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.SortExecTransformer
 
-class GlutenRemoveRedundantSortsSuite extends RemoveRedundantSortsSuite with GlutenSQLTestsTrait {}
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{DataFrame, GlutenSQLTestsTrait}
+import org.apache.spark.sql.catalyst.plans.physical.{RangePartitioning, UnknownPartitioning}
+import org.apache.spark.sql.execution.joins.ShuffledJoin
+import org.apache.spark.sql.internal.SQLConf
+
+class GlutenRemoveRedundantSortsSuite extends RemoveRedundantSortsSuite
+  with GlutenSQLTestsTrait {
+  import testImplicits._
+
+  override def sparkConf: SparkConf = {
+    super.sparkConf
+      .set(GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "false")
+  }
+
+  private def checkNumSorts(df: DataFrame, count: Int): Unit = {
+    val plan = df.queryExecution.executedPlan
+    assert(collectWithSubqueries(plan) { case s: SortExecTransformer => s }.length == count)
+  }
+
+  private def checkSorts(query: String, enabledCount: Int, disabledCount: Int): Unit = {
+    withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "true") {
+      val df = sql(query)
+      checkNumSorts(df, enabledCount)
+      val result = df.collect()
+      withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "false") {
+        val df = sql(query)
+        checkNumSorts(df, disabledCount)
+        checkAnswer(df, result)
+      }
+    }
+  }
+
+  testGluten("remove redundant sorts with limit") {
+    withTempView("t") {
+      spark.range(100).select($"id".as("key")).createOrReplaceTempView("t")
+      val query =
+        """
+          |SELECT key FROM
+          | (SELECT key FROM t WHERE key > 10 ORDER BY key DESC LIMIT 10)
+          |ORDER BY key DESC
+          |""".stripMargin
+      checkSorts(query, 0, 1)
+    }
+  }
+
+  testGluten("remove redundant sorts with broadcast hash join") {
+    withTempView("t1", "t2") {
+      spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t1")
+      spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t2")
+
+      val queryTemplate = """
+                            |SELECT /*+ BROADCAST(%s) */ t1.key FROM
+                            | (SELECT key FROM t1 WHERE key > 10 ORDER BY key DESC LIMIT 10) t1
+                            |JOIN
+                            | (SELECT key FROM t2 WHERE key > 50 ORDER BY key DESC LIMIT 100) t2
+                            |ON t1.key = t2.key
+                            |ORDER BY %s
+      """.stripMargin
+
+      // No sort should be removed since the stream side (t2) order DESC
+      // does not satisfy the required sort order ASC.
+      val buildLeftOrderByRightAsc = queryTemplate.format("t1", "t2.key ASC")
+      checkSorts(buildLeftOrderByRightAsc, 1, 1)
+
+      // The top sort node should be removed since the stream side (t2) order DESC already
+      // satisfies the required sort order DESC.
+      val buildLeftOrderByRightDesc = queryTemplate.format("t1", "t2.key DESC")
+      checkSorts(buildLeftOrderByRightDesc, 0, 1)
+
+      // No sort should be removed since the sort ordering from broadcast-hash join is based
+      // on the stream side (t2) and the required sort order is from t1.
+      val buildLeftOrderByLeftDesc = queryTemplate.format("t1", "t1.key DESC")
+      checkSorts(buildLeftOrderByLeftDesc, 1, 1)
+
+      // The top sort node should be removed since the stream side (t1) order DESC already
+      // satisfies the required sort order DESC.
+      val buildRightOrderByLeftDesc = queryTemplate.format("t2", "t1.key DESC")
+      checkSorts(buildRightOrderByLeftDesc, 0, 1)
+    }
+  }
+
+  testGluten("remove redundant sorts with sort merge join") {
+    withTempView("t1", "t2") {
+      spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t1")
+      spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t2")
+      val query = """
+                    |SELECT /*+ MERGE(t1) */ t1.key FROM
+                    | (SELECT key FROM t1 WHERE key > 10 ORDER BY key DESC LIMIT 10) t1
+                    |JOIN
+                    | (SELECT key FROM t2 WHERE key > 50 ORDER BY key DESC LIMIT 100) t2
+                    |ON t1.key = t2.key
+                    |ORDER BY t1.key
+      """.stripMargin
+
+      val queryAsc = query + " ASC"
+      checkSorts(queryAsc, 2, 3)
+
+      // The top level sort should not be removed since the child output ordering is ASC and
+      // the required ordering is DESC.
+      val queryDesc = query + " DESC"
+      checkSorts(queryDesc, 3, 3)
+    }
+  }
+
+  testGluten("cached sorted data doesn't need to be re-sorted") {
+    withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "true") {
+      val df = spark.range(1000).select($"id".as("key")).sort($"key".desc).cache()
+      df.collect()
+      val resorted = df.sort($"key".desc)
+      val sortedAsc = df.sort($"key".asc)
+      checkNumSorts(df, 0)
+      checkNumSorts(resorted, 0)
+      checkNumSorts(sortedAsc, 1)
+      val result = resorted.collect()
+      withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "false") {
+        val resorted = df.sort($"key".desc)
+        resorted.collect()
+        checkNumSorts(resorted, 1)
+        checkAnswer(resorted, result)
+      }
+    }
+  }
+
+  testGluten("SPARK-33472: shuffled join with different left and right side partition numbers") {
+    withTempView("t1", "t2") {
+      spark.range(0, 100, 1, 2).select($"id".as("key")).createOrReplaceTempView("t1")
+      (0 to 100).toDF("key").createOrReplaceTempView("t2")
+
+      val queryTemplate = """
+                            |SELECT /*+ %s(t1) */ t1.key
+                            |FROM t1 JOIN t2 ON t1.key = t2.key
+                            |WHERE t1.key > 10 AND t2.key < 50
+                            |ORDER BY t1.key ASC
+      """.stripMargin
+
+      Seq(("MERGE", 3), ("SHUFFLE_HASH", 1)).foreach {
+        case (hint, count) =>
+          val query = queryTemplate.format(hint)
+          val df = sql(query)
+          val sparkPlan = df.queryExecution.sparkPlan
+          val join = sparkPlan.collect { case j: ShuffledJoin => j }.head
+          val leftPartitioning = join.left.outputPartitioning
+          assert(leftPartitioning.isInstanceOf[RangePartitioning])
+          assert(leftPartitioning.numPartitions == 2)
+          assert(join.right.outputPartitioning == UnknownPartitioning(0))
+          checkSorts(query, count, count)
+      }
+    }
+  }
+}

gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -715,7 +715,9 @@ class VeloxTestSettings extends BackendTestSettings {
   // TODO: 4.x enableSuite[GlutenProjectedOrderingAndPartitioningSuite]  // 6 failures
   enableSuite[GlutenQueryPlanningTrackerEndToEndSuite]
   // TODO: 4.x enableSuite[GlutenRemoveRedundantProjectsSuite]  // 14 failures
-  // TODO: 4.x enableSuite[GlutenRemoveRedundantSortsSuite]  // 1 failure
+  enableSuite[GlutenRemoveRedundantSortsSuite]
+    // Rewrite as it check spark SortExec.
+    .includeAllGlutenTests()
   enableSuite[GlutenRowToColumnConverterSuite]
   enableSuite[GlutenSQLExecutionSuite]
   enableSuite[GlutenSQLFunctionSuite]

gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantSortsSuite.scala

@@ -16,6 +16,157 @@
  */
 package org.apache.spark.sql.execution
 
-import org.apache.spark.sql.GlutenSQLTestsTrait
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.execution.SortExecTransformer
 
-class GlutenRemoveRedundantSortsSuite extends RemoveRedundantSortsSuite with GlutenSQLTestsTrait {}
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{DataFrame, GlutenSQLTestsTrait}
+import org.apache.spark.sql.catalyst.plans.physical.{RangePartitioning, UnknownPartitioning}
+import org.apache.spark.sql.execution.joins.ShuffledJoin
+import org.apache.spark.sql.internal.SQLConf
+
+class GlutenRemoveRedundantSortsSuite extends RemoveRedundantSortsSuite
+  with GlutenSQLTestsTrait {
+  import testImplicits._
+
+  override def sparkConf: SparkConf = {
+    super.sparkConf
+      .set(GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "false")
+  }
+
+  private def checkNumSorts(df: DataFrame, count: Int): Unit = {
+    val plan = df.queryExecution.executedPlan
+    assert(collectWithSubqueries(plan) { case s: SortExecTransformer => s }.length == count)
+  }
+
+  private def checkSorts(query: String, enabledCount: Int, disabledCount: Int): Unit = {
+    withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "true") {
+      val df = sql(query)
+      checkNumSorts(df, enabledCount)
+      val result = df.collect()
+      withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "false") {
+        val df = sql(query)
+        checkNumSorts(df, disabledCount)
+        checkAnswer(df, result)
+      }
+    }
+  }
+
+  testGluten("remove redundant sorts with limit") {
+    withTempView("t") {
+      spark.range(100).select($"id".as("key")).createOrReplaceTempView("t")
+      val query =
+        """
+          |SELECT key FROM
+          | (SELECT key FROM t WHERE key > 10 ORDER BY key DESC LIMIT 10)
+          |ORDER BY key DESC
+          |""".stripMargin
+      checkSorts(query, 0, 1)
+    }
+  }
+
+  testGluten("remove redundant sorts with broadcast hash join") {
+    withTempView("t1", "t2") {
+      spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t1")
+      spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t2")
+
+      val queryTemplate = """
+                            |SELECT /*+ BROADCAST(%s) */ t1.key FROM
+                            | (SELECT key FROM t1 WHERE key > 10 ORDER BY key DESC LIMIT 10) t1
+                            |JOIN
+                            | (SELECT key FROM t2 WHERE key > 50 ORDER BY key DESC LIMIT 100) t2
+                            |ON t1.key = t2.key
+                            |ORDER BY %s
+      """.stripMargin
+
+      // No sort should be removed since the stream side (t2) order DESC
+      // does not satisfy the required sort order ASC.
+      val buildLeftOrderByRightAsc = queryTemplate.format("t1", "t2.key ASC")
+      checkSorts(buildLeftOrderByRightAsc, 1, 1)
+
+      // The top sort node should be removed since the stream side (t2) order DESC already
+      // satisfies the required sort order DESC.
+      val buildLeftOrderByRightDesc = queryTemplate.format("t1", "t2.key DESC")
+      checkSorts(buildLeftOrderByRightDesc, 0, 1)
+
+      // No sort should be removed since the sort ordering from broadcast-hash join is based
+      // on the stream side (t2) and the required sort order is from t1.
+      val buildLeftOrderByLeftDesc = queryTemplate.format("t1", "t1.key DESC")
+      checkSorts(buildLeftOrderByLeftDesc, 1, 1)
+
+      // The top sort node should be removed since the stream side (t1) order DESC already
+      // satisfies the required sort order DESC.
+      val buildRightOrderByLeftDesc = queryTemplate.format("t2", "t1.key DESC")
+      checkSorts(buildRightOrderByLeftDesc, 0, 1)
+    }
+  }
+
+  testGluten("remove redundant sorts with sort merge join") {
+    withTempView("t1", "t2") {
+      spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t1")
+      spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t2")
+      val query = """
+                    |SELECT /*+ MERGE(t1) */ t1.key FROM
+                    | (SELECT key FROM t1 WHERE key > 10 ORDER BY key DESC LIMIT 10) t1
+                    |JOIN
+                    | (SELECT key FROM t2 WHERE key > 50 ORDER BY key DESC LIMIT 100) t2
+                    |ON t1.key = t2.key
+                    |ORDER BY t1.key
+      """.stripMargin
+
+      val queryAsc = query + " ASC"
+      checkSorts(queryAsc, 2, 3)
+
+      // The top level sort should not be removed since the child output ordering is ASC and
+      // the required ordering is DESC.
+      val queryDesc = query + " DESC"
+      checkSorts(queryDesc, 3, 3)
+    }
+  }
+
+  testGluten("cached sorted data doesn't need to be re-sorted") {
+    withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "true") {
+      val df = spark.range(1000).select($"id".as("key")).sort($"key".desc).cache()
+      df.collect()
+      val resorted = df.sort($"key".desc)
+      val sortedAsc = df.sort($"key".asc)
+      checkNumSorts(df, 0)
+      checkNumSorts(resorted, 0)
+      checkNumSorts(sortedAsc, 1)
+      val result = resorted.collect()
+      withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "false") {
+        val resorted = df.sort($"key".desc)
+        resorted.collect()
+        checkNumSorts(resorted, 1)
+        checkAnswer(resorted, result)
+      }
+    }
+  }
+
+  testGluten("SPARK-33472: shuffled join with different left and right side partition numbers") {
+    withTempView("t1", "t2") {
+      spark.range(0, 100, 1, 2).select($"id".as("key")).createOrReplaceTempView("t1")
+      (0 to 100).toDF("key").createOrReplaceTempView("t2")
+
+      val queryTemplate = """
+                            |SELECT /*+ %s(t1) */ t1.key
+                            |FROM t1 JOIN t2 ON t1.key = t2.key
+                            |WHERE t1.key > 10 AND t2.key < 50
+                            |ORDER BY t1.key ASC
+      """.stripMargin
+
+      Seq(("MERGE", 3), ("SHUFFLE_HASH", 1)).foreach {
+        case (hint, count) =>
+          val query = queryTemplate.format(hint)
+          val df = sql(query)
+          val sparkPlan = df.queryExecution.sparkPlan
+          val join = sparkPlan.collect { case j: ShuffledJoin => j }.head
+          val leftPartitioning = join.left.outputPartitioning
+          assert(leftPartitioning.isInstanceOf[RangePartitioning])
+          assert(leftPartitioning.numPartitions == 2)
+          assert(join.right.outputPartitioning == UnknownPartitioning(0))
+          checkSorts(query, count, count)
+      }
+    }
+  }
+}