You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2023/02/24 05:54:20 UTC
[spark] branch branch-3.4 updated: [SPARK-42049][SQL][FOLLOWUP] Always filter away invalid ordering/partitioning
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.4 by this push:
new 149458c50d1 [SPARK-42049][SQL][FOLLOWUP] Always filter away invalid ordering/partitioning
149458c50d1 is described below
commit 149458c50d1086409bcfda3532d078b05efbfb92
Author: Wenchen Fan <we...@databricks.com>
AuthorDate: Thu Feb 23 21:54:00 2023 -0800
[SPARK-42049][SQL][FOLLOWUP] Always filter away invalid ordering/partitioning
### What changes were proposed in this pull request?
This is a follow-up of https://github.com/apache/spark/pull/37525 . When the project list has aliases, we go to the `projectExpression` branch which filters away invalid partitioning/ordering that reference non-existing attributes in the current plan node. However, this filtering is missing when the project list has no alias, where we directly return the child partitioning/ordering.
This PR fixes it.
### Why are the changes needed?
to make sure we always return valid output partitioning/ordering.
### Does this PR introduce _any_ user-facing change?
no
### How was this patch tested?
new tests
Closes #40137 from cloud-fan/alias.
Authored-by: Wenchen Fan <we...@databricks.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 72922adc8a78e8d31f03205a148b89291a9a4d19)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../catalyst/plans/AliasAwareOutputExpression.scala | 18 +++++++++++++++---
.../sql/execution/AliasAwareOutputExpression.scala | 20 +++++++++++++-------
.../apache/spark/sql/execution/PlannerSuite.scala | 7 ++++---
.../ProjectedOrderingAndPartitioningSuite.scala | 14 ++++++++++++++
4 files changed, 46 insertions(+), 13 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/AliasAwareOutputExpression.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/AliasAwareOutputExpression.scala
index 4d9d69d14fe..cfe22994592 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/AliasAwareOutputExpression.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/AliasAwareOutputExpression.scala
@@ -95,7 +95,7 @@ trait AliasAwareQueryOutputOrdering[T <: QueryPlan[T]]
}
override final def outputOrdering: Seq[SortOrder] = {
- if (hasAlias) {
+ val newOrdering: Iterator[Option[SortOrder]] = if (hasAlias) {
// Take the first `SortOrder`s only until they can be projected.
// E.g. we have child ordering `Seq(SortOrder(a), SortOrder(b))` then
// if only `a AS x` can be projected then we can return Seq(SortOrder(x))`
@@ -112,9 +112,21 @@ trait AliasAwareQueryOutputOrdering[T <: QueryPlan[T]]
} else {
None
}
- }.takeWhile(_.isDefined).flatten.toSeq
+ }
} else {
- orderingExpressions
+ // Make sure the returned ordering are valid (only reference output attributes of the current
+ // plan node). Same as above (the if branch), we take the first ordering expressions that are
+ // all valid.
+ val outputSet = AttributeSet(outputExpressions.map(_.toAttribute))
+ orderingExpressions.iterator.map { order =>
+ val validChildren = order.children.filter(_.references.subsetOf(outputSet))
+ if (validChildren.nonEmpty) {
+ Some(order.copy(child = validChildren.head, sameOrderExpressions = validChildren.tail))
+ } else {
+ None
+ }
+ }
}
+ newOrdering.takeWhile(_.isDefined).flatten.toSeq
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala
index a09c719cf84..e1dcab80af3 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala
@@ -18,7 +18,7 @@ package org.apache.spark.sql.execution
import scala.collection.mutable
-import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.expressions.{AttributeSet, Expression}
import org.apache.spark.sql.catalyst.plans.{AliasAwareOutputExpression, AliasAwareQueryOutputOrdering}
import org.apache.spark.sql.catalyst.plans.physical.{Partitioning, PartitioningCollection, UnknownPartitioning}
@@ -29,7 +29,7 @@ import org.apache.spark.sql.catalyst.plans.physical.{Partitioning, PartitioningC
trait PartitioningPreservingUnaryExecNode extends UnaryExecNode
with AliasAwareOutputExpression {
final override def outputPartitioning: Partitioning = {
- if (hasAlias) {
+ val partitionings: Seq[Partitioning] = if (hasAlias) {
flattenPartitioning(child.outputPartitioning).flatMap {
case e: Expression =>
// We need unique partitionings but if the input partitioning is
@@ -44,13 +44,19 @@ trait PartitioningPreservingUnaryExecNode extends UnaryExecNode
.take(aliasCandidateLimit)
.asInstanceOf[Stream[Partitioning]]
case o => Seq(o)
- } match {
- case Seq() => UnknownPartitioning(child.outputPartitioning.numPartitions)
- case Seq(p) => p
- case ps => PartitioningCollection(ps)
}
} else {
- child.outputPartitioning
+ // Filter valid partitiongs (only reference output attributes of the current plan node)
+ val outputSet = AttributeSet(outputExpressions.map(_.toAttribute))
+ flattenPartitioning(child.outputPartitioning).filter {
+ case e: Expression => e.references.subsetOf(outputSet)
+ case _ => true
+ }
+ }
+ partitionings match {
+ case Seq() => UnknownPartitioning(child.outputPartitioning.numPartitions)
+ case Seq(p) => p
+ case ps => PartitioningCollection(ps)
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
index e9cb77ec95c..4b3d3a4b805 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
@@ -1129,9 +1129,10 @@ class PlannerSuite extends SharedSparkSession with AdaptiveSparkPlanHelper {
assert(sortNodes.size == 3)
val outputOrdering = planned.outputOrdering
assert(outputOrdering.size == 1)
- // Sort order should have 3 childrens, not 4. This is because t1.id*2 and 2*t1.id are same
- assert(outputOrdering.head.children.size == 3)
- assert(outputOrdering.head.children.count(_.isInstanceOf[AttributeReference]) == 2)
+ // Sort order should have 2 childrens, not 4. This is because t1.id*2 and 2*t1.id are same
+ // and t2.id is not a valid ordering (the final plan doesn't output t2.id)
+ assert(outputOrdering.head.children.size == 2)
+ assert(outputOrdering.head.children.count(_.isInstanceOf[AttributeReference]) == 1)
assert(outputOrdering.head.children.count(_.isInstanceOf[Multiply]) == 1)
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
index e4ecdb9c445..f5839e99756 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
@@ -177,4 +177,18 @@ class ProjectedOrderingAndPartitioningSuite
val outputOrdering3 = df3.queryExecution.optimizedPlan.outputOrdering
assert(outputOrdering3.size == 0)
}
+
+ test("SPARK-42049: Improve AliasAwareOutputExpression - no alias but still prune expressions") {
+ val df = spark.range(2).select($"id" + 1 as "a", $"id" + 2 as "b")
+
+ val df1 = df.repartition($"a", $"b").selectExpr("a")
+ val outputPartitioning = stripAQEPlan(df1.queryExecution.executedPlan).outputPartitioning
+ assert(outputPartitioning.isInstanceOf[UnknownPartitioning])
+
+ val df2 = df.orderBy("a", "b").select("a")
+ val outputOrdering = df2.queryExecution.optimizedPlan.outputOrdering
+ assert(outputOrdering.size == 1)
+ assert(outputOrdering.head.child.asInstanceOf[Attribute].name == "a")
+ assert(outputOrdering.head.sameOrderExpressions.size == 0)
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org