This is an automated email from the ASF dual-hosted git repository.
gengliang 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 d92018e [SPARK-35298][SQL] Migrate to transformWithPruning for rules
in Optimizer.scala
d92018e is described below
commit d92018ee358b0009dac626e2c5568db8363f53ee
Author: Yingyi Bu <yingyi...@databricks.com>
AuthorDate: Wed May 12 20:42:47 2021 +0800
[SPARK-35298][SQL] Migrate to transformWithPruning for rules in
Optimizer.scala
### What changes were proposed in this pull request?
Added the following TreePattern enums:
- ALIAS
- AND_OR
- AVERAGE
- GENERATE
- INTERSECT
- SORT
- SUM
- DISTINCT_LIKE
- PROJECT
- REPARTITION_OPERATION
- UNION
Added tree traversal pruning to the following rules in Optimizer.scala:
- EliminateAggregateFilter
- RemoveRedundantAggregates
- RemoveNoopOperators
- RemoveNoopUnion
- LimitPushDown
- ColumnPruning
- CollapseRepartition
- OptimizeRepartition
- OptimizeWindowFunctions
- CollapseWindow
- TransposeWindow
- InferFiltersFromGenerate
- InferFiltersFromConstraints
- CombineUnions
- CombineFilters
- EliminateSorts
- PruneFilters
- EliminateLimits
- DecimalAggregates
- ConvertToLocalRelation
- ReplaceDistinctWithAggregate
- ReplaceIntersectWithSemiJoin
- ReplaceExceptWithAntiJoin
- RewriteExceptAll
- RewriteIntersectAll
- RemoveLiteralFromGroupExpressions
- RemoveRepetitionFromGroupExpressions
- OptimizeLimitZero
### Why are the changes needed?
Reduce the number of tree traversals and hence improve the query
compilation latency.
perf diff:
Rule name | Total Time (baseline) | Total Time (experiment) |
experiment/baseline
RemoveRedundantAggregates | 51290766 | 67070477 | 1.31
RemoveNoopOperators | 192371141 | 196631275 | 1.02
RemoveNoopUnion | 49222561 | 43266681 | 0.88
LimitPushDown | 40885185 | 21672646 | 0.53
ColumnPruning | 2003406120 | 1285562149 | 0.64
CollapseRepartition | 40648048 | 72646515 | 1.79
OptimizeRepartition | 37813850 | 20600803 | 0.54
OptimizeWindowFunctions | 174426904 | 46741409 | 0.27
CollapseWindow | 38959957 | 24542426 | 0.63
TransposeWindow | 33533191 | 20414930 | 0.61
InferFiltersFromGenerate | 21758688 | 15597344 | 0.72
InferFiltersFromConstraints | 518009794 | 493282321 | 0.95
CombineUnions | 67694022 | 70550382 | 1.04
CombineFilters | 35265060 | 29005424 | 0.82
EliminateSorts | 57025509 | 19795776 | 0.35
PruneFilters | 433964815 | 465579200 | 1.07
EliminateLimits | 44275393 | 24476859 | 0.55
DecimalAggregates | 83143172 | 28816090 | 0.35
ReplaceDistinctWithAggregate | 21783760 | 18287489 | 0.84
ReplaceIntersectWithSemiJoin | 22311271 | 16566393 | 0.74
ReplaceExceptWithAntiJoin | 23838520 | 16588808 | 0.70
RewriteExceptAll | 32750296 | 29421957 | 0.90
RewriteIntersectAll | 29760454 | 21243599 | 0.71
RemoveLiteralFromGroupExpressions | 28151861 | 25270947 | 0.90
RemoveRepetitionFromGroupExpressions | 29587030 | 23447041 | 0.79
OptimizeLimitZero | 18081943 | 15597344 | 0.86
**Accumulated | 4129959311 | 3112676285 | 0.75**
### How was this patch tested?
Existing tests.
Closes #32439 from sigmod/optimizer.
Authored-by: Yingyi Bu <yingyi...@databricks.com>
Signed-off-by: Gengliang Wang <ltn...@gmail.com>
---
.../catalyst/expressions/aggregate/Average.scala | 3 +
.../sql/catalyst/expressions/aggregate/Sum.scala | 3 +
.../catalyst/expressions/namedExpressions.scala | 2 +
.../spark/sql/catalyst/optimizer/Optimizer.scala | 113 ++++++++++++++-------
.../plans/logical/basicLogicalOperators.scala | 10 ++
.../sql/catalyst/rules/RuleIdCollection.scala | 24 +++++
.../spark/sql/catalyst/trees/TreePatterns.scala | 11 +-
7 files changed, 128 insertions(+), 38 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Average.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Average.scala
index 8ae24e5..82ad2df 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Average.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Average.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.catalyst.expressions.aggregate
import org.apache.spark.sql.catalyst.analysis.{DecimalPrecision,
FunctionRegistry, TypeCheckResult}
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.trees.TreePattern.{AVERAGE, TreePattern}
import org.apache.spark.sql.catalyst.trees.UnaryLike
import org.apache.spark.sql.catalyst.util.TypeUtils
import org.apache.spark.sql.types._
@@ -51,6 +52,8 @@ case class Average(child: Expression) extends
DeclarativeAggregate with Implicit
// Return data type.
override def dataType: DataType = resultType
+ final override val nodePatterns: Seq[TreePattern] = Seq(AVERAGE)
+
private lazy val resultType = child.dataType match {
case DecimalType.Fixed(p, s) =>
DecimalType.bounded(p + 4, s + 4)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Sum.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Sum.scala
index 31150fc..16cd9d7 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Sum.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Sum.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.catalyst.expressions.aggregate
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.trees.TreePattern.{SUM, TreePattern}
import org.apache.spark.sql.catalyst.trees.UnaryLike
import org.apache.spark.sql.catalyst.util.TypeUtils
import org.apache.spark.sql.internal.SQLConf
@@ -52,6 +53,8 @@ case class Sum(child: Expression) extends
DeclarativeAggregate with ImplicitCast
override def checkInputDataTypes(): TypeCheckResult =
TypeUtils.checkForAnsiIntervalOrNumericType(child.dataType, "sum")
+ final override val nodePatterns: Seq[TreePattern] = Seq(SUM)
+
private lazy val resultType = child.dataType match {
case DecimalType.Fixed(precision, scale) =>
DecimalType.bounded(precision + 10, scale)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala
index 69f7d24..52487d4 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/namedExpressions.scala
@@ -158,6 +158,8 @@ case class Alias(child: Expression, name: String)(
val nonInheritableMetadataKeys: Seq[String] = Seq.empty)
extends UnaryExpression with NamedExpression {
+ final override val nodePatterns: Seq[TreePattern] = Seq(ALIAS)
+
// Alias(Generator, xx) need to be transformed into Generate(generator, ...)
override lazy val resolved =
childrenResolved && checkInputDataTypes().isSuccess &&
!child.isInstanceOf[Generator]
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
index 07c86a7..19e9312 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -27,7 +27,8 @@ import org.apache.spark.sql.catalyst.expressions.aggregate._
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules._
-import org.apache.spark.sql.catalyst.trees.TreePattern.PLAN_EXPRESSION
+import org.apache.spark.sql.catalyst.trees.AlwaysProcess
+import org.apache.spark.sql.catalyst.trees.TreePattern._
import org.apache.spark.sql.connector.catalog.CatalogManager
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
@@ -377,7 +378,8 @@ object EliminateDistinct extends Rule[LogicalPlan] {
* This rule should be applied before RewriteDistinctAggregates.
*/
object EliminateAggregateFilter extends Rule[LogicalPlan] {
- override def apply(plan: LogicalPlan): LogicalPlan = plan
transformExpressions {
+ override def apply(plan: LogicalPlan): LogicalPlan =
plan.transformExpressionsWithPruning(
+ _.containsAllPatterns(TRUE_OR_FALSE_LITERAL), ruleId) {
case ae @ AggregateExpression(_, _, _, Some(Literal.TrueLiteral), _) =>
ae.copy(filter = None)
case AggregateExpression(af: DeclarativeAggregate, _, _,
Some(Literal.FalseLiteral), _) =>
@@ -445,6 +447,9 @@ object RemoveRedundantAliases extends Rule[LogicalPlan] {
* (self) join or to prevent the removal of top-level subquery attributes.
*/
private def removeRedundantAliases(plan: LogicalPlan, excluded:
AttributeSet): LogicalPlan = {
+ if (!plan.containsPattern(ALIAS)) {
+ return plan
+ }
plan match {
// We want to keep the same output attributes for subqueries. This means
we cannot remove
// the aliases that produce these attributes
@@ -506,7 +511,8 @@ object RemoveRedundantAliases extends Rule[LogicalPlan] {
* only goal is to keep distinct values, while its parent aggregate would
ignore duplicate values.
*/
object RemoveRedundantAggregates extends Rule[LogicalPlan] with AliasHelper {
- def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(
+ _.containsPattern(AGGREGATE), ruleId) {
case upper @ Aggregate(_, _, lower: Aggregate) if lowerIsRedundant(upper,
lower) =>
val aliasMap = getAliasMap(lower)
@@ -545,7 +551,8 @@ object RemoveRedundantAggregates extends Rule[LogicalPlan]
with AliasHelper {
* Remove no-op operators from the query plan that do not make any
modifications.
*/
object RemoveNoopOperators extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(
+ _.containsAnyPattern(PROJECT, WINDOW), ruleId) {
// Eliminate no-op Projects
case p @ Project(_, child) if child.sameOutput(p) => child
@@ -597,7 +604,8 @@ object RemoveNoopUnion extends Rule[LogicalPlan] {
}
}
- def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(
+ _.containsAllPatterns(DISTINCT_LIKE, UNION)) {
case d @ Distinct(u: Union) =>
d.withNewChildren(Seq(simplifyUnion(u)))
case d @ Deduplicate(_, u: Union) =>
@@ -648,7 +656,8 @@ object LimitPushDown extends Rule[LogicalPlan] {
}
}
- def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
+ _.containsPattern(LIMIT), ruleId) {
// Adding extra Limits below UNION ALL for children which are not Limit or
do not have Limit
// descendants whose maxRow is larger. This heuristic is valid assuming
there does not exist any
// Limit push-down rule that is unable to infer the value of maxRows.
@@ -745,7 +754,8 @@ object PushProjectionThroughUnion extends Rule[LogicalPlan]
with PredicateHelper
*/
object ColumnPruning extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = removeProjectBeforeFilter(plan
transform {
+ def apply(plan: LogicalPlan): LogicalPlan = removeProjectBeforeFilter(
+ plan.transformWithPruning(AlwaysProcess.fn, ruleId) {
// Prunes the unused columns from project list of Project/Aggregate/Expand
case p @ Project(_, p2: Project) if !p2.outputSet.subsetOf(p.references) =>
p.copy(child = p2.copy(projectList =
p2.projectList.filter(p.references.contains)))
@@ -863,7 +873,8 @@ object ColumnPruning extends Rule[LogicalPlan] {
*/
object CollapseProject extends Rule[LogicalPlan] with AliasHelper {
- def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(
+ _.containsPattern(PROJECT), ruleId) {
case p1 @ Project(_, p2: Project) =>
if (haveCommonNonDeterministicOutput(p1.projectList, p2.projectList)) {
p1
@@ -921,7 +932,8 @@ object CollapseProject extends Rule[LogicalPlan] with
AliasHelper {
* Combines adjacent [[RepartitionOperation]] operators
*/
object CollapseRepartition extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(
+ _.containsPattern(REPARTITION_OPERATION), ruleId) {
// Case 1: When a Repartition has a child of Repartition or
RepartitionByExpression,
// 1) When the top node does not enable the shuffle (i.e., coalesce API),
but the child
// enables the shuffle. Returns the child node if the last numPartitions
is bigger;
@@ -943,7 +955,8 @@ object CollapseRepartition extends Rule[LogicalPlan] {
* and user not specify.
*/
object OptimizeRepartition extends Rule[LogicalPlan] {
- override def apply(plan: LogicalPlan): LogicalPlan = plan.transform {
+ override def apply(plan: LogicalPlan): LogicalPlan =
plan.transformWithPruning(
+ _.containsPattern(REPARTITION_OPERATION), ruleId) {
case r @ RepartitionByExpression(partitionExpressions, _, numPartitions)
if partitionExpressions.nonEmpty &&
partitionExpressions.forall(_.foldable) &&
numPartitions.isEmpty =>
@@ -955,7 +968,8 @@ object OptimizeRepartition extends Rule[LogicalPlan] {
* Replaces first(col) to nth_value(col, 1) for better performance.
*/
object OptimizeWindowFunctions extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan resolveExpressions {
+ def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveExpressionsWithPruning(
+ _.containsPattern(WINDOW_EXPRESSION), ruleId) {
case we @ WindowExpression(AggregateExpression(first: First, _, _, _, _),
WindowSpecDefinition(_, orderSpec, frameSpecification:
SpecifiedWindowFrame))
if orderSpec.nonEmpty && frameSpecification.frameType == RowFrame &&
@@ -972,7 +986,8 @@ object OptimizeWindowFunctions extends Rule[LogicalPlan] {
* independent and are of the same window function type, collapse into the
parent.
*/
object CollapseWindow extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(
+ _.containsPattern(WINDOW), ruleId) {
case w1 @ Window(we1, ps1, os1, w2 @ Window(we2, ps2, os2, grandChild))
if ps1 == ps2 && os1 == os2 &&
w1.references.intersect(w2.windowOutputSet).isEmpty &&
we1.nonEmpty && we2.nonEmpty &&
@@ -995,7 +1010,8 @@ object TransposeWindow extends Rule[LogicalPlan] {
})
}
- def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(
+ _.containsPattern(WINDOW), ruleId) {
case w1 @ Window(we1, ps1, os1, w2 @ Window(we2, ps2, os2, grandChild))
if w1.references.intersect(w2.windowOutputSet).isEmpty &&
w1.expressions.forall(_.deterministic) &&
@@ -1010,7 +1026,8 @@ object TransposeWindow extends Rule[LogicalPlan] {
* by this [[Generate]] can be removed earlier - before joins and in data
sources.
*/
object InferFiltersFromGenerate extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(
+ _.containsPattern(GENERATE)) {
// This rule does not infer filters from foldable expressions to avoid
constant filters
// like 'size([1, 2, 3]) > 0'. These do not show up in child's constraints
and
// then the idempotence will break.
@@ -1060,7 +1077,8 @@ object InferFiltersFromConstraints extends
Rule[LogicalPlan]
}
}
- private def inferFilters(plan: LogicalPlan): LogicalPlan = plan transform {
+ private def inferFilters(plan: LogicalPlan): LogicalPlan =
plan.transformWithPruning(
+ _.containsAnyPattern(FILTER, JOIN)) {
case filter @ Filter(condition, child) =>
val newFilters = filter.constraints --
(child.constraints ++ splitConjunctivePredicates(condition))
@@ -1123,7 +1141,8 @@ object InferFiltersFromConstraints extends
Rule[LogicalPlan]
* Combines all adjacent [[Union]] operators into a single [[Union]].
*/
object CombineUnions extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transformDown {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformDownWithPruning(
+ _.containsAnyPattern(UNION, DISTINCT_LIKE), ruleId) {
case u: Union => flattenUnion(u, false)
case Distinct(u: Union) => Distinct(flattenUnion(u, true))
// Only handle distinct-like 'Deduplicate', where the keys == output
@@ -1167,7 +1186,8 @@ object CombineUnions extends Rule[LogicalPlan] {
* one conjunctive predicate.
*/
object CombineFilters extends Rule[LogicalPlan] with PredicateHelper {
- def apply(plan: LogicalPlan): LogicalPlan = plan transform applyLocally
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
+ _.containsPattern(FILTER), ruleId)(applyLocally)
val applyLocally: PartialFunction[LogicalPlan, LogicalPlan] = {
// The query execution/optimization does not guarantee the expressions are
evaluated in order.
@@ -1202,7 +1222,8 @@ object CombineFilters extends Rule[LogicalPlan] with
PredicateHelper {
* function is order irrelevant
*/
object EliminateSorts extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transform applyLocally
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
+ _.containsPattern(SORT))(applyLocally)
private val applyLocally: PartialFunction[LogicalPlan, LogicalPlan] = {
case s @ Sort(orders, _, child) if orders.isEmpty ||
orders.exists(_.child.foldable) =>
@@ -1221,11 +1242,16 @@ object EliminateSorts extends Rule[LogicalPlan] {
g.copy(child = recursiveRemoveSort(originChild))
}
- private def recursiveRemoveSort(plan: LogicalPlan): LogicalPlan = plan match
{
- case Sort(_, _, child) => recursiveRemoveSort(child)
- case other if canEliminateSort(other) =>
- other.withNewChildren(other.children.map(recursiveRemoveSort))
- case _ => plan
+ private def recursiveRemoveSort(plan: LogicalPlan): LogicalPlan = {
+ if (!plan.containsPattern(SORT)) {
+ return plan
+ }
+ plan match {
+ case Sort(_, _, child) => recursiveRemoveSort(child)
+ case other if canEliminateSort(other) =>
+ other.withNewChildren(other.children.map(recursiveRemoveSort))
+ case _ => plan
+ }
}
private def canEliminateSort(plan: LogicalPlan): Boolean = plan match {
@@ -1264,7 +1290,8 @@ object EliminateSorts extends Rule[LogicalPlan] {
* 3) by eliminating the always-true conditions given the constraints on the
child's output.
*/
object PruneFilters extends Rule[LogicalPlan] with PredicateHelper {
- def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
+ _.containsPattern(FILTER), ruleId) {
// If the filter condition always evaluate to true, remove the filter.
case Filter(Literal(true, BooleanType), child) => child
// If the filter condition always evaluate to null or false,
@@ -1620,7 +1647,8 @@ object EliminateLimits extends Rule[LogicalPlan] {
limitExpr.foldable && child.maxRows.exists { _ <=
limitExpr.eval().asInstanceOf[Int] }
}
- def apply(plan: LogicalPlan): LogicalPlan = plan transformDown {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformDownWithPruning(
+ _.containsPattern(LIMIT), ruleId) {
case Limit(l, child) if canEliminate(l, child) =>
child
case GlobalLimit(l, child) if canEliminate(l, child) =>
@@ -1667,7 +1695,7 @@ object CheckCartesianProducts extends Rule[LogicalPlan]
with PredicateHelper {
def apply(plan: LogicalPlan): LogicalPlan =
if (conf.crossJoinEnabled) {
plan
- } else plan transform {
+ } else plan.transformWithPruning(_.containsAnyPattern(INNER_LIKE_JOIN,
OUTER_JOIN)) {
case j @ Join(left, right, Inner | LeftOuter | RightOuter | FullOuter,
_, _)
if isCartesianProduct(j) =>
throw new AnalysisException(
@@ -1695,8 +1723,10 @@ object DecimalAggregates extends Rule[LogicalPlan] {
/** Maximum number of decimal digits representable precisely in a Double */
private val MAX_DOUBLE_DIGITS = 15
- def apply(plan: LogicalPlan): LogicalPlan = plan transform {
- case q: LogicalPlan => q transformExpressionsDown {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
+ _.containsAnyPattern(SUM, AVERAGE), ruleId) {
+ case q: LogicalPlan => q.transformExpressionsDownWithPruning(
+ _.containsAnyPattern(SUM, AVERAGE), ruleId) {
case we @ WindowExpression(ae @ AggregateExpression(af, _, _, _, _), _)
=> af match {
case Sum(e @ DecimalType.Expression(prec, scale)) if prec + 10 <=
MAX_LONG_DIGITS =>
MakeDecimal(we.copy(windowFunction = ae.copy(aggregateFunction =
Sum(UnscaledValue(e)))),
@@ -1732,7 +1762,8 @@ object DecimalAggregates extends Rule[LogicalPlan] {
* another `LocalRelation`.
*/
object ConvertToLocalRelation extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
+ _.containsPattern(LOCAL_RELATION), ruleId) {
case Project(projectList, LocalRelation(output, data, isStreaming))
if !projectList.exists(hasUnevaluableExpr) =>
val projection = new InterpretedMutableProjection(projectList, output)
@@ -1761,7 +1792,8 @@ object ConvertToLocalRelation extends Rule[LogicalPlan] {
* }}}
*/
object ReplaceDistinctWithAggregate extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
+ _.containsPattern(DISTINCT_LIKE), ruleId) {
case Distinct(child) => Aggregate(child.output, child.output, child)
}
}
@@ -1805,7 +1837,8 @@ object ReplaceDeduplicateWithAggregate extends
Rule[LogicalPlan] {
* join conditions will be incorrect.
*/
object ReplaceIntersectWithSemiJoin extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
+ _.containsPattern(INTERSECT), ruleId) {
case Intersect(left, right, false) =>
assert(left.output.size == right.output.size)
val joinCond = left.output.zip(right.output).map { case (l, r) =>
EqualNullSafe(l, r) }
@@ -1826,7 +1859,8 @@ object ReplaceIntersectWithSemiJoin extends
Rule[LogicalPlan] {
* join conditions will be incorrect.
*/
object ReplaceExceptWithAntiJoin extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
+ _.containsPattern(EXCEPT), ruleId) {
case Except(left, right, false) =>
assert(left.output.size == right.output.size)
val joinCond = left.output.zip(right.output).map { case (l, r) =>
EqualNullSafe(l, r) }
@@ -1866,7 +1900,8 @@ object ReplaceExceptWithAntiJoin extends
Rule[LogicalPlan] {
*/
object RewriteExceptAll extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
+ _.containsPattern(EXCEPT), ruleId) {
case Except(left, right, true) =>
assert(left.output.size == right.output.size)
@@ -1923,7 +1958,8 @@ object RewriteExceptAll extends Rule[LogicalPlan] {
* }}}
*/
object RewriteIntersectAll extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
+ _.containsPattern(INTERSECT), ruleId) {
case Intersect(left, right, true) =>
assert(left.output.size == right.output.size)
@@ -1975,7 +2011,8 @@ object RewriteIntersectAll extends Rule[LogicalPlan] {
* but only makes the grouping key bigger.
*/
object RemoveLiteralFromGroupExpressions extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
+ _.containsPattern(AGGREGATE), ruleId) {
case a @ Aggregate(grouping, _, _) if grouping.nonEmpty =>
val newGrouping = grouping.filter(!_.foldable)
if (newGrouping.nonEmpty) {
@@ -1994,7 +2031,8 @@ object RemoveLiteralFromGroupExpressions extends
Rule[LogicalPlan] {
* but only makes the grouping key bigger.
*/
object RemoveRepetitionFromGroupExpressions extends Rule[LogicalPlan] {
- def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
+ _.containsPattern(AGGREGATE), ruleId) {
case a @ Aggregate(grouping, _, _) if grouping.size > 1 =>
val newGrouping = ExpressionSet(grouping).toSeq
if (newGrouping.size == grouping.size) {
@@ -2014,7 +2052,8 @@ object OptimizeLimitZero extends Rule[LogicalPlan] {
private def empty(plan: LogicalPlan) =
LocalRelation(plan.output, data = Seq.empty, isStreaming =
plan.isStreaming)
- def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {
+ def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(
+ _.containsAllPatterns(LIMIT, LITERAL)) {
// Nodes below GlobalLimit or LocalLimit can be pruned if the limit value
is zero (0).
// Any subtree in the logical plan that has GlobalLimit 0 or LocalLimit 0
as its root is
// semantically equivalent to an empty relation.
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
index d3c5b51..88a58fd 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
@@ -127,6 +127,8 @@ case class Generate(
child: LogicalPlan)
extends UnaryNode {
+ final override val nodePatterns: Seq[TreePattern] = Seq(GENERATE)
+
lazy val requiredChildOutput: Seq[Attribute] = {
val unrequiredSet = unrequiredChildIndex.toSet
child.output.zipWithIndex.filterNot(t =>
unrequiredSet.contains(t._2)).map(_._1)
@@ -211,6 +213,8 @@ case class Intersect(
override def nodeName: String = getClass.getSimpleName + ( if ( isAll )
"All" else "" )
+ final override val nodePatterns: Seq[TreePattern] = Seq(INTERSECT)
+
override def output: Seq[Attribute] =
left.output.zip(right.output).map { case (leftAttr, rightAttr) =>
leftAttr.withNullability(leftAttr.nullable && rightAttr.nullable)
@@ -280,6 +284,8 @@ case class Union(
}
}
+ final override val nodePatterns: Seq[TreePattern] = Seq(UNION)
+
/**
* Note the definition has assumption about how union is implemented
physically.
*/
@@ -632,6 +638,7 @@ case class Sort(
override def output: Seq[Attribute] = child.output
override def maxRows: Option[Long] = child.maxRows
override def outputOrdering: Seq[SortOrder] = order
+ final override val nodePatterns: Seq[TreePattern] = Seq(SORT)
override protected def withNewChildInternal(newChild: LogicalPlan): Sort =
copy(child = newChild)
}
@@ -1203,6 +1210,7 @@ case class Sample(
case class Distinct(child: LogicalPlan) extends UnaryNode {
override def maxRows: Option[Long] = child.maxRows
override def output: Seq[Attribute] = child.output
+ final override val nodePatterns: Seq[TreePattern] = Seq(DISTINCT_LIKE)
override protected def withNewChildInternal(newChild: LogicalPlan): Distinct
=
copy(child = newChild)
}
@@ -1215,6 +1223,7 @@ abstract class RepartitionOperation extends UnaryNode {
def numPartitions: Int
override final def maxRows: Option[Long] = child.maxRows
override def output: Seq[Attribute] = child.output
+ final override val nodePatterns: Seq[TreePattern] =
Seq(REPARTITION_OPERATION)
def partitioning: Partitioning
}
@@ -1314,6 +1323,7 @@ case class Deduplicate(
child: LogicalPlan) extends UnaryNode {
override def maxRows: Option[Long] = child.maxRows
override def output: Seq[Attribute] = child.output
+ final override val nodePatterns: Seq[TreePattern] = Seq(DISTINCT_LIKE)
override protected def withNewChildInternal(newChild: LogicalPlan):
Deduplicate =
copy(child = newChild)
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
index 62f09d0..605b57e 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
@@ -88,38 +88,62 @@ object RuleIdCollection {
"org.apache.spark.sql.catalyst.analysis.UpdateOuterReferences" ::
// Catalyst Optimizer rules
"org.apache.spark.sql.catalyst.optimizer.BooleanSimplification" ::
+ "org.apache.spark.sql.catalyst.optimizer.CollapseProject" ::
+ "org.apache.spark.sql.catalyst.optimizer.CollapseRepartition" ::
+ "org.apache.spark.sql.catalyst.optimizer.CollapseWindow" ::
+ "org.apache.spark.sql.catalyst.optimizer.ColumnPruning" ::
"org.apache.spark.sql.catalyst.optimizer.CombineConcats" ::
+ "org.apache.spark.sql.catalyst.optimizer.CombineFilters" ::
"org.apache.spark.sql.catalyst.optimizer.CombineTypedFilters" ::
+ "org.apache.spark.sql.catalyst.optimizer.CombineUnions" ::
"org.apache.spark.sql.catalyst.optimizer.ConstantFolding" ::
"org.apache.spark.sql.catalyst.optimizer.ConstantPropagation" ::
+ "org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation" ::
"org.apache.spark.sql.catalyst.optimizer.CostBasedJoinReorder" ::
+ "org.apache.spark.sql.catalyst.optimizer.DecimalAggregates" ::
+ "org.apache.spark.sql.catalyst.optimizer.EliminateAggregateFilter" ::
+ "org.apache.spark.sql.catalyst.optimizer.EliminateLimits" ::
"org.apache.spark.sql.catalyst.optimizer.EliminateMapObjects" ::
"org.apache.spark.sql.catalyst.optimizer.EliminateOuterJoin" ::
"org.apache.spark.sql.catalyst.optimizer.EliminateSerialization" ::
"org.apache.spark.sql.catalyst.optimizer.LikeSimplification" ::
+ "org.apache.spark.sql.catalyst.optimizer.LimitPushDown" ::
"org.apache.spark.sql.catalyst.optimizer.LimitPushDownThroughWindow" ::
"org.apache.spark.sql.catalyst.optimizer.NullPropagation" ::
"org.apache.spark.sql.catalyst.optimizer.ObjectSerializerPruning" ::
"org.apache.spark.sql.catalyst.optimizer.OptimizeCsvJsonExprs" ::
"org.apache.spark.sql.catalyst.optimizer.OptimizeIn" ::
"org.apache.spark.sql.catalyst.optimizer.Optimizer$OptimizeSubqueries" ::
+ "org.apache.spark.sql.catalyst.optimizer.OptimizeRepartition" ::
+ "org.apache.spark.sql.catalyst.optimizer.OptimizeWindowFunctions" ::
"org.apache.spark.sql.catalyst.optimizer.OptimizeUpdateFields"::
"org.apache.spark.sql.catalyst.optimizer.PropagateEmptyRelation" ::
+ "org.apache.spark.sql.catalyst.optimizer.PruneFilters" ::
"org.apache.spark.sql.catalyst.optimizer.PushDownLeftSemiAntiJoin" ::
"org.apache.spark.sql.catalyst.optimizer.PushExtraPredicateThroughJoin"
::
"org.apache.spark.sql.catalyst.optimizer.PushFoldableIntoBranches" ::
"org.apache.spark.sql.catalyst.optimizer.PushLeftSemiLeftAntiThroughJoin" ::
"org.apache.spark.sql.catalyst.optimizer.ReassignLambdaVariableID" ::
"org.apache.spark.sql.catalyst.optimizer.RemoveDispensableExpressions" ::
+
"org.apache.spark.sql.catalyst.optimizer.RemoveLiteralFromGroupExpressions" ::
+ "org.apache.spark.sql.catalyst.optimizer.RemoveNoopOperators" ::
+ "org.apache.spark.sql.catalyst.optimizer.RemoveRedundantAggregates" ::
+
"org.apache.spark.sql.catalyst.optimizer.RemoveRepetitionFromGroupExpressions"
::
"org.apache.spark.sql.catalyst.optimizer.ReorderAssociativeOperator" ::
"org.apache.spark.sql.catalyst.optimizer.ReorderJoin" ::
+ "org.apache.spark.sql.catalyst.optimizer.ReplaceExceptWithAntiJoin" ::
"org.apache.spark.sql.catalyst.optimizer.ReplaceExceptWithFilter" ::
+ "org.apache.spark.sql.catalyst.optimizer.ReplaceDistinctWithAggregate" ::
"org.apache.spark.sql.catalyst.optimizer.ReplaceNullWithFalseInPredicate" ::
+ "org.apache.spark.sql.catalyst.optimizer.ReplaceIntersectWithSemiJoin" ::
+ "org.apache.spark.sql.catalyst.optimizer.RewriteExceptAll" ::
+ "org.apache.spark.sql.catalyst.optimizer.RewriteIntersectAll" ::
"org.apache.spark.sql.catalyst.optimizer.SimplifyBinaryComparison" ::
"org.apache.spark.sql.catalyst.optimizer.SimplifyCaseConversionExpressions" ::
"org.apache.spark.sql.catalyst.optimizer.SimplifyCasts" ::
"org.apache.spark.sql.catalyst.optimizer.SimplifyConditionals" ::
"org.apache.spark.sql.catalyst.optimizer.SimplifyConditionalsInPredicate" ::
+ "org.apache.spark.sql.catalyst.optimizer.TransposeWindow" ::
"org.apache.spark.sql.catalyst.optimizer.UnwrapCastInBinaryComparison"
:: Nil
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
index d1ba832..40ef7cb 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
@@ -23,9 +23,11 @@ object TreePattern extends Enumeration {
// Enum Ids start from 0.
// Expression patterns (alphabetically ordered)
- val AND_OR: Value = Value(0)
+ val ALIAS: Value = Value(0)
+ val AND_OR: Value = Value
val ATTRIBUTE_REFERENCE: Value = Value
val APPEND_COLUMNS: Value = Value
+ val AVERAGE: Value = Value
val BINARY_ARITHMETIC: Value = Value
val BINARY_COMPARISON: Value = Value
val BOOL_AGG: Value = Value
@@ -41,10 +43,12 @@ object TreePattern extends Enumeration {
val EXISTS_SUBQUERY = Value
val EXPRESSION_WITH_RANDOM_SEED: Value = Value
val EXTRACT_VALUE: Value = Value
+ val GENERATE: Value = Value
val IF: Value = Value
val IN: Value = Value
val IN_SUBQUERY: Value = Value
val INSET: Value = Value
+ val INTERSECT: Value = Value
val JSON_TO_STRUCT: Value = Value
val LAMBDA_VARIABLE: Value = Value
val LIKE_FAMLIY: Value = Value
@@ -59,6 +63,8 @@ object TreePattern extends Enumeration {
val PLAN_EXPRESSION: Value = Value
val RUNTIME_REPLACEABLE: Value = Value
val SCALAR_SUBQUERY: Value = Value
+ val SORT: Value = Value
+ val SUM: Value = Value
val TRUE_OR_FALSE_LITERAL: Value = Value
val WINDOW_EXPRESSION: Value = Value
val UNARY_POSITIVE: Value = Value
@@ -66,6 +72,7 @@ object TreePattern extends Enumeration {
// Logical plan patterns (alphabetically ordered)
val AGGREGATE: Value = Value
+ val DISTINCT_LIKE: Value = Value
val EXCEPT: Value = Value
val FILTER: Value = Value
val INNER_LIKE_JOIN: Value = Value
@@ -76,6 +83,8 @@ object TreePattern extends Enumeration {
val NATURAL_LIKE_JOIN: Value = Value
val OUTER_JOIN: Value = Value
val PROJECT: Value = Value
+ val REPARTITION_OPERATION: Value = Value
+ val UNION: Value = Value
val TYPED_FILTER: Value = Value
val WINDOW: Value = Value
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
