You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2020/12/17 05:54:40 UTC
[spark] branch branch-3.1 updated: [SPARK-33697][SQL]
RemoveRedundantProjects should require column ordering by default
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new 6b0a48e [SPARK-33697][SQL] RemoveRedundantProjects should require column ordering by default
6b0a48e is described below
commit 6b0a48ee71980941081f2b261d0f8149df3cd527
Author: allisonwang-db <66...@users.noreply.github.com>
AuthorDate: Thu Dec 17 05:47:44 2020 +0000
[SPARK-33697][SQL] RemoveRedundantProjects should require column ordering by default
### What changes were proposed in this pull request?
This PR changes the rule `RemoveRedundantProjects` from by default passing column ordering requirements from parent nodes to always require column orders regardless of the requirements from parent nodes unless otherwise specified. More specifically, instead of excluding a few nodes like GenerateExec, UnionExec that are known to require children columns to be ordered, the rule now includes a whitelist of nodes that allow passing through the ordering requirements from their parents.
### Why are the changes needed?
Currently, this rule passes through ordering requirements from parents directly to children except for a few excluded nodes. This incorrectly removes the necessary project nodes below a UnionExec since it is not excluded. An earlier PR also fixed a similar issue for GenerateExec (SPARK-32861). In order to prevent similar issues, the rule should be changed to always require column ordering except for a few specific nodes that we know for sure can pass through the requirements.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Unit tests
Closes #30659 from allisonwang-db/spark-33697-remove-project-union.
Authored-by: allisonwang-db <66...@users.noreply.github.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 1e85707738a830d33598ca267a6740b3f06b1861)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../sql/execution/RemoveRedundantProjects.scala | 24 ++++++++---
.../execution/RemoveRedundantProjectsSuite.scala | 47 ++++++++++++++++++++++
2 files changed, 65 insertions(+), 6 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/RemoveRedundantProjects.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/RemoveRedundantProjects.scala
index bbe3f50..bfb6e80 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/RemoveRedundantProjects.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/RemoveRedundantProjects.scala
@@ -22,6 +22,8 @@ import org.apache.spark.sql.catalyst.expressions.aggregate.{Final, PartialMerge}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanExecBase
+import org.apache.spark.sql.execution.joins.BaseJoinExec
+import org.apache.spark.sql.execution.window.WindowExec
import org.apache.spark.sql.internal.SQLConf
/**
@@ -61,16 +63,26 @@ object RemoveRedundantProjects extends Rule[SparkPlan] {
val keepOrdering = a.aggregateExpressions
.exists(ae => ae.mode.equals(Final) || ae.mode.equals(PartialMerge))
a.mapChildren(removeProject(_, keepOrdering))
- // GenerateExec requires column ordering since it binds input rows directly with its
- // requiredChildOutput without using child's output schema.
- case g: GenerateExec => g.mapChildren(removeProject(_, true))
- // JoinExec ordering requirement will inherit from its parent. If there is no ProjectExec in
- // its ancestors, JoinExec should require output columns to be ordered.
- case o => o.mapChildren(removeProject(_, requireOrdering))
+ case o =>
+ val required = if (canPassThrough(o)) requireOrdering else true
+ o.mapChildren(removeProject(_, requireOrdering = required))
}
}
/**
+ * Check if the given node can pass the ordering requirement from its parent.
+ */
+ private def canPassThrough(plan: SparkPlan): Boolean = plan match {
+ case _: FilterExec => true
+ // JoinExec ordering requirement should inherit from its parent. If there is no ProjectExec in
+ // its ancestors, JoinExec should require output columns to be ordered, and vice versa.
+ case _: BaseJoinExec => true
+ case _: WindowExec => true
+ case _: ExpandExec => true
+ case _ => false
+ }
+
+ /**
* Check if the nullability change is positive. It catches the case when the project output
* attribute is not nullable, but the child output attribute is nullable.
*/
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantProjectsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantProjectsSuite.scala
index 2de9d21..040c518 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantProjectsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantProjectsSuite.scala
@@ -166,6 +166,53 @@ abstract class RemoveRedundantProjectsSuiteBase
assertProjectExec(query, 0, 1)
}
}
+
+ test("SPARK-33697: UnionExec should require column ordering") {
+ withTable("t1", "t2") {
+ spark.range(-10, 20)
+ .selectExpr(
+ "id",
+ "date_add(date '1950-01-01', cast(id as int)) as datecol",
+ "cast(id as string) strcol")
+ .write.mode("overwrite").format("parquet").saveAsTable("t1")
+ spark.range(-10, 20)
+ .selectExpr(
+ "cast(id as string) strcol",
+ "id",
+ "date_add(date '1950-01-01', cast(id as int)) as datecol")
+ .write.mode("overwrite").format("parquet").saveAsTable("t2")
+
+ val queryTemplate =
+ """
+ |SELECT DISTINCT datecol, strcol FROM
+ |(
+ |(SELECT datecol, id, strcol from t1)
+ | %s
+ |(SELECT datecol, id, strcol from t2)
+ |)
+ |""".stripMargin
+
+ Seq(("UNION", 2, 2), ("UNION ALL", 1, 2)).foreach { case (setOperation, enabled, disabled) =>
+ val query = queryTemplate.format(setOperation)
+ assertProjectExec(query, enabled = enabled, disabled = disabled)
+ }
+ }
+ }
+
+ test("SPARK-33697: remove redundant projects under expand") {
+ val query =
+ """
+ |SELECT t1.key, t2.key, sum(t1.a) AS s1, sum(t2.b) AS s2 FROM
+ |(SELECT a, key FROM testView) t1
+ |JOIN
+ |(SELECT b, key FROM testView) t2
+ |ON t1.key = t2.key
+ |GROUP BY t1.key, t2.key GROUPING SETS(t1.key, t2.key)
+ |ORDER BY t1.key, t2.key, s1, s2
+ |LIMIT 10
+ |""".stripMargin
+ assertProjectExec(query, 0, 3)
+ }
}
class RemoveRedundantProjectsSuite extends RemoveRedundantProjectsSuiteBase
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org