Copilot commented on code in PR #12506:
URL: https://github.com/apache/gluten/pull/12506#discussion_r3584241285
##########
gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantProjectsSuite.scala:
##########
@@ -16,8 +16,192 @@
*/
package org.apache.spark.sql.execution
-import org.apache.spark.sql.GlutenSQLTestsTrait
+import org.apache.spark.sql.{GlutenSQLTestsTrait, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
class GlutenRemoveRedundantProjectsSuite
extends RemoveRedundantProjectsSuite
- with GlutenSQLTestsTrait {}
+ with GlutenSQLTestsTrait {
+
+ // The original tests count Spark ProjectExec nodes, while Gluten converts
offloaded projects to
+ // ProjectExecTransformer. PullOutPreProject and PullOutPostProject may also
insert additional
+ // projects while rewriting the Spark physical plan, so project counts are
not directly
+ // comparable. Therefore, these tests only verify that query results are
identical with redundant
+ // project removal enabled and disabled.
+ private def assertProjectExec(query: String, enabled: Int, disabled: Int):
Unit = {
+ val df = sql(query)
+ // When enabling AQE, the DPP subquery filters is replaced in runtime.
+ df.collect()
+ // assertProjectExecCount(df, enabled)
+ val result = df.collect()
+ withSQLConf(SQLConf.REMOVE_REDUNDANT_PROJECTS_ENABLED.key -> "false") {
+ val df2 = sql(query)
+ df2.collect()
+ // assertProjectExecCount(df2, disabled)
+ checkAnswer(df2, result)
+ }
+ }
+
+ testGluten("project with filter") {
+ val query = "select * from testView where a > 5"
+ assertProjectExec(query, 0, 1)
+ }
+
+ testGluten("project with specific column ordering") {
+ val query = "select key, a, b, c from testView"
+ assertProjectExec(query, 1, 1)
+ }
+
+ testGluten("project with extra columns") {
+ val query = "select a, b, c, key, a from testView"
+ assertProjectExec(query, 1, 1)
+ }
+
+ testGluten("project with fewer columns") {
+ val query = "select a from testView where a > 3"
+ assertProjectExec(query, 1, 1)
+ }
+
+ testGluten("aggregate without ordering requirement") {
+ val query = "select sum(a) as sum_a, key, last(b) as last_b " +
+ "from (select key, a, b from testView where a > 100) group by key"
+ assertProjectExec(query, 0, 1)
+ }
+
+ testGluten("aggregate with ordering requirement") {
+ val query = "select a, sum(b) as sum_b from testView group by a"
+ assertProjectExec(query, 1, 1)
+ }
+
+ testGluten("join without ordering requirement") {
+ val query = "select t1.key, t2.key, t1.a, t2.b from (select key, a, b, c
from testView)" +
+ " as t1 join (select key, a, b, c from testView) as t2 on t1.c > t2.c
and t1.key > 10"
+ assertProjectExec(query, 1, 3)
+ }
+
+ testGluten("join with ordering requirement") {
+ val query = "select * from (select key, a, c, b from testView) as t1 join
" +
+ "(select key, a, b, c from testView) as t2 on t1.key = t2.key where t2.a
> 50"
+ assertProjectExec(query, 2, 2)
+ }
+
+ testGluten("window function") {
+ val query = "select key, b, avg(a) over (partition by key order by a " +
+ "rows between 1 preceding and 1 following) as avg from testView"
+ assertProjectExec(query, 1, 2)
+ }
+
+ testGluten("generate should require column ordering") {
+ withTempView("testData") {
+ spark.range(0, 10, 1)
+ .selectExpr("id as key", "id * 2 as a", "id * 3 as b")
+ .createOrReplaceTempView("testData")
+
+ val data = sql("select key, a, b, count(*) from testData group by key,
a, b limit 2")
+ val df = data.selectExpr("a", "b", "key", "explode(array(key, a, b)) as
d").filter("d > 0")
+ df.collect()
+ val plan = df.queryExecution.executedPlan
+ val numProjects = collectWithSubqueries(plan) { case p: ProjectExec => p
}.length
+
+ // Create a new plan that reverse the GenerateExec output and add a new
ProjectExec between
+ // GenerateExec and its child. This is to test if the ProjectExec is
removed, the output of
+ // the query will be incorrect.
+ val newPlan = stripAQEPlan(plan).transform {
+ case g @ GenerateExec(_, requiredChildOutput, _, _, child) =>
+ g.copy(
+ requiredChildOutput = requiredChildOutput.reverse,
+ child = ProjectExec(requiredChildOutput.reverse, child))
+ }
+
+ // Re-apply remove redundant project rule.
+ val rule = RemoveRedundantProjects
+ val newExecutedPlan = rule.apply(newPlan)
+ // The manually added ProjectExec node shouldn't be removed.
+ // assert(collectWithSubqueries(newExecutedPlan) {
+ // case p: ProjectExec => p
+ // }.size == numProjects + 1)
+
+ // Check the original plan's output and the new plan's output are the
same.
Review Comment:
numProjects and the commented-out assertion block are dead code now
(numProjects is never used and the assertion is fully commented). Removing them
makes the test easier to follow and avoids confusion about what is actually
being validated.
##########
gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantProjectsSuite.scala:
##########
@@ -16,8 +16,192 @@
*/
package org.apache.spark.sql.execution
-import org.apache.spark.sql.GlutenSQLTestsTrait
+import org.apache.spark.sql.{GlutenSQLTestsTrait, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
class GlutenRemoveRedundantProjectsSuite
extends RemoveRedundantProjectsSuite
- with GlutenSQLTestsTrait {}
+ with GlutenSQLTestsTrait {
+
+ // The original tests count Spark ProjectExec nodes, while Gluten converts
offloaded projects to
+ // ProjectExecTransformer. PullOutPreProject and PullOutPostProject may also
insert additional
+ // projects while rewriting the Spark physical plan, so project counts are
not directly
+ // comparable. Therefore, these tests only verify that query results are
identical with redundant
+ // project removal enabled and disabled.
+ private def assertProjectExec(query: String, enabled: Int, disabled: Int):
Unit = {
+ val df = sql(query)
+ // When enabling AQE, the DPP subquery filters is replaced in runtime.
+ df.collect()
+ // assertProjectExecCount(df, enabled)
+ val result = df.collect()
+ withSQLConf(SQLConf.REMOVE_REDUNDANT_PROJECTS_ENABLED.key -> "false") {
+ val df2 = sql(query)
+ df2.collect()
+ // assertProjectExecCount(df2, disabled)
+ checkAnswer(df2, result)
+ }
+ }
Review Comment:
assertProjectExec triggers multiple redundant Spark actions (df.collect
twice, df2.collect, and then checkAnswer runs actions again). This makes the
UTs significantly slower and increases flakiness risk under AQE; you can get
the enabled-side result with a single collect and let checkAnswer execute the
disabled side.
##########
gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantProjectsSuite.scala:
##########
@@ -16,8 +16,192 @@
*/
package org.apache.spark.sql.execution
-import org.apache.spark.sql.GlutenSQLTestsTrait
+import org.apache.spark.sql.{GlutenSQLTestsTrait, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
class GlutenRemoveRedundantProjectsSuite
extends RemoveRedundantProjectsSuite
- with GlutenSQLTestsTrait {}
+ with GlutenSQLTestsTrait {
+
+ // The original tests count Spark ProjectExec nodes, while Gluten converts
offloaded projects to
+ // ProjectExecTransformer. PullOutPreProject and PullOutPostProject may also
insert additional
+ // projects while rewriting the Spark physical plan, so project counts are
not directly
+ // comparable. Therefore, these tests only verify that query results are
identical with redundant
+ // project removal enabled and disabled.
+ private def assertProjectExec(query: String, enabled: Int, disabled: Int):
Unit = {
+ val df = sql(query)
+ // When enabling AQE, the DPP subquery filters is replaced in runtime.
+ df.collect()
+ // assertProjectExecCount(df, enabled)
+ val result = df.collect()
+ withSQLConf(SQLConf.REMOVE_REDUNDANT_PROJECTS_ENABLED.key -> "false") {
+ val df2 = sql(query)
+ df2.collect()
+ // assertProjectExecCount(df2, disabled)
+ checkAnswer(df2, result)
+ }
+ }
Review Comment:
assertProjectExec triggers multiple redundant Spark actions (df.collect
twice, df2.collect, and then checkAnswer runs actions again). This makes the
UTs significantly slower and increases flakiness risk under AQE; you can get
the enabled-side result with a single collect and let checkAnswer execute the
disabled side.
##########
gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala:
##########
@@ -716,7 +716,22 @@ class VeloxTestSettings extends BackendTestSettings {
// TODO: 4.x enableSuite[GlutenPlannerSuite] // 1 failure
// TODO: 4.x enableSuite[GlutenProjectedOrderingAndPartitioningSuite] // 6
failures
enableSuite[GlutenQueryPlanningTrackerEndToEndSuite]
- // TODO: 4.x enableSuite[GlutenRemoveRedundantProjectsSuite] // 14 failures
+ enableSuite[GlutenRemoveRedundantProjectsSuite]
+ // Rewrite as result checks because Gluten transforms and may pull out
additional projects.
+ .exclude("project with filter")
+ .exclude("project with specific column ordering")
+ .exclude("project with extra columns")
+ .exclude("project with fewer columns")
+ .exclude("aggregate without ordering requirement")
+ .exclude("aggregate with ordering requirement")
+ .exclude("join without ordering requirement")
+ .exclude("join with ordering requirement")
+ .exclude("window function")
+ .exclude("generate should require column ordering")
+ .exclude("subquery")
+ .exclude("SPARK-33697: UnionExec should require column ordering")
+ .exclude("SPARK-33697: remove redundant projects under expand")
+ .exclude("SPARK-36020: Project should not be removed when child's logical
link is different")
Review Comment:
This suite is enabled but relies on an explicit exclude list of upstream
Spark test names. That approach is brittle: if Spark adds new tests to
RemoveRedundantProjectsSuite, they will start running here unintentionally.
Using includeAllGlutenTests() is more robust and documents the intent to only
run Gluten-prefixed replacement tests.
##########
gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala:
##########
@@ -737,7 +737,22 @@ class VeloxTestSettings extends BackendTestSettings {
// TODO: 4.x enableSuite[GlutenPlannerSuite] // 1 failure
// TODO: 4.x enableSuite[GlutenProjectedOrderingAndPartitioningSuite] // 6
failures
enableSuite[GlutenQueryPlanningTrackerEndToEndSuite]
- // TODO: 4.x enableSuite[GlutenRemoveRedundantProjectsSuite] // 14 failures
+ enableSuite[GlutenRemoveRedundantProjectsSuite]
+ // Rewrite as result checks because Gluten transforms and may pull out
additional projects.
+ .exclude("project with filter")
+ .exclude("project with specific column ordering")
+ .exclude("project with extra columns")
+ .exclude("project with fewer columns")
+ .exclude("aggregate without ordering requirement")
+ .exclude("aggregate with ordering requirement")
+ .exclude("join without ordering requirement")
+ .exclude("join with ordering requirement")
+ .exclude("window function")
+ .exclude("generate should require column ordering")
+ .exclude("subquery")
+ .exclude("SPARK-33697: UnionExec should require column ordering")
+ .exclude("SPARK-33697: remove redundant projects under expand")
+ .exclude("SPARK-36020: Project should not be removed when child's logical
link is different")
Review Comment:
This suite is enabled but relies on an explicit exclude list of upstream
Spark test names. That approach is brittle: if Spark adds new tests to
RemoveRedundantProjectsSuite, they will start running here unintentionally.
Using includeAllGlutenTests() is more robust and documents the intent to only
run Gluten-prefixed replacement tests.
##########
gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantProjectsSuite.scala:
##########
@@ -16,8 +16,192 @@
*/
package org.apache.spark.sql.execution
-import org.apache.spark.sql.GlutenSQLTestsTrait
+import org.apache.spark.sql.{GlutenSQLTestsTrait, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
class GlutenRemoveRedundantProjectsSuite
extends RemoveRedundantProjectsSuite
- with GlutenSQLTestsTrait {}
+ with GlutenSQLTestsTrait {
+
+ // The original tests count Spark ProjectExec nodes, while Gluten converts
offloaded projects to
+ // ProjectExecTransformer. PullOutPreProject and PullOutPostProject may also
insert additional
+ // projects while rewriting the Spark physical plan, so project counts are
not directly
+ // comparable. Therefore, these tests only verify that query results are
identical with redundant
+ // project removal enabled and disabled.
+ private def assertProjectExec(query: String, enabled: Int, disabled: Int):
Unit = {
+ val df = sql(query)
+ // When enabling AQE, the DPP subquery filters is replaced in runtime.
+ df.collect()
+ // assertProjectExecCount(df, enabled)
+ val result = df.collect()
+ withSQLConf(SQLConf.REMOVE_REDUNDANT_PROJECTS_ENABLED.key -> "false") {
+ val df2 = sql(query)
+ df2.collect()
+ // assertProjectExecCount(df2, disabled)
+ checkAnswer(df2, result)
+ }
+ }
+
+ testGluten("project with filter") {
+ val query = "select * from testView where a > 5"
+ assertProjectExec(query, 0, 1)
+ }
+
+ testGluten("project with specific column ordering") {
+ val query = "select key, a, b, c from testView"
+ assertProjectExec(query, 1, 1)
+ }
+
+ testGluten("project with extra columns") {
+ val query = "select a, b, c, key, a from testView"
+ assertProjectExec(query, 1, 1)
+ }
+
+ testGluten("project with fewer columns") {
+ val query = "select a from testView where a > 3"
+ assertProjectExec(query, 1, 1)
+ }
+
+ testGluten("aggregate without ordering requirement") {
+ val query = "select sum(a) as sum_a, key, last(b) as last_b " +
+ "from (select key, a, b from testView where a > 100) group by key"
+ assertProjectExec(query, 0, 1)
+ }
+
+ testGluten("aggregate with ordering requirement") {
+ val query = "select a, sum(b) as sum_b from testView group by a"
+ assertProjectExec(query, 1, 1)
+ }
+
+ testGluten("join without ordering requirement") {
+ val query = "select t1.key, t2.key, t1.a, t2.b from (select key, a, b, c
from testView)" +
+ " as t1 join (select key, a, b, c from testView) as t2 on t1.c > t2.c
and t1.key > 10"
+ assertProjectExec(query, 1, 3)
+ }
+
+ testGluten("join with ordering requirement") {
+ val query = "select * from (select key, a, c, b from testView) as t1 join
" +
+ "(select key, a, b, c from testView) as t2 on t1.key = t2.key where t2.a
> 50"
+ assertProjectExec(query, 2, 2)
+ }
+
+ testGluten("window function") {
+ val query = "select key, b, avg(a) over (partition by key order by a " +
+ "rows between 1 preceding and 1 following) as avg from testView"
+ assertProjectExec(query, 1, 2)
+ }
+
+ testGluten("generate should require column ordering") {
+ withTempView("testData") {
+ spark.range(0, 10, 1)
+ .selectExpr("id as key", "id * 2 as a", "id * 3 as b")
+ .createOrReplaceTempView("testData")
+
+ val data = sql("select key, a, b, count(*) from testData group by key,
a, b limit 2")
+ val df = data.selectExpr("a", "b", "key", "explode(array(key, a, b)) as
d").filter("d > 0")
+ df.collect()
+ val plan = df.queryExecution.executedPlan
+ val numProjects = collectWithSubqueries(plan) { case p: ProjectExec => p
}.length
+
+ // Create a new plan that reverse the GenerateExec output and add a new
ProjectExec between
+ // GenerateExec and its child. This is to test if the ProjectExec is
removed, the output of
+ // the query will be incorrect.
+ val newPlan = stripAQEPlan(plan).transform {
+ case g @ GenerateExec(_, requiredChildOutput, _, _, child) =>
+ g.copy(
+ requiredChildOutput = requiredChildOutput.reverse,
+ child = ProjectExec(requiredChildOutput.reverse, child))
+ }
+
+ // Re-apply remove redundant project rule.
+ val rule = RemoveRedundantProjects
+ val newExecutedPlan = rule.apply(newPlan)
+ // The manually added ProjectExec node shouldn't be removed.
+ // assert(collectWithSubqueries(newExecutedPlan) {
+ // case p: ProjectExec => p
+ // }.size == numProjects + 1)
+
+ // Check the original plan's output and the new plan's output are the
same.
Review Comment:
numProjects and the commented-out assertion block are dead code now
(numProjects is never used and the assertion is fully commented). Removing them
makes the test easier to follow and avoids confusion about what is actually
being validated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
