Skip to content

Commit 8a60e59

Browse files
zml1206zml1206
authored
[GLUTEN-11550][VL] Fix GlutenRemoveRedundantProjectsSuite in Spark 4.x (#12506)
1 parent 522b45d commit 8a60e59

4 files changed

Lines changed: 404 additions & 6 deletions

File tree

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

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -741,7 +741,22 @@ class VeloxTestSettings extends BackendTestSettings {
741741
disableSuite[GlutenProjectedOrderingAndPartitioningSuite](
742742
"Validates Spark planner output ordering and partitioning metadata")
743743
enableSuite[GlutenQueryPlanningTrackerEndToEndSuite]
744-
// TODO: 4.x enableSuite[GlutenRemoveRedundantProjectsSuite] // 14 failures
744+
enableSuite[GlutenRemoveRedundantProjectsSuite]
745+
// Rewrite as result checks because Gluten transforms and may pull out additional projects.
746+
.exclude("project with filter")
747+
.exclude("project with specific column ordering")
748+
.exclude("project with extra columns")
749+
.exclude("project with fewer columns")
750+
.exclude("aggregate without ordering requirement")
751+
.exclude("aggregate with ordering requirement")
752+
.exclude("join without ordering requirement")
753+
.exclude("join with ordering requirement")
754+
.exclude("window function")
755+
.exclude("generate should require column ordering")
756+
.exclude("subquery")
757+
.exclude("SPARK-33697: UnionExec should require column ordering")
758+
.exclude("SPARK-33697: remove redundant projects under expand")
759+
.exclude("SPARK-36020: Project should not be removed when child's logical link is different")
745760
enableSuite[GlutenRemoveRedundantSortsSuite]
746761
// Rewrite as it check spark SortExec.
747762
.includeAllGlutenTests()

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

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

19-
import org.apache.spark.sql.GlutenSQLTestsTrait
19+
import org.apache.spark.sql.{GlutenSQLTestsTrait, Row}
20+
import org.apache.spark.sql.internal.SQLConf
21+
import org.apache.spark.sql.types.StructType
2022

2123
class GlutenRemoveRedundantProjectsSuite
2224
extends RemoveRedundantProjectsSuite
23-
with GlutenSQLTestsTrait {}
25+
with GlutenSQLTestsTrait {
26+
27+
// The original tests count Spark ProjectExec nodes, while Gluten converts offloaded projects to
28+
// ProjectExecTransformer. PullOutPreProject and PullOutPostProject may also insert additional
29+
// projects while rewriting the Spark physical plan, so project counts are not directly
30+
// comparable. Therefore, these tests only verify that query results are identical with redundant
31+
// project removal enabled and disabled.
32+
private def assertProjectExec(query: String, enabled: Int, disabled: Int): Unit = {
33+
val df = sql(query)
34+
// When enabling AQE, the DPP subquery filters is replaced in runtime.
35+
df.collect()
36+
// assertProjectExecCount(df, enabled)
37+
val result = df.collect()
38+
withSQLConf(SQLConf.REMOVE_REDUNDANT_PROJECTS_ENABLED.key -> "false") {
39+
val df2 = sql(query)
40+
df2.collect()
41+
// assertProjectExecCount(df2, disabled)
42+
checkAnswer(df2, result)
43+
}
44+
}
45+
46+
testGluten("project with filter") {
47+
val query = "select * from testView where a > 5"
48+
assertProjectExec(query, 0, 1)
49+
}
50+
51+
testGluten("project with specific column ordering") {
52+
val query = "select key, a, b, c from testView"
53+
assertProjectExec(query, 1, 1)
54+
}
55+
56+
testGluten("project with extra columns") {
57+
val query = "select a, b, c, key, a from testView"
58+
assertProjectExec(query, 1, 1)
59+
}
60+
61+
testGluten("project with fewer columns") {
62+
val query = "select a from testView where a > 3"
63+
assertProjectExec(query, 1, 1)
64+
}
65+
66+
testGluten("aggregate without ordering requirement") {
67+
val query = "select sum(a) as sum_a, key, last(b) as last_b " +
68+
"from (select key, a, b from testView where a > 100) group by key"
69+
assertProjectExec(query, 0, 1)
70+
}
71+
72+
testGluten("aggregate with ordering requirement") {
73+
val query = "select a, sum(b) as sum_b from testView group by a"
74+
assertProjectExec(query, 1, 1)
75+
}
76+
77+
testGluten("join without ordering requirement") {
78+
val query = "select t1.key, t2.key, t1.a, t2.b from (select key, a, b, c from testView)" +
79+
" as t1 join (select key, a, b, c from testView) as t2 on t1.c > t2.c and t1.key > 10"
80+
assertProjectExec(query, 1, 3)
81+
}
82+
83+
testGluten("join with ordering requirement") {
84+
val query = "select * from (select key, a, c, b from testView) as t1 join " +
85+
"(select key, a, b, c from testView) as t2 on t1.key = t2.key where t2.a > 50"
86+
assertProjectExec(query, 2, 2)
87+
}
88+
89+
testGluten("window function") {
90+
val query = "select key, b, avg(a) over (partition by key order by a " +
91+
"rows between 1 preceding and 1 following) as avg from testView"
92+
assertProjectExec(query, 1, 2)
93+
}
94+
95+
testGluten("generate should require column ordering") {
96+
withTempView("testData") {
97+
spark.range(0, 10, 1)
98+
.selectExpr("id as key", "id * 2 as a", "id * 3 as b")
99+
.createOrReplaceTempView("testData")
100+
101+
val data = sql("select key, a, b, count(*) from testData group by key, a, b limit 2")
102+
val df = data.selectExpr("a", "b", "key", "explode(array(key, a, b)) as d").filter("d > 0")
103+
df.collect()
104+
val plan = df.queryExecution.executedPlan
105+
val numProjects = collectWithSubqueries(plan) { case p: ProjectExec => p }.length
106+
107+
// Create a new plan that reverse the GenerateExec output and add a new ProjectExec between
108+
// GenerateExec and its child. This is to test if the ProjectExec is removed, the output of
109+
// the query will be incorrect.
110+
val newPlan = stripAQEPlan(plan).transform {
111+
case g @ GenerateExec(_, requiredChildOutput, _, _, child) =>
112+
g.copy(
113+
requiredChildOutput = requiredChildOutput.reverse,
114+
child = ProjectExec(requiredChildOutput.reverse, child))
115+
}
116+
117+
// Re-apply remove redundant project rule.
118+
val rule = RemoveRedundantProjects
119+
val newExecutedPlan = rule.apply(newPlan)
120+
// The manually added ProjectExec node shouldn't be removed.
121+
// assert(collectWithSubqueries(newExecutedPlan) {
122+
// case p: ProjectExec => p
123+
// }.size == numProjects + 1)
124+
125+
// Check the original plan's output and the new plan's output are the same.
126+
val expectedRows = plan.executeCollect()
127+
val actualRows = newExecutedPlan.executeCollect()
128+
assert(expectedRows.length == actualRows.length)
129+
expectedRows.zip(actualRows).foreach { case (expected, actual) => assert(expected == actual) }
130+
}
131+
}
132+
133+
testGluten("subquery") {
134+
withTempView("testData") {
135+
val data = spark.sparkContext.parallelize((1 to 100).map(i => Row(i, i.toString)))
136+
val schema = new StructType().add("key", "int").add("value", "string")
137+
spark.createDataFrame(data, schema).createOrReplaceTempView("testData")
138+
val query = "select key, value from testData where key in " +
139+
"(select sum(a) from testView where a > 5 group by key)"
140+
assertProjectExec(query, 0, 1)
141+
}
142+
}
143+
144+
testGluten("SPARK-33697: UnionExec should require column ordering") {
145+
withTable("t1", "t2") {
146+
spark.range(-10, 20)
147+
.selectExpr(
148+
"id",
149+
"date_add(date '1950-01-01', cast(id as int)) as datecol",
150+
"cast(id as string) strcol")
151+
.write.mode("overwrite").format("parquet").saveAsTable("t1")
152+
spark.range(-10, 20)
153+
.selectExpr(
154+
"cast(id as string) strcol",
155+
"id",
156+
"date_add(date '1950-01-01', cast(id as int)) as datecol")
157+
.write.mode("overwrite").format("parquet").saveAsTable("t2")
158+
159+
val queryTemplate =
160+
"""
161+
|SELECT DISTINCT datecol, strcol FROM
162+
|(
163+
|(SELECT datecol, id, strcol from t1)
164+
| %s
165+
|(SELECT datecol, id, strcol from t2)
166+
|)
167+
|""".stripMargin
168+
169+
Seq(("UNION", 1, 2), ("UNION ALL", 1, 2)).foreach {
170+
case (setOperation, enabled, disabled) =>
171+
val query = queryTemplate.format(setOperation)
172+
assertProjectExec(query, enabled = enabled, disabled = disabled)
173+
}
174+
}
175+
}
176+
177+
testGluten("SPARK-33697: remove redundant projects under expand") {
178+
val query =
179+
"""
180+
|SELECT t1.key, t2.key, sum(t1.a) AS s1, sum(t2.b) AS s2 FROM
181+
|(SELECT a, key FROM testView) t1
182+
|JOIN
183+
|(SELECT b, key FROM testView) t2
184+
|ON t1.key = t2.key
185+
|GROUP BY t1.key, t2.key GROUPING SETS(t1.key, t2.key)
186+
|ORDER BY t1.key, t2.key, s1, s2
187+
|LIMIT 10
188+
|""".stripMargin
189+
// The Project above the Expand is not removed due to SPARK-36020.
190+
assertProjectExec(query, 1, 3)
191+
}
192+
193+
testGluten("SPARK-36020: Project should not be removed when child's logical link is different") {
194+
val query =
195+
"""
196+
|WITH t AS (
197+
| SELECT key, a, b, c, explode(d) AS d FROM testView
198+
|)
199+
|SELECT t1.key, t1.d, t2.key
200+
|FROM (SELECT d, key FROM t) t1
201+
|JOIN testView t2 ON t1.key = t2.key
202+
|""".stripMargin
203+
// The ProjectExec above the GenerateExec should not be removed because
204+
// they have different logical links.
205+
assertProjectExec(query, enabled = 2, disabled = 3)
206+
}
207+
}

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

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -720,7 +720,22 @@ class VeloxTestSettings extends BackendTestSettings {
720720
disableSuite[GlutenProjectedOrderingAndPartitioningSuite](
721721
"Validates Spark planner output ordering and partitioning metadata")
722722
enableSuite[GlutenQueryPlanningTrackerEndToEndSuite]
723-
// TODO: 4.x enableSuite[GlutenRemoveRedundantProjectsSuite] // 14 failures
723+
enableSuite[GlutenRemoveRedundantProjectsSuite]
724+
// Rewrite as result checks because Gluten transforms and may pull out additional projects.
725+
.exclude("project with filter")
726+
.exclude("project with specific column ordering")
727+
.exclude("project with extra columns")
728+
.exclude("project with fewer columns")
729+
.exclude("aggregate without ordering requirement")
730+
.exclude("aggregate with ordering requirement")
731+
.exclude("join without ordering requirement")
732+
.exclude("join with ordering requirement")
733+
.exclude("window function")
734+
.exclude("generate should require column ordering")
735+
.exclude("subquery")
736+
.exclude("SPARK-33697: UnionExec should require column ordering")
737+
.exclude("SPARK-33697: remove redundant projects under expand")
738+
.exclude("SPARK-36020: Project should not be removed when child's logical link is different")
724739
enableSuite[GlutenRemoveRedundantSortsSuite]
725740
// Rewrite as it check spark SortExec.
726741
.includeAllGlutenTests()

0 commit comments

Comments
 (0)