ScalaUDAF — Catalyst Expression Adapter for UserDefinedAggregateFunction

ScalaUDAF is a Catalyst expression adapter to manage the lifecycle of UserDefinedAggregateFunction and hook it in Spark SQL’s Catalyst execution path.

ScalaUDAF is created when:

UserDefinedAggregateFunction creates a Column for a user-defined aggregate function using all and distinct values (to use the UDAF in Dataset operators)
UDFRegistration is requested to register a user-defined aggregate function (to use the UDAF in SQL mode)

Table 1. ScalaUDAF’s ImperativeAggregate Methods
Method Name	Behaviour
initialize	Requests UserDefinedAggregateFunction to initialize
merge	Requests UserDefinedAggregateFunction to merge
update	Requests UserDefinedAggregateFunction to update

When evaluated, ScalaUDAF…FIXME

Table 2. ScalaUDAF’s Properties
Name	Description
`aggBufferAttributes`	AttributeReferences of aggBufferSchema
`aggBufferSchema`	bufferSchema of UserDefinedAggregateFunction
`dataType`	DataType of UserDefinedAggregateFunction
`deterministic`	`deterministic` of UserDefinedAggregateFunction
`inputAggBufferAttributes`	Copy of aggBufferAttributes
`inputTypes`	Data types from inputSchema of UserDefinedAggregateFunction
`nullable`	Always enabled (i.e. `true`)

ScalaUDAF takes the following when created:

initialize(buffer: InternalRow): Unit

Figure 1. ScalaUDAF initializes UserDefinedAggregateFunction

Note	`initialize` is part of ImperativeAggregate Contract.

update(mutableAggBuffer: InternalRow, inputRow: InternalRow): Unit

update sets the input buffer internal binary row as underlyingBuffer of MutableAggregationBufferImpl and requests the UserDefinedAggregateFunction to update.

Note	`update` uses inputProjection on the input `input` and converts it using inputToScalaConverters.

Figure 2. ScalaUDAF updates UserDefinedAggregateFunction

Note	`update` is part of ImperativeAggregate Contract.

merge(buffer1: InternalRow, buffer2: InternalRow): Unit

merge first sets:

Figure 3. ScalaUDAF requests UserDefinedAggregateFunction to merge

Note	`merge` is part of ImperativeAggregate Contract.

ScalaUDAF