Fix

maropu · maropu · commit 753e9e1df144 · 2020-01-15T16:00:34.000+09:00
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/interfaces.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/interfaces.scala
@@ -150,21 +150,17 @@ case class AggregateExpression(
       case Final | Complete => ""
     }
     val aggFuncStr = prefix + aggregateFunction.toAggString(isDistinct)
-    mode match {
-      case Partial | Complete if filter.isDefined =>
-        s"$aggFuncStr filter ${filter.get.toString}"
-      case _ =>
-        aggFuncStr
+    filter match {
+      case Some(predicate) => s"$aggFuncStr filter $predicate"
+      case _ => aggFuncStr
     }
   }
 
   override def sql: String = {
     val aggFuncStr = aggregateFunction.sql(isDistinct)
-    mode match {
-      case Partial | Complete if filter.isDefined =>
-        s"$aggFuncStr FILTER ${filter.get.sql}"
-      case _ =>
-        aggFuncStr
+    filter match {
+      case Some(predicate) => s"$aggFuncStr FILTER ${predicate.sql}"
+      case _ => aggFuncStr
     }
   }
 }
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
@@ -19,14 +19,29 @@ package org.apache.spark.sql.execution.aggregate
 
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.aggregate._
-import org.apache.spark.sql.catalyst.optimizer.NormalizeFloatingNumbers
 import org.apache.spark.sql.execution.SparkPlan
 import org.apache.spark.sql.execution.streaming.{StateStoreRestoreExec, StateStoreSaveExec}
 
 /**
  * Utility functions used by the query planner to convert our plan to new aggregation code path.
  */
 object AggUtils {
+
+  private def mayRemoveAggFilters(exprs: Seq[AggregateExpression]): Seq[AggregateExpression] = {
+    exprs.map { ae =>
+      if (ae.filter.isDefined) {
+        ae.mode match {
+          // Aggregate filters are applicable only in partial/complete modes;
+          // this method filters out them, otherwise.
+          case Partial | Complete => ae
+          case _ => ae.copy(filter = None)
+        }
+      } else {
+        ae
+      }
+    }
+  }
+
   private def createAggregate(
       requiredChildDistributionExpressions: Option[Seq[Expression]] = None,
       groupingExpressions: Seq[NamedExpression] = Nil,
@@ -41,7 +56,7 @@ object AggUtils {
       HashAggregateExec(
         requiredChildDistributionExpressions = requiredChildDistributionExpressions,
         groupingExpressions = groupingExpressions,
-        aggregateExpressions = aggregateExpressions,
+        aggregateExpressions = mayRemoveAggFilters(aggregateExpressions),
         aggregateAttributes = aggregateAttributes,
         initialInputBufferOffset = initialInputBufferOffset,
         resultExpressions = resultExpressions,
@@ -54,7 +69,7 @@ object AggUtils {
         ObjectHashAggregateExec(
           requiredChildDistributionExpressions = requiredChildDistributionExpressions,
           groupingExpressions = groupingExpressions,
-          aggregateExpressions = aggregateExpressions,
+          aggregateExpressions = mayRemoveAggFilters(aggregateExpressions),
           aggregateAttributes = aggregateAttributes,
           initialInputBufferOffset = initialInputBufferOffset,
           resultExpressions = resultExpressions,
@@ -63,7 +78,7 @@ object AggUtils {
         SortAggregateExec(
           requiredChildDistributionExpressions = requiredChildDistributionExpressions,
           groupingExpressions = groupingExpressions,
-          aggregateExpressions = aggregateExpressions,
+          aggregateExpressions = mayRemoveAggFilters(aggregateExpressions),
           aggregateAttributes = aggregateAttributes,
           initialInputBufferOffset = initialInputBufferOffset,
           resultExpressions = resultExpressions,

Original file line number	Diff line number	Diff line change
`@@ -150,21 +150,17 @@ case class AggregateExpression(`
`150`	`150`	`case Final \| Complete => ""`
`151`	`151`	`}`
`152`	`152`	`val aggFuncStr = prefix + aggregateFunction.toAggString(isDistinct)`
`153`		`- mode match {`
`154`		`- case Partial \| Complete if filter.isDefined =>`
`155`		`- s"$aggFuncStr filter ${filter.get.toString}"`
`156`		`- case _ =>`
`157`		`- aggFuncStr`
	`153`	`+ filter match {`
	`154`	`+ case Some(predicate) => s"$aggFuncStr filter $predicate"`
	`155`	`+ case _ => aggFuncStr`
`158`	`156`	`}`
`159`	`157`	`}`
`160`	`158`
`161`	`159`	`override def sql: String = {`
`162`	`160`	`val aggFuncStr = aggregateFunction.sql(isDistinct)`
`163`		`- mode match {`
`164`		`- case Partial \| Complete if filter.isDefined =>`
`165`		`- s"$aggFuncStr FILTER ${filter.get.sql}"`
`166`		`- case _ =>`
`167`		`- aggFuncStr`
	`161`	`+ filter match {`
	`162`	`+ case Some(predicate) => s"$aggFuncStr FILTER ${predicate.sql}"`
	`163`	`+ case _ => aggFuncStr`
`168`	`164`	`}`
`169`	`165`	`}`
`170`	`166`	`}`