Skip to content

Commit 061b799

Browse files
zml1206zml1206
authored
[VL] Fix GlutenRemoveRedundantSortsSuite in Spark 4.x (#12497)
1 parent e6cc336 commit 061b799

4 files changed

Lines changed: 312 additions & 6 deletions

File tree

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -738,7 +738,9 @@ class VeloxTestSettings extends BackendTestSettings {
738738
// TODO: 4.x enableSuite[GlutenProjectedOrderingAndPartitioningSuite] // 6 failures
739739
enableSuite[GlutenQueryPlanningTrackerEndToEndSuite]
740740
// TODO: 4.x enableSuite[GlutenRemoveRedundantProjectsSuite] // 14 failures
741-
// TODO: 4.x enableSuite[GlutenRemoveRedundantSortsSuite] // 1 failure
741+
enableSuite[GlutenRemoveRedundantSortsSuite]
742+
// Rewrite as it check spark SortExec.
743+
.includeAllGlutenTests()
742744
enableSuite[GlutenRowToColumnConverterSuite]
743745
enableSuite[GlutenSQLExecutionSuite]
744746
enableSuite[GlutenSQLFunctionSuite]

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

Lines changed: 153 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,157 @@
1616
*/
1717
package org.apache.spark.sql.execution
1818

19-
import org.apache.spark.sql.GlutenSQLTestsTrait
19+
import org.apache.gluten.config.GlutenConfig
20+
import org.apache.gluten.execution.SortExecTransformer
2021

21-
class GlutenRemoveRedundantSortsSuite extends RemoveRedundantSortsSuite with GlutenSQLTestsTrait {}
22+
import org.apache.spark.SparkConf
23+
import org.apache.spark.sql.{DataFrame, GlutenSQLTestsTrait}
24+
import org.apache.spark.sql.catalyst.plans.physical.{RangePartitioning, UnknownPartitioning}
25+
import org.apache.spark.sql.execution.joins.ShuffledJoin
26+
import org.apache.spark.sql.internal.SQLConf
27+
28+
class GlutenRemoveRedundantSortsSuite extends RemoveRedundantSortsSuite
29+
with GlutenSQLTestsTrait {
30+
import testImplicits._
31+
32+
override def sparkConf: SparkConf = {
33+
super.sparkConf
34+
.set(GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "false")
35+
}
36+
37+
private def checkNumSorts(df: DataFrame, count: Int): Unit = {
38+
val plan = df.queryExecution.executedPlan
39+
assert(collectWithSubqueries(plan) { case s: SortExecTransformer => s }.length == count)
40+
}
41+
42+
private def checkSorts(query: String, enabledCount: Int, disabledCount: Int): Unit = {
43+
withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "true") {
44+
val df = sql(query)
45+
checkNumSorts(df, enabledCount)
46+
val result = df.collect()
47+
withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "false") {
48+
val df = sql(query)
49+
checkNumSorts(df, disabledCount)
50+
checkAnswer(df, result)
51+
}
52+
}
53+
}
54+
55+
testGluten("remove redundant sorts with limit") {
56+
withTempView("t") {
57+
spark.range(100).select($"id".as("key")).createOrReplaceTempView("t")
58+
val query =
59+
"""
60+
|SELECT key FROM
61+
| (SELECT key FROM t WHERE key > 10 ORDER BY key DESC LIMIT 10)
62+
|ORDER BY key DESC
63+
|""".stripMargin
64+
checkSorts(query, 0, 1)
65+
}
66+
}
67+
68+
testGluten("remove redundant sorts with broadcast hash join") {
69+
withTempView("t1", "t2") {
70+
spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t1")
71+
spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t2")
72+
73+
val queryTemplate = """
74+
|SELECT /*+ BROADCAST(%s) */ t1.key FROM
75+
| (SELECT key FROM t1 WHERE key > 10 ORDER BY key DESC LIMIT 10) t1
76+
|JOIN
77+
| (SELECT key FROM t2 WHERE key > 50 ORDER BY key DESC LIMIT 100) t2
78+
|ON t1.key = t2.key
79+
|ORDER BY %s
80+
""".stripMargin
81+
82+
// No sort should be removed since the stream side (t2) order DESC
83+
// does not satisfy the required sort order ASC.
84+
val buildLeftOrderByRightAsc = queryTemplate.format("t1", "t2.key ASC")
85+
checkSorts(buildLeftOrderByRightAsc, 1, 1)
86+
87+
// The top sort node should be removed since the stream side (t2) order DESC already
88+
// satisfies the required sort order DESC.
89+
val buildLeftOrderByRightDesc = queryTemplate.format("t1", "t2.key DESC")
90+
checkSorts(buildLeftOrderByRightDesc, 0, 1)
91+
92+
// No sort should be removed since the sort ordering from broadcast-hash join is based
93+
// on the stream side (t2) and the required sort order is from t1.
94+
val buildLeftOrderByLeftDesc = queryTemplate.format("t1", "t1.key DESC")
95+
checkSorts(buildLeftOrderByLeftDesc, 1, 1)
96+
97+
// The top sort node should be removed since the stream side (t1) order DESC already
98+
// satisfies the required sort order DESC.
99+
val buildRightOrderByLeftDesc = queryTemplate.format("t2", "t1.key DESC")
100+
checkSorts(buildRightOrderByLeftDesc, 0, 1)
101+
}
102+
}
103+
104+
testGluten("remove redundant sorts with sort merge join") {
105+
withTempView("t1", "t2") {
106+
spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t1")
107+
spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t2")
108+
val query = """
109+
|SELECT /*+ MERGE(t1) */ t1.key FROM
110+
| (SELECT key FROM t1 WHERE key > 10 ORDER BY key DESC LIMIT 10) t1
111+
|JOIN
112+
| (SELECT key FROM t2 WHERE key > 50 ORDER BY key DESC LIMIT 100) t2
113+
|ON t1.key = t2.key
114+
|ORDER BY t1.key
115+
""".stripMargin
116+
117+
val queryAsc = query + " ASC"
118+
checkSorts(queryAsc, 2, 3)
119+
120+
// The top level sort should not be removed since the child output ordering is ASC and
121+
// the required ordering is DESC.
122+
val queryDesc = query + " DESC"
123+
checkSorts(queryDesc, 3, 3)
124+
}
125+
}
126+
127+
testGluten("cached sorted data doesn't need to be re-sorted") {
128+
withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "true") {
129+
val df = spark.range(1000).select($"id".as("key")).sort($"key".desc).cache()
130+
df.collect()
131+
val resorted = df.sort($"key".desc)
132+
val sortedAsc = df.sort($"key".asc)
133+
checkNumSorts(df, 0)
134+
checkNumSorts(resorted, 0)
135+
checkNumSorts(sortedAsc, 1)
136+
val result = resorted.collect()
137+
withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "false") {
138+
val resorted = df.sort($"key".desc)
139+
resorted.collect()
140+
checkNumSorts(resorted, 1)
141+
checkAnswer(resorted, result)
142+
}
143+
}
144+
}
145+
146+
testGluten("SPARK-33472: shuffled join with different left and right side partition numbers") {
147+
withTempView("t1", "t2") {
148+
spark.range(0, 100, 1, 2).select($"id".as("key")).createOrReplaceTempView("t1")
149+
(0 to 100).toDF("key").createOrReplaceTempView("t2")
150+
151+
val queryTemplate = """
152+
|SELECT /*+ %s(t1) */ t1.key
153+
|FROM t1 JOIN t2 ON t1.key = t2.key
154+
|WHERE t1.key > 10 AND t2.key < 50
155+
|ORDER BY t1.key ASC
156+
""".stripMargin
157+
158+
Seq(("MERGE", 3), ("SHUFFLE_HASH", 1)).foreach {
159+
case (hint, count) =>
160+
val query = queryTemplate.format(hint)
161+
val df = sql(query)
162+
val sparkPlan = df.queryExecution.sparkPlan
163+
val join = sparkPlan.collect { case j: ShuffledJoin => j }.head
164+
val leftPartitioning = join.left.outputPartitioning
165+
assert(leftPartitioning.isInstanceOf[RangePartitioning])
166+
assert(leftPartitioning.numPartitions == 2)
167+
assert(join.right.outputPartitioning == UnknownPartitioning(0))
168+
checkSorts(query, count, count)
169+
}
170+
}
171+
}
172+
}

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -717,7 +717,9 @@ class VeloxTestSettings extends BackendTestSettings {
717717
// TODO: 4.x enableSuite[GlutenProjectedOrderingAndPartitioningSuite] // 6 failures
718718
enableSuite[GlutenQueryPlanningTrackerEndToEndSuite]
719719
// TODO: 4.x enableSuite[GlutenRemoveRedundantProjectsSuite] // 14 failures
720-
// TODO: 4.x enableSuite[GlutenRemoveRedundantSortsSuite] // 1 failure
720+
enableSuite[GlutenRemoveRedundantSortsSuite]
721+
// Rewrite as it check spark SortExec.
722+
.includeAllGlutenTests()
721723
enableSuite[GlutenRowToColumnConverterSuite]
722724
enableSuite[GlutenSQLExecutionSuite]
723725
enableSuite[GlutenSQLFunctionSuite]

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

Lines changed: 153 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,157 @@
1616
*/
1717
package org.apache.spark.sql.execution
1818

19-
import org.apache.spark.sql.GlutenSQLTestsTrait
19+
import org.apache.gluten.config.GlutenConfig
20+
import org.apache.gluten.execution.SortExecTransformer
2021

21-
class GlutenRemoveRedundantSortsSuite extends RemoveRedundantSortsSuite with GlutenSQLTestsTrait {}
22+
import org.apache.spark.SparkConf
23+
import org.apache.spark.sql.{DataFrame, GlutenSQLTestsTrait}
24+
import org.apache.spark.sql.catalyst.plans.physical.{RangePartitioning, UnknownPartitioning}
25+
import org.apache.spark.sql.execution.joins.ShuffledJoin
26+
import org.apache.spark.sql.internal.SQLConf
27+
28+
class GlutenRemoveRedundantSortsSuite extends RemoveRedundantSortsSuite
29+
with GlutenSQLTestsTrait {
30+
import testImplicits._
31+
32+
override def sparkConf: SparkConf = {
33+
super.sparkConf
34+
.set(GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "false")
35+
}
36+
37+
private def checkNumSorts(df: DataFrame, count: Int): Unit = {
38+
val plan = df.queryExecution.executedPlan
39+
assert(collectWithSubqueries(plan) { case s: SortExecTransformer => s }.length == count)
40+
}
41+
42+
private def checkSorts(query: String, enabledCount: Int, disabledCount: Int): Unit = {
43+
withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "true") {
44+
val df = sql(query)
45+
checkNumSorts(df, enabledCount)
46+
val result = df.collect()
47+
withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "false") {
48+
val df = sql(query)
49+
checkNumSorts(df, disabledCount)
50+
checkAnswer(df, result)
51+
}
52+
}
53+
}
54+
55+
testGluten("remove redundant sorts with limit") {
56+
withTempView("t") {
57+
spark.range(100).select($"id".as("key")).createOrReplaceTempView("t")
58+
val query =
59+
"""
60+
|SELECT key FROM
61+
| (SELECT key FROM t WHERE key > 10 ORDER BY key DESC LIMIT 10)
62+
|ORDER BY key DESC
63+
|""".stripMargin
64+
checkSorts(query, 0, 1)
65+
}
66+
}
67+
68+
testGluten("remove redundant sorts with broadcast hash join") {
69+
withTempView("t1", "t2") {
70+
spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t1")
71+
spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t2")
72+
73+
val queryTemplate = """
74+
|SELECT /*+ BROADCAST(%s) */ t1.key FROM
75+
| (SELECT key FROM t1 WHERE key > 10 ORDER BY key DESC LIMIT 10) t1
76+
|JOIN
77+
| (SELECT key FROM t2 WHERE key > 50 ORDER BY key DESC LIMIT 100) t2
78+
|ON t1.key = t2.key
79+
|ORDER BY %s
80+
""".stripMargin
81+
82+
// No sort should be removed since the stream side (t2) order DESC
83+
// does not satisfy the required sort order ASC.
84+
val buildLeftOrderByRightAsc = queryTemplate.format("t1", "t2.key ASC")
85+
checkSorts(buildLeftOrderByRightAsc, 1, 1)
86+
87+
// The top sort node should be removed since the stream side (t2) order DESC already
88+
// satisfies the required sort order DESC.
89+
val buildLeftOrderByRightDesc = queryTemplate.format("t1", "t2.key DESC")
90+
checkSorts(buildLeftOrderByRightDesc, 0, 1)
91+
92+
// No sort should be removed since the sort ordering from broadcast-hash join is based
93+
// on the stream side (t2) and the required sort order is from t1.
94+
val buildLeftOrderByLeftDesc = queryTemplate.format("t1", "t1.key DESC")
95+
checkSorts(buildLeftOrderByLeftDesc, 1, 1)
96+
97+
// The top sort node should be removed since the stream side (t1) order DESC already
98+
// satisfies the required sort order DESC.
99+
val buildRightOrderByLeftDesc = queryTemplate.format("t2", "t1.key DESC")
100+
checkSorts(buildRightOrderByLeftDesc, 0, 1)
101+
}
102+
}
103+
104+
testGluten("remove redundant sorts with sort merge join") {
105+
withTempView("t1", "t2") {
106+
spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t1")
107+
spark.range(1000).select($"id".as("key")).createOrReplaceTempView("t2")
108+
val query = """
109+
|SELECT /*+ MERGE(t1) */ t1.key FROM
110+
| (SELECT key FROM t1 WHERE key > 10 ORDER BY key DESC LIMIT 10) t1
111+
|JOIN
112+
| (SELECT key FROM t2 WHERE key > 50 ORDER BY key DESC LIMIT 100) t2
113+
|ON t1.key = t2.key
114+
|ORDER BY t1.key
115+
""".stripMargin
116+
117+
val queryAsc = query + " ASC"
118+
checkSorts(queryAsc, 2, 3)
119+
120+
// The top level sort should not be removed since the child output ordering is ASC and
121+
// the required ordering is DESC.
122+
val queryDesc = query + " DESC"
123+
checkSorts(queryDesc, 3, 3)
124+
}
125+
}
126+
127+
testGluten("cached sorted data doesn't need to be re-sorted") {
128+
withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "true") {
129+
val df = spark.range(1000).select($"id".as("key")).sort($"key".desc).cache()
130+
df.collect()
131+
val resorted = df.sort($"key".desc)
132+
val sortedAsc = df.sort($"key".asc)
133+
checkNumSorts(df, 0)
134+
checkNumSorts(resorted, 0)
135+
checkNumSorts(sortedAsc, 1)
136+
val result = resorted.collect()
137+
withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "false") {
138+
val resorted = df.sort($"key".desc)
139+
resorted.collect()
140+
checkNumSorts(resorted, 1)
141+
checkAnswer(resorted, result)
142+
}
143+
}
144+
}
145+
146+
testGluten("SPARK-33472: shuffled join with different left and right side partition numbers") {
147+
withTempView("t1", "t2") {
148+
spark.range(0, 100, 1, 2).select($"id".as("key")).createOrReplaceTempView("t1")
149+
(0 to 100).toDF("key").createOrReplaceTempView("t2")
150+
151+
val queryTemplate = """
152+
|SELECT /*+ %s(t1) */ t1.key
153+
|FROM t1 JOIN t2 ON t1.key = t2.key
154+
|WHERE t1.key > 10 AND t2.key < 50
155+
|ORDER BY t1.key ASC
156+
""".stripMargin
157+
158+
Seq(("MERGE", 3), ("SHUFFLE_HASH", 1)).foreach {
159+
case (hint, count) =>
160+
val query = queryTemplate.format(hint)
161+
val df = sql(query)
162+
val sparkPlan = df.queryExecution.sparkPlan
163+
val join = sparkPlan.collect { case j: ShuffledJoin => j }.head
164+
val leftPartitioning = join.left.outputPartitioning
165+
assert(leftPartitioning.isInstanceOf[RangePartitioning])
166+
assert(leftPartitioning.numPartitions == 2)
167+
assert(join.right.outputPartitioning == UnknownPartitioning(0))
168+
checkSorts(query, count, count)
169+
}
170+
}
171+
}
172+
}

0 commit comments

Comments
 (0)