[GLUTEN-11550][VL] Fix GlutenRemoveRedundantProjectsSuite in Spark 4.x#12506
Open
zml1206 wants to merge 3 commits into
Open
[GLUTEN-11550][VL] Fix GlutenRemoveRedundantProjectsSuite in Spark 4.x#12506zml1206 wants to merge 3 commits into
zml1206 wants to merge 3 commits into
Conversation
zml1206
commented
Jul 14, 2026
| enableSuite[GlutenPersistedViewTestSuite] | ||
| // TODO: 4.x enableSuite[GlutenPlannerSuite] // 1 failure | ||
| // TODO: 4.x enableSuite[GlutenProjectedOrderingAndPartitioningSuite] // 6 failures | ||
| // GlutenPlannerSuite is not enabled: it validates Spark planner implementation details. |
Contributor
Author
There was a problem hiding this comment.
The failures fall into two categories:
7 failures occur because the original Spark tests inspect native Spark physical operators, which are replaced by Gluten transformer operators.
8 failures are caused by spark.sql.shuffle.partitions=1 in GlutenSQLTestsBaseTrait, which changes the expected partitioning, exchange, and sort plans.
zml1206
commented
Jul 14, 2026
Comment on lines
+717
to
+718
| // GlutenProjectedOrderingAndPartitioningSuite is not enabled: it validates Spark planner | ||
| // output ordering and partitioning metadata. |
Contributor
Author
There was a problem hiding this comment.
6 failures are caused by spark.sql.shuffle.partitions=1 in GlutenSQLTestsBaseTrait, which changes the expected partitioning, exchange, and sort plans.
Contributor
There was a problem hiding this comment.
Pull request overview
Updates the Spark 4.0/4.1 Velox test configuration and GlutenRemoveRedundantProjectsSuite to avoid relying on Spark ProjectExec node counts (which diverge under Gluten due to transformer nodes and plan rewrites), aiming to validate correctness via result equivalence instead.
Changes:
- Reworks
GlutenRemoveRedundantProjectsSuite(Spark 4.0/4.1) to validate query results with redundant project removal enabled vs disabled. - Enables
GlutenRemoveRedundantProjectsSuitein Spark 4.0/4.1 Velox test settings and attempts to exclude the upstream Spark test cases that assert planner-node counts. - Clarifies why certain Spark-planner-detail suites remain disabled in Spark 4.x Velox settings.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 6 comments.
| File | Description |
|---|---|
| gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantProjectsSuite.scala | Adds Gluten-specific result-based assertions intended to replace Spark’s plan-node-count assertions. |
| gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala | Enables the suite and excludes upstream Spark test names; updates comments about other disabled suites. |
| gluten-ut/spark40/src/test/scala/org/apache/spark/sql/execution/GlutenRemoveRedundantProjectsSuite.scala | Same as Spark 4.1 variant, for Spark 4.0. |
| gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala | Same enable/exclude configuration as Spark 4.1 variant, for Spark 4.0. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+32
to
+44
| 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) | ||
| } | ||
| } |
Comment on lines
+32
to
+44
| 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) | ||
| } | ||
| } |
Comment on lines
+105
to
+123
| 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) |
Comment on lines
+105
to
+123
| 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) |
Comment on lines
+720
to
+734
| enableSuite[GlutenRemoveRedundantProjectsSuite] | ||
| .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") |
Comment on lines
+741
to
+755
| enableSuite[GlutenRemoveRedundantProjectsSuite] | ||
| .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") |
Comment on lines
+32
to
+44
| 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) | ||
| } | ||
| } |
Comment on lines
+104
to
+105
| val plan = df.queryExecution.executedPlan | ||
| val numProjects = collectWithSubqueries(plan) { case p: ProjectExec => p }.length |
Comment on lines
+118
to
+123
| 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) |
Comment on lines
+720
to
+735
| 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") |
Comment on lines
+741
to
+756
| 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") |
| // 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. |
philo-he
reviewed
Jul 14, 2026
| // TODO: 4.x enableSuite[GlutenPlannerSuite] // 1 failure | ||
| // TODO: 4.x enableSuite[GlutenProjectedOrderingAndPartitioningSuite] // 6 failures | ||
| // GlutenPlannerSuite is not enabled: it validates Spark planner implementation details. | ||
| // GlutenProjectedOrderingAndPartitioningSuite is not enabled: it validates Spark planner |
Member
There was a problem hiding this comment.
@zml1206, Could we add a disableSuite[X]("reason") method in BackendTestSettings.scala to track intentionally disabled suites? This would be more explicit and easier to track than the current comment-based approach. Today we can get a similar effect with enableSuite[X].disable("reason"), but that reads oddly: we enable a suite and then immediately disable it. A dedicated disableSuite would state the intent directly.
private val disabledSuites: java.util.Map[String, String] = new util.HashMap() // name -> reason
protected def disableSuite[T: ClassTag](reason: String): Unit =
disableSuite(implicitly[ClassTag[T]].runtimeClass.getCanonicalName, reason)
protected def disableSuite(suiteName: String, reason: String): Unit = {
require(reason.nonEmpty, "Disable reason must not be empty")
if (enabledSuites.containsKey(suiteName)) {
throw new IllegalArgumentException("Suite is already enabled: " + suiteName)
}
if (disabledSuites.containsKey(suiteName)) {
throw new IllegalArgumentException("Duplicated disabled suite: " + suiteName)
}
disabledSuites.put(suiteName, reason)
}
Contributor
Author
There was a problem hiding this comment.
Thanks, @philo-he that's a good suggestion, but I'd like to handle it consistently in the next PR, since there are existing instances that need to be addressed in the same way.
Comment on lines
+32
to
+44
| 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) | ||
| } | ||
| } |
Comment on lines
+103
to
+125
| 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. |
Comment on lines
+719
to
+734
| 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") |
Comment on lines
+32
to
+44
| 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) | ||
| } | ||
| } |
Comment on lines
+103
to
+125
| 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. |
Comment on lines
+740
to
+755
| 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") |
zml1206
changed the title
zml1206
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Related issue: #11550