Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/7687#discussion_r35510193
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/udaf.scala ---
@@ -15,87 +15,29 @@
* limitations under the License.
*/
-package org.apache.spark.sql.expressions.aggregate
+package org.apache.spark.sql.execution.aggregate
import org.apache.spark.Logging
-import
org.apache.spark.sql.catalyst.expressions.codegen.GenerateMutableProjection
+import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.{InternalRow, CatalystTypeConverters}
-import org.apache.spark.sql.catalyst.expressions._
+import
org.apache.spark.sql.catalyst.expressions.codegen.GenerateMutableProjection
+import org.apache.spark.sql.catalyst.expressions.{MutableRow,
InterpretedMutableProjection, AttributeReference, Expression}
import
org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction2
-import org.apache.spark.sql.types._
-import org.apache.spark.sql.Row
+import org.apache.spark.sql.expressions.{MutableAggregationBuffer,
UserDefinedAggregateFunction}
+import org.apache.spark.sql.types.{Metadata, StructField, StructType,
DataType}
/**
- * The abstract class for implementing user-defined aggregate function.
+ * A Mutable [[Row]] representing an mutable aggregation buffer.
*/
-abstract class UserDefinedAggregateFunction extends Serializable {
-
- /**
- * A [[StructType]] represents data types of input arguments of this
aggregate function.
- * For example, if a [[UserDefinedAggregateFunction]] expects two input
arguments
- * with type of [[DoubleType]] and [[LongType]], the returned
[[StructType]] will look like
- *
- * ```
- * StructType(Seq(StructField("doubleInput", DoubleType),
StructField("longInput", LongType)))
- * ```
- *
- * The name of a field of this [[StructType]] is only used to identify
the corresponding
- * input argument. Users can choose names to identify the input
arguments.
- */
- def inputSchema: StructType
-
- /**
- * A [[StructType]] represents data types of values in the aggregation
buffer.
- * For example, if a [[UserDefinedAggregateFunction]]'s buffer has two
values
- * (i.e. two intermediate values) with type of [[DoubleType]] and
[[LongType]],
- * the returned [[StructType]] will look like
- *
- * ```
- * StructType(Seq(StructField("doubleInput", DoubleType),
StructField("longInput", LongType)))
- * ```
- *
- * The name of a field of this [[StructType]] is only used to identify
the corresponding
- * buffer value. Users can choose names to identify the input arguments.
- */
- def bufferSchema: StructType
-
- /**
- * The [[DataType]] of the returned value of this
[[UserDefinedAggregateFunction]].
- */
- def returnDataType: DataType
-
- /** Indicates if this function is deterministic. */
- def deterministic: Boolean
-
- /**
- * Initializes the given aggregation buffer. Initial values set by this
method should satisfy
- * the condition that when merging two buffers with initial values, the
new buffer should
- * still store initial values.
- */
- def initialize(buffer: MutableAggregationBuffer): Unit
-
- /** Updates the given aggregation buffer `buffer` with new input data
from `input`. */
- def update(buffer: MutableAggregationBuffer, input: Row): Unit
-
- /** Merges two aggregation buffers and stores the updated buffer values
back in `buffer1`. */
- def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit
-
- /**
- * Calculates the final result of this [[UserDefinedAggregateFunction]]
based on the given
- * aggregation buffer.
- */
- def evaluate(buffer: Row): Any
-}
-
-private[sql] abstract class AggregationBuffer(
+private[sql] class MutableAggregationBufferImpl (
+ schema: StructType,
toCatalystConverters: Array[Any => Any],
toScalaConverters: Array[Any => Any],
- bufferOffset: Int)
- extends Row {
-
- override def length: Int = toCatalystConverters.length
+ bufferOffset: Int,
+ var underlyingBuffer: MutableRow)
+ extends MutableAggregationBuffer {
- protected val offsets: Array[Int] = {
+ private val offsets: Array[Int] = {
--- End diff --
private[this]
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
