This is an automated email from the ASF dual-hosted git repository.
zml1206 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 061b799fbb [VL] Fix GlutenRemoveRedundantSortsSuite in Spark 4.x
(#12497)
061b799fbb is described below
commit 061b799fbb7dd7fbe7952daa375c1958d84d2df0
Author: Mingliang Zhu <[email protected]>
AuthorDate: Tue Jul 14 13:30:45 2026 +0800
[VL] Fix GlutenRemoveRedundantSortsSuite in Spark 4.x (#12497)
---
.../gluten/utils/velox/VeloxTestSettings.scala | 4 +-
.../GlutenRemoveRedundantSortsSuite.scala | 155 ++++++++++++++++++++-
.../gluten/utils/velox/VeloxTestSettings.scala | 4 +-
.../GlutenRemoveRedundantSortsSuite.scala | 155 ++++++++++++++++++++-
4 files changed, 312 insertions(+), 6 deletions(-)
diff --git
a/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index 48c70d1713..a7367ecd7a 100644
---
a/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -738,7 +738,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]
diff --git
a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantSortsSuite.scala
b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantSortsSuite.scala
index 02df13e30c..5c4b439d41 100644
---
a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantSortsSuite.scala
+++
b/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)
+ }
+ }
+ }
+}
diff --git
a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
index a6a5c67396..18fc0e2c22 100644
---
a/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
+++
b/gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala
@@ -717,7 +717,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]
diff --git
a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantSortsSuite.scala
b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantSortsSuite.scala
index 02df13e30c..5c4b439d41 100644
---
a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantSortsSuite.scala
+++
b/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)
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
