Apply comments

maropu · maropu · commit 00aac26c6a71 · 2016-06-25T10:46:26.000+09:00
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala
@@ -45,7 +45,7 @@ abstract class Collect extends ImperativeAggregate {
 
   override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
 
-  override def supportsPartial: Boolean = false
+  override def forceSortAggregate: Boolean = true
 
   override def aggBufferAttributes: Seq[AttributeReference] = Nil
 
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
@@ -175,9 +175,9 @@ sealed abstract class AggregateFunction extends Expression with ImplicitCastInpu
 
   /**
    * Indicates if this function needs to aggregate values group-by-group in a single step.
-   * If false, we must always use a `SortAggregateExec` operator without partial aggregates.
+   * If true, we must always use a `SortAggregateExec` operator without partial aggregates.
    */
-  def supportsPartial: Boolean = true
+  def forceSortAggregate: Boolean = false
 
   /**
    * Result of the aggregate function when the input is empty. This is currently only used for the
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/windowExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/windowExpressions.scala
@@ -431,7 +431,7 @@ abstract class AggregateWindowFunction extends DeclarativeAggregate with WindowF
   override val frame = SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow)
   override def dataType: DataType = IntegerType
   override def nullable: Boolean = true
-  override def supportsPartial: Boolean = false
+  override def forceSortAggregate: Boolean = true
   override lazy val mergeExpressions =
     throw new UnsupportedOperationException("Window Functions do not support merging.")
 }
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
@@ -261,7 +261,7 @@ private[sql] abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
         }
 
         val aggregateOperator =
-          if (aggregateExpressions.map(_.aggregateFunction).exists(!_.supportsPartial)) {
+          if (aggregateExpressions.map(_.aggregateFunction).exists(_.forceSortAggregate)) {
             if (functionsWithDistinct.nonEmpty) {
               sys.error("Distinct columns cannot exist in Aggregate operator containing " +
                 "aggregate functions which don't support partial aggregation.")
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFs.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFs.scala
@@ -336,7 +336,7 @@ private[hive] case class HiveUDAFFunction(
 
   override def nullable: Boolean = true
 
-  override def supportsPartial: Boolean = false
+  override def forceSortAggregate: Boolean = true
 
   override lazy val dataType: DataType = inspectorToDataType(returnInspector)
 

Original file line number	Diff line number	Diff line change
`@@ -431,7 +431,7 @@ abstract class AggregateWindowFunction extends DeclarativeAggregate with WindowF`
`431`	`431`	`override val frame = SpecifiedWindowFrame(RowFrame, UnboundedPreceding, CurrentRow)`
`432`	`432`	`override def dataType: DataType = IntegerType`
`433`	`433`	`override def nullable: Boolean = true`
`434`		`- override def supportsPartial: Boolean = false`
	`434`	`+ override def forceSortAggregate: Boolean = true`
`435`	`435`	`override lazy val mergeExpressions =`
`436`	`436`	`throw new UnsupportedOperationException("Window Functions do not support merging.")`
`437`	`437`	`}`
Original file line number	Diff line number	Diff line change
`@@ -261,7 +261,7 @@ private[sql] abstract class SparkStrategies extends QueryPlanner[SparkPlan] {`
`261`	`261`	`}`
`262`	`262`
`263`	`263`	`val aggregateOperator =`
`264`		`- if (aggregateExpressions.map(_.aggregateFunction).exists(!_.supportsPartial)) {`
	`264`	`+ if (aggregateExpressions.map(_.aggregateFunction).exists(_.forceSortAggregate)) {`
`265`	`265`	`if (functionsWithDistinct.nonEmpty) {`
`266`	`266`	`sys.error("Distinct columns cannot exist in Aggregate operator containing " +`
`267`	`267`	`"aggregate functions which don't support partial aggregation.")`