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 bf2f77b6237..1a8c9f6ef86 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 @@ -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] 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 02df13e30c0..5c4b439d41e 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 7048695d3ef..78e152f8ff9 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 @@ -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] 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 02df13e30c0..5c4b439d41e 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) + } + } + } +}