This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 80929d6b549 [SPARK-38832][SQL][FOLLOWUP] Support propagate empty
expression set for distinct key
80929d6b549 is described below
commit 80929d6b549dfc61ade130a9d59dfa1abe72d681
Author: ulysses-you <ulyssesyo...@gmail.com>
AuthorDate: Fri Apr 22 18:17:04 2022 +0800
[SPARK-38832][SQL][FOLLOWUP] Support propagate empty expression set for
distinct key
### What changes were proposed in this pull request?
- Improve `DistinctKeyVisitor` that support propagate empty set
- Small improvement for match alias
### Why are the changes needed?
Make distinct keys can be used to optimize more case, see comment
https://github.com/apache/spark/pull/36117#discussion_r853755920
### Does this PR introduce _any_ user-facing change?
Improve performance
### How was this patch tested?
add test
Closes #36281 from ulysses-you/SPARK-38832-followup.
Authored-by: ulysses-you <ulyssesyo...@gmail.com>
Signed-off-by: Wenchen Fan <wenc...@databricks.com>
---
.../plans/logical/DistinctKeyVisitor.scala | 25 ++++++++++------------
.../plans/logical/LogicalPlanDistinctKeys.scala | 8 +------
.../plans/logical/DistinctKeyVisitorSuite.scala | 6 +++++-
3 files changed, 17 insertions(+), 22 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/DistinctKeyVisitor.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/DistinctKeyVisitor.scala
index 5b25a326831..1f495688bc5 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/DistinctKeyVisitor.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/DistinctKeyVisitor.scala
@@ -29,25 +29,21 @@ object DistinctKeyVisitor extends
LogicalPlanVisitor[Set[ExpressionSet]] {
private def projectDistinctKeys(
keys: Set[ExpressionSet], projectList: Seq[NamedExpression]):
Set[ExpressionSet] = {
val outputSet = ExpressionSet(projectList.map(_.toAttribute))
- val aliases = projectList.filter(_.isInstanceOf[Alias])
+ val aliases = projectList.collect {
+ // TODO: Expand distinctKeys for redundant aliases on the same expression
+ case alias: Alias if alias.child.deterministic =>
alias.child.canonicalized -> alias
+ }.toMap
if (aliases.isEmpty) {
keys.filter(_.subsetOf(outputSet))
} else {
- val aliasedDistinctKeys = keys.map { expressionSet =>
- expressionSet.map { expression =>
- expression transform {
- case expr: Expression =>
- // TODO: Expand distinctKeys for redundant aliases on the same
expression
- aliases
- .collectFirst { case a: Alias if a.child.semanticEquals(expr)
=> a.toAttribute }
- .getOrElse(expr)
- }
- }
- }
+ val aliasedDistinctKeys = keys.map(_.map(_.transform {
+ case expr: Expression =>
+ aliases.get(expr.canonicalized).map(_.toAttribute).getOrElse(expr)
+ }))
aliasedDistinctKeys.collect {
case es: ExpressionSet if es.subsetOf(outputSet) => ExpressionSet(es)
} ++ keys.filter(_.subsetOf(outputSet))
- }.filter(_.nonEmpty)
+ }
}
/**
@@ -69,7 +65,8 @@ object DistinctKeyVisitor extends
LogicalPlanVisitor[Set[ExpressionSet]] {
override def default(p: LogicalPlan): Set[ExpressionSet] =
Set.empty[ExpressionSet]
override def visitAggregate(p: Aggregate): Set[ExpressionSet] = {
- val groupingExps = ExpressionSet(p.groupingExpressions) // handle group by
a, a
+ // handle group by a, a and global aggregate
+ val groupingExps = ExpressionSet(p.groupingExpressions)
projectDistinctKeys(addDistinctKey(p.child.distinctKeys, groupingExps),
p.aggregateExpressions)
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlanDistinctKeys.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlanDistinctKeys.scala
index 2ffa5a0e594..1843c2da478 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlanDistinctKeys.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlanDistinctKeys.scala
@@ -29,12 +29,6 @@ import
org.apache.spark.sql.internal.SQLConf.PROPAGATE_DISTINCT_KEYS_ENABLED
*/
trait LogicalPlanDistinctKeys { self: LogicalPlan =>
lazy val distinctKeys: Set[ExpressionSet] = {
- if (conf.getConf(PROPAGATE_DISTINCT_KEYS_ENABLED)) {
- val keys = DistinctKeyVisitor.visit(self)
- require(keys.forall(_.nonEmpty))
- keys
- } else {
- Set.empty
- }
+ if (conf.getConf(PROPAGATE_DISTINCT_KEYS_ENABLED))
DistinctKeyVisitor.visit(self) else Set.empty
}
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/DistinctKeyVisitorSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/DistinctKeyVisitorSuite.scala
index bbf69ae622b..55ce5dcd72a 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/DistinctKeyVisitorSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/DistinctKeyVisitorSuite.scala
@@ -61,7 +61,11 @@ class DistinctKeyVisitorSuite extends PlanTest {
checkDistinctAttributes(t1.groupBy($"a")($"a", max($"b")),
Set(ExpressionSet(Seq(a))))
checkDistinctAttributes(t1.groupBy($"a", $"b")($"a", $"b", d, e),
Set(ExpressionSet(Seq(a, b)), ExpressionSet(Seq(d.toAttribute,
e.toAttribute))))
- checkDistinctAttributes(t1.groupBy()(sum($"c")), Set.empty)
+ checkDistinctAttributes(t1.groupBy()(sum($"c")), Set(ExpressionSet()))
+ // ExpressionSet() is a subset of anything, so we do not need
ExpressionSet(c2)
+ checkDistinctAttributes(t1.groupBy()(sum($"c") as
"c2").groupBy($"c2")("c2"),
+ Set(ExpressionSet()))
+ checkDistinctAttributes(t1.groupBy()(), Set(ExpressionSet()))
checkDistinctAttributes(t1.groupBy($"a")($"a", $"a" % 10, d, sum($"b")),
Set(ExpressionSet(Seq(a)), ExpressionSet(Seq(d.toAttribute))))
checkDistinctAttributes(t1.groupBy(f.child, $"b")(f, $"b", sum($"c")),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
