apache
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala‎
Lines changed: 6 additions & 3 deletions b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala‎
Lines changed: 6 additions & 3 deletions
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala‎
Lines changed: 2 additions & 2 deletions b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala‎
Lines changed: 2 additions & 2 deletions
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/AliasHelper.scala‎
Lines changed: 2 additions & 2 deletions b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/AliasHelper.scala‎
Lines changed: 2 additions & 2 deletions
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Average.scala‎
Lines changed: 1 addition & 1 deletion b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Average.scala‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Sum.scala‎
Lines changed: 13 additions & 7 deletions b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Sum.scala‎
Lines changed: 13 additions & 7 deletions
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala‎
Lines changed: 1 addition & 1 deletion b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/predicates.scala‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DeduplicateRightSideOfLeftSemiAntiJoin.scala‎
Lines changed: 43 additions & 0 deletions b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DeduplicateRightSideOfLeftSemiAntiJoin.scala‎
Lines changed: 43 additions & 0 deletions
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala‎
Lines changed: 17 additions & 13 deletions b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala‎
Lines changed: 17 additions & 13 deletions
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PredicateReorder.scala‎
Lines changed: 5 additions & 1 deletion b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PredicateReorder.scala‎
Lines changed: 5 additions & 1 deletion
diff --git a/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala‎
Lines changed: 1 addition & 1 deletion b/‎sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala‎
Lines changed: 1 addition & 1 deletion
@@ -3729,9 +3729,12 @@ object CleanupAliases extends Rule[LogicalPlan] with AliasHelper {
       val cleanedProjectList = projectList.map(trimNonTopLevelAliases)
       Project(cleanedProjectList, child)
 
-    case Aggregate(grouping, aggs, child) =>
-      val cleanedAggs = aggs.map(trimNonTopLevelAliases)
-      Aggregate(grouping.map(trimAliases), cleanedAggs, child)
+    case a: AggregateBase =>
+      val cleanedGroupings = a.groupingExpressions.map(trimAliases)
+      val cleanedAggs = a.aggregateExpressions.map(trimNonTopLevelAliases)
+      a.withGroupingExpressions(cleanedGroupings)
+        .withAggregateExpressions(cleanedAggs)
+        .withNewChildren(Seq(a.child))
 
     case Window(windowExprs, partitionSpec, orderSpec, limit, child) =>
       val cleanedWindowExprs = windowExprs.map(trimNonTopLevelAliases)
 
@@ -433,7 +433,7 @@ object TypeCoercion {
         p.makeCopy(Array(castExpr(left, commonType), castExpr(right, commonType)))
 
       case Abs(e @ StringType()) => Abs(Cast(e, DoubleType))
-      case Sum(e @ StringType()) => Sum(Cast(e, DoubleType))
+      case Sum(e @ StringType(), None) => Sum(Cast(e, DoubleType))
       case Average(e @ StringType()) => Average(Cast(e, DoubleType))
       case StddevPop(e @ StringType()) => StddevPop(Cast(e, DoubleType))
       case StddevSamp(e @ StringType()) => StddevSamp(Cast(e, DoubleType))
@@ -628,7 +628,7 @@ object TypeCoercion {
         m.copy(newKeys.zip(newValues).flatMap { case (k, v) => Seq(k, v) })
 
       // Hive lets you do aggregation of timestamps... for some reason
-      case Sum(e @ TimestampType()) => Sum(Cast(e, DoubleType))
+      case Sum(e @ TimestampType(), None) => Sum(Cast(e, DoubleType))
       case Average(e @ TimestampType()) => Average(Cast(e, DoubleType))
 
       // Coalesce should return the first non-null value, which could be any column
 
@@ -19,7 +19,7 @@ package org.apache.spark.sql.catalyst.expressions
 
 import org.apache.spark.sql.catalyst.analysis.MultiAlias
 import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
-import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Project}
+import org.apache.spark.sql.catalyst.plans.logical.{AggregateBase, Project}
 
 /**
  * Helper methods for collecting and replacing aliases.
@@ -32,7 +32,7 @@ trait AliasHelper {
     getAliasMap(plan.projectList)
   }
 
-  protected def getAliasMap(plan: Aggregate): AttributeMap[Alias] = {
+  protected def getAliasMap(plan: AggregateBase): AttributeMap[Alias] = {
     // Find all the aliased expressions in the aggregate list that don't include any actual
     // AggregateExpression or PythonUDF, and create a map from the alias to the expression
     val aliasMap = plan.aggregateExpressions.collect {
 
@@ -56,7 +56,7 @@ case class Average(child: Expression) extends DeclarativeAggregate with Implicit
     case _ => DoubleType
   }
 
-  private lazy val sumDataType = child.dataType match {
+  private[sql] lazy val sumDataType = child.dataType match {
     case _ @ DecimalType.Fixed(p, s) => DecimalType.bounded(p + 10, s)
     case _ => DoubleType
   }
 
@@ -37,7 +37,12 @@ import org.apache.spark.sql.types._
   """,
   group = "agg_funcs",
   since = "1.0.0")
-case class Sum(child: Expression) extends DeclarativeAggregate with ImplicitCastInputTypes {
+case class Sum(child: Expression, resultDataType: Option[DataType] = None)
+  extends DeclarativeAggregate with ImplicitCastInputTypes {
+
+  def this(child: Expression) = {
+    this(child, None)
+  }
 
   override def children: Seq[Expression] = child :: Nil
 
@@ -51,12 +56,13 @@ case class Sum(child: Expression) extends DeclarativeAggregate with ImplicitCast
   override def checkInputDataTypes(): TypeCheckResult =
     TypeUtils.checkForNumericExpr(child.dataType, "function sum")
 
-  private lazy val resultType = child.dataType match {
-    case DecimalType.Fixed(precision, scale) =>
-      DecimalType.bounded(precision + 10, scale)
-    case _: IntegralType => LongType
-    case _ => DoubleType
-  }
+  private lazy val resultType = resultDataType.getOrElse(
+    child.dataType match {
+      case DecimalType.Fixed(precision, scale) => DecimalType.bounded(precision + 10, scale)
+      case _: IntegralType => LongType
+      case _ => DoubleType
+    }
+  )
 
   private lazy val sum = AttributeReference("sum", resultType)()
 
 
@@ -128,7 +128,7 @@ trait PredicateHelper extends AliasHelper with Logging {
         val aliases = getAliasMap(p)
         findExpressionAndTrackLineageDown(replaceAlias(exp, aliases), p.child)
       // we can unwrap only if there are row projections, and no aggregation operation
-      case a: Aggregate =>
+      case a: AggregateBase =>
         val aliasMap = getAliasMap(a)
         findExpressionAndTrackLineageDown(replaceAlias(exp, aliasMap), a.child)
       case l: LeafNode if exp.references.subsetOf(l.outputSet) =>
 
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.plans.LeftSemiOrAnti
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+
+/**
+ * Deduplicate the right side of left semi/anti join.
+ */
+object DeduplicateRightSideOfLeftSemiAntiJoin extends Rule[LogicalPlan] with JoinSelectionHelper {
+
+  def apply(plan: LogicalPlan): LogicalPlan = {
+    if (conf.partialAggregationOptimizationEnabled) {
+      plan.transform {
+        case j @ Join(_, _: AggregateBase, LeftSemiOrAnti(_), _, _) =>
+          j
+        case j @ Join(_, right, LeftSemiOrAnti(_), _, _)
+            if PushPartialAggregationThroughJoin.pushPartialAggHasBenefit(right.output, right) =>
+          j.copy(right = PartialAggregate(right.output, right.output, right))
+      }
+    } else {
+      plan
+    }
+  }
+}
@@ -207,6 +207,9 @@ abstract class Optimizer(catalogManager: CatalogManager)
     // aggregate distinct column
     Batch("Distinct Aggregate Rewrite", Once,
       RewriteDistinctAggregates) :+
+    Batch("Partial Aggregation Optimization", fixedPoint,
+      PushPartialAggregationThroughJoin,
+      DeduplicateRightSideOfLeftSemiAntiJoin) :+
     Batch("Object Expressions Optimization", fixedPoint,
       EliminateMapObjects,
       CombineTypedFilters,
@@ -224,6 +227,7 @@ abstract class Optimizer(catalogManager: CatalogManager)
       LimitPushDown,
       ColumnPruning,
       CollapseProject,
+      DeduplicateRightSideOfLeftSemiAntiJoin,
       RemoveNoopOperators) :+
     // This batch must be executed after the `RewriteSubquery` batch, which creates joins.
     Batch("NormalizeFloatingNumbers", Once, NormalizeFloatingNumbers)
@@ -632,9 +636,9 @@ object ColumnPruning extends Rule[LogicalPlan] {
     // 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)))
-    case p @ Project(_, a: Aggregate) if !a.outputSet.subsetOf(p.references) =>
-      p.copy(
-        child = a.copy(aggregateExpressions = a.aggregateExpressions.filter(p.references.contains)))
+    case p @ Project(_, a: AggregateBase) if !a.outputSet.subsetOf(p.references) =>
+      val newAggregateExpressions = a.aggregateExpressions.filter(p.references.contains)
+      p.copy(child = a.withAggregateExpressions(newAggregateExpressions))
     case a @ Project(_, e @ Expand(_, _, grandChild)) if !e.outputSet.subsetOf(a.references) =>
       val newOutput = e.output.filter(a.references.contains(_))
       val newProjects = e.projections.map { proj =>
@@ -649,8 +653,8 @@ object ColumnPruning extends Rule[LogicalPlan] {
       d.copy(child = prunedChild(child, d.references))
 
     // Prunes the unused columns from child of Aggregate/Expand/Generate/ScriptTransformation
-    case a @ Aggregate(_, _, child) if !child.outputSet.subsetOf(a.references) =>
-      a.copy(child = prunedChild(child, a.references))
+    case a: AggregateBase if !a.child.outputSet.subsetOf(a.references) =>
+      a.withNewChildren(Seq(prunedChild(a.child, a.references)))
     case f @ FlatMapGroupsInPandas(_, _, _, child) if !child.outputSet.subsetOf(f.references) =>
       f.copy(child = prunedChild(child, f.references))
     case e @ Expand(_, _, child) if !child.outputSet.subsetOf(e.references) =>
@@ -738,7 +742,7 @@ object ColumnPruning extends Rule[LogicalPlan] {
   })
 
   /** Applies a projection only when the child is producing unnecessary attributes */
-  private def prunedChild(c: LogicalPlan, allReferences: AttributeSet) =
+  def prunedChild(c: LogicalPlan, allReferences: AttributeSet): LogicalPlan =
     if (!c.outputSet.subsetOf(allReferences)) {
       Project(c.output.filter(allReferences.contains), c)
     } else {
@@ -776,12 +780,12 @@ object CollapseProject extends Rule[LogicalPlan] with AliasHelper {
       } else {
         p2.copy(projectList = buildCleanedProjectList(p1.projectList, p2.projectList))
       }
-    case p @ Project(_, agg: Aggregate) =>
+    case p @ Project(_, agg: AggregateBase) =>
       if (haveCommonNonDeterministicOutput(p.projectList, agg.aggregateExpressions)) {
         p
       } else {
-        agg.copy(aggregateExpressions = buildCleanedProjectList(
-          p.projectList, agg.aggregateExpressions))
+        val newAggregateExprs = buildCleanedProjectList(p.projectList, agg.aggregateExpressions)
+        agg.withAggregateExpressions(newAggregateExprs)
       }
     case Project(l1, g @ GlobalLimit(_, limit @ LocalLimit(_, p2 @ Project(l2, _))))
         if isRenaming(l1, l2) =>
@@ -1377,7 +1381,7 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
       val aliasMap = getAliasMap(project)
       project.copy(child = Filter(replaceAlias(condition, aliasMap), grandChild))
 
-    case filter @ Filter(condition, aggregate: Aggregate)
+    case filter @ Filter(condition, aggregate: AggregateBase)
       if aggregate.aggregateExpressions.forall(_.deterministic)
         && aggregate.groupingExpressions.nonEmpty =>
       val aliasMap = getAliasMap(aggregate)
@@ -1397,7 +1401,7 @@ object PushPredicateThroughNonJoin extends Rule[LogicalPlan] with PredicateHelpe
       if (pushDown.nonEmpty) {
         val pushDownPredicate = pushDown.reduce(And)
         val replaced = replaceAlias(pushDownPredicate, aliasMap)
-        val newAggregate = aggregate.copy(child = Filter(replaced, aggregate.child))
+        val newAggregate = aggregate.withNewChildren(Seq(Filter(replaced, aggregate.child)))
         // If there is no more filter to stay up, just eliminate the filter.
         // Otherwise, create "Filter(stayUp) <- Aggregate <- Filter(pushDownPredicate)".
         if (stayUp.isEmpty) newAggregate else Filter(stayUp.reduce(And), newAggregate)
@@ -1777,7 +1781,7 @@ object DecimalAggregates extends Rule[LogicalPlan] {
   def apply(plan: LogicalPlan): LogicalPlan = plan transform {
     case q: LogicalPlan => q transformExpressionsDown {
       case we @ WindowExpression(ae @ AggregateExpression(af, _, _, _, _), _) => af match {
-        case Sum(e @ DecimalType.Expression(prec, scale)) if prec + 10 <= MAX_LONG_DIGITS =>
+        case Sum(e @ DecimalType.Expression(prec, scale), None) if prec + 10 <= MAX_LONG_DIGITS =>
           MakeDecimal(we.copy(windowFunction = ae.copy(aggregateFunction = Sum(UnscaledValue(e)))),
             prec + 10, scale)
 
@@ -1791,7 +1795,7 @@ object DecimalAggregates extends Rule[LogicalPlan] {
         case _ => we
       }
       case ae @ AggregateExpression(af, _, _, _, _) => af match {
-        case Sum(e @ DecimalType.Expression(prec, scale)) if prec + 10 <= MAX_LONG_DIGITS =>
+        case Sum(e @ DecimalType.Expression(prec, scale), None) if prec + 10 <= MAX_LONG_DIGITS =>
           MakeDecimal(ae.copy(aggregateFunction = Sum(UnscaledValue(e))), prec + 10, scale)
 
         case Average(e @ DecimalType.Expression(prec, scale)) if prec + 4 <= MAX_DOUBLE_DIGITS =>
 
@@ -18,6 +18,7 @@
 package org.apache.spark.sql.catalyst.optimizer
 
 import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.DeclarativeAggregate
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.plans.logical.statsEstimation.FilterEstimation
 import org.apache.spark.sql.catalyst.rules.Rule
@@ -43,6 +44,7 @@ object PredicateReorder extends Rule[LogicalPlan] with PredicateHelper {
   private val QUATERNARY_OP_COST = 4.0D
   private val SEPTENARY_OP_COST = 7.0D
   private val COMPLEX_TYPE_MERGING_OP_COST = 10.0D
+  private val AGGREGATE_OP_COST = 15.0D
   private val USER_DEFINED_OP_COST = 20.0D
   private val UNKNOWN_COST = 30.0D
 
@@ -65,7 +67,7 @@ object PredicateReorder extends Rule[LogicalPlan] with PredicateHelper {
 
   // The cost of a call expression exp is computed as:
   //   cost(exp) = typeSize + functionCost + cost(children).
-  private def expressionCost(exp: Expression): Double = exp match {
+  def expressionCost(exp: Expression): Double = exp match {
     case e: Expression if e.children.isEmpty =>
       e.dataType.defaultSize
     case e: IsNull =>
@@ -104,6 +106,8 @@ object PredicateReorder extends Rule[LogicalPlan] with PredicateHelper {
       e.dataType.defaultSize + SEPTENARY_OP_COST + e.children.map(expressionCost).sum
     case e: ComplexTypeMergingExpression =>
       e.dataType.defaultSize + COMPLEX_TYPE_MERGING_OP_COST + e.children.map(expressionCost).sum
+    case e: DeclarativeAggregate =>
+      e.dataType.defaultSize + AGGREGATE_OP_COST + e.children.map(expressionCost).sum
     case e: UserDefinedExpression =>
       e.dataType.defaultSize + USER_DEFINED_OP_COST + e.children.map(expressionCost).sum
     case e =>
 
@@ -100,7 +100,7 @@ object PropagateEmptyRelation extends Rule[LogicalPlan] with PredicateHelper wit
       // Aggregation on empty LocalRelation generated from a streaming source is not eliminated
       // as stateful streaming aggregation need to perform other state management operations other
       // than just processing the input data.
-      case Aggregate(ge, _, _) if ge.nonEmpty && !p.isStreaming => empty(p)
+      case a: AggregateBase if a.groupingExpressions.nonEmpty && !p.isStreaming => empty(p)
       // Generators like Hive-style UDTF may return their records within `close`.
       case Generate(_: Explode, _, _, _, _, _) => empty(p)
       case Expand(_, _, _) => empty(p)
Original file line number	Diff line number	Diff line change
`@@ -56,7 +56,7 @@ case class Average(child: Expression) extends DeclarativeAggregate with Implicit`
`56`	`56`	`case _ => DoubleType`
`57`	`57`	`}`
`58`	`58`
`59`		`- private lazy val sumDataType = child.dataType match {`
	`59`	`+ private[sql] lazy val sumDataType = child.dataType match {`
`60`	`60`	`case _ @ DecimalType.Fixed(p, s) => DecimalType.bounded(p + 10, s)`
`61`	`61`	`case _ => DoubleType`
`62`	`62`	`}`