Github user yhuai commented on a diff in the pull request:
https://github.com/apache/spark/pull/10228#discussion_r47439999
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggregationIterator.scala
---
@@ -154,248 +131,112 @@ abstract class AggregationIterator(
}
// All imperative AggregateFunctions.
- private[this] val allImperativeAggregateFunctions:
Array[ImperativeAggregate] =
+ protected[this] val allImperativeAggregateFunctions:
Array[ImperativeAggregate] =
allImperativeAggregateFunctionPositions
- .map(allAggregateFunctions)
+ .map(aggregateFunctions)
.map(_.asInstanceOf[ImperativeAggregate])
-
///////////////////////////////////////////////////////////////////////////
- // Methods and fields used by sub-classes.
-
///////////////////////////////////////////////////////////////////////////
-
// Initializing functions used to process a row.
- protected val processRow: (MutableRow, InternalRow) => Unit = {
- val rowToBeProcessed = new JoinedRow
- val aggregationBufferSchema =
allAggregateFunctions.flatMap(_.aggBufferAttributes)
- aggregationMode match {
- // Partial-only
- case (Some(Partial), None) =>
- val updateExpressions = nonCompleteAggregateFunctions.flatMap {
- case ae: DeclarativeAggregate => ae.updateExpressions
- case agg: AggregateFunction =>
Seq.fill(agg.aggBufferAttributes.length)(NoOp)
- }
- val expressionAggUpdateProjection =
- newMutableProjection(updateExpressions, aggregationBufferSchema
++ valueAttributes)()
-
- (currentBuffer: MutableRow, row: InternalRow) => {
- expressionAggUpdateProjection.target(currentBuffer)
- // Process all expression-based aggregate functions.
- expressionAggUpdateProjection(rowToBeProcessed(currentBuffer,
row))
- // Process all imperative aggregate functions.
- var i = 0
- while (i < nonCompleteImperativeAggregateFunctions.length) {
-
nonCompleteImperativeAggregateFunctions(i).update(currentBuffer, row)
- i += 1
- }
- }
-
- // PartialMerge-only or Final-only
- case (Some(PartialMerge), None) | (Some(Final), None) =>
- val inputAggregationBufferSchema = if (initialInputBufferOffset ==
0) {
- // If initialInputBufferOffset, the input value does not contain
- // grouping keys.
- // This part is pretty hacky.
- allAggregateFunctions.flatMap(_.inputAggBufferAttributes).toSeq
- } else {
- groupingKeyAttributes ++
allAggregateFunctions.flatMap(_.inputAggBufferAttributes)
- }
- // val inputAggregationBufferSchema =
- // groupingKeyAttributes ++
- // allAggregateFunctions.flatMap(_.cloneBufferAttributes)
- val mergeExpressions = nonCompleteAggregateFunctions.flatMap {
- case ae: DeclarativeAggregate => ae.mergeExpressions
- case agg: AggregateFunction =>
Seq.fill(agg.aggBufferAttributes.length)(NoOp)
- }
- // This projection is used to merge buffer values for all
expression-based aggregates.
- val expressionAggMergeProjection =
- newMutableProjection(
- mergeExpressions,
- aggregationBufferSchema ++ inputAggregationBufferSchema)()
-
- (currentBuffer: MutableRow, row: InternalRow) => {
- // Process all expression-based aggregate functions.
-
expressionAggMergeProjection.target(currentBuffer)(rowToBeProcessed(currentBuffer,
row))
- // Process all imperative aggregate functions.
- var i = 0
- while (i < nonCompleteImperativeAggregateFunctions.length) {
-
nonCompleteImperativeAggregateFunctions(i).merge(currentBuffer, row)
- i += 1
- }
- }
-
- // Final-Complete
- case (Some(Final), Some(Complete)) =>
- val completeAggregateFunctions: Array[AggregateFunction] =
-
allAggregateFunctions.takeRight(completeAggregateExpressions.length)
- // All imperative aggregate functions with mode Complete.
- val completeImperativeAggregateFunctions:
Array[ImperativeAggregate] =
- completeAggregateFunctions.collect { case func:
ImperativeAggregate => func }
-
- // The first initialInputBufferOffset values of the input
aggregation buffer is
- // for grouping expressions and distinct columns.
- val groupingAttributesAndDistinctColumns =
valueAttributes.take(initialInputBufferOffset)
-
- val completeOffsetExpressions =
-
Seq.fill(completeAggregateFunctions.map(_.aggBufferAttributes.length).sum)(NoOp)
- // We do not touch buffer values of aggregate functions with the
Final mode.
- val finalOffsetExpressions =
-
Seq.fill(nonCompleteAggregateFunctions.map(_.aggBufferAttributes.length).sum)(NoOp)
-
- val mergeInputSchema =
- aggregationBufferSchema ++
- groupingAttributesAndDistinctColumns ++
-
nonCompleteAggregateFunctions.flatMap(_.inputAggBufferAttributes)
- val mergeExpressions =
- nonCompleteAggregateFunctions.flatMap {
- case ae: DeclarativeAggregate => ae.mergeExpressions
- case agg: AggregateFunction =>
Seq.fill(agg.aggBufferAttributes.length)(NoOp)
- } ++ completeOffsetExpressions
- val finalExpressionAggMergeProjection =
- newMutableProjection(mergeExpressions, mergeInputSchema)()
-
- val updateExpressions =
- finalOffsetExpressions ++ completeAggregateFunctions.flatMap {
- case ae: DeclarativeAggregate => ae.updateExpressions
- case agg: AggregateFunction =>
Seq.fill(agg.aggBufferAttributes.length)(NoOp)
- }
- val completeExpressionAggUpdateProjection =
- newMutableProjection(updateExpressions, aggregationBufferSchema
++ valueAttributes)()
-
- (currentBuffer: MutableRow, row: InternalRow) => {
- val input = rowToBeProcessed(currentBuffer, row)
- // For all aggregate functions with mode Complete, update
buffers.
-
completeExpressionAggUpdateProjection.target(currentBuffer)(input)
- var i = 0
- while (i < completeImperativeAggregateFunctions.length) {
- completeImperativeAggregateFunctions(i).update(currentBuffer,
row)
- i += 1
- }
-
- // For all aggregate functions with mode Final, merge buffers.
- finalExpressionAggMergeProjection.target(currentBuffer)(input)
- i = 0
- while (i < nonCompleteImperativeAggregateFunctions.length) {
-
nonCompleteImperativeAggregateFunctions(i).merge(currentBuffer, row)
- i += 1
+ protected def generateProcessRow(
+ expressions: Seq[AggregateExpression],
+ functions: Seq[AggregateFunction],
+ inputAttributes: Seq[Attribute]): (MutableRow, InternalRow) => Unit = {
+ val joinedRow = new JoinedRow
+ if (expressions.nonEmpty) {
+ val mergeExpressions = functions.zipWithIndex.flatMap {
+ case (ae: DeclarativeAggregate, i) =>
+ expressions(i).mode match {
+ case Partial | Complete => ae.updateExpressions
+ case PartialMerge | Final => ae.mergeExpressions
}
- }
-
- // Complete-only
- case (None, Some(Complete)) =>
- val completeAggregateFunctions: Array[AggregateFunction] =
-
allAggregateFunctions.takeRight(completeAggregateExpressions.length)
- // All imperative aggregate functions with mode Complete.
- val completeImperativeAggregateFunctions:
Array[ImperativeAggregate] =
- completeAggregateFunctions.collect { case func:
ImperativeAggregate => func }
-
- val updateExpressions =
- completeAggregateFunctions.flatMap {
- case ae: DeclarativeAggregate => ae.updateExpressions
- case agg: AggregateFunction =>
Seq.fill(agg.aggBufferAttributes.length)(NoOp)
- }
- val completeExpressionAggUpdateProjection =
- newMutableProjection(updateExpressions, aggregationBufferSchema
++ valueAttributes)()
-
- (currentBuffer: MutableRow, row: InternalRow) => {
- val input = rowToBeProcessed(currentBuffer, row)
- // For all aggregate functions with mode Complete, update
buffers.
-
completeExpressionAggUpdateProjection.target(currentBuffer)(input)
- var i = 0
- while (i < completeImperativeAggregateFunctions.length) {
- completeImperativeAggregateFunctions(i).update(currentBuffer,
row)
- i += 1
+ case (agg: AggregateFunction, _) =>
Seq.fill(agg.aggBufferAttributes.length)(NoOp)
+ }
+ val updateFunctions = functions.zipWithIndex.collect {
+ case (ae: ImperativeAggregate, i) =>
+ expressions(i).mode match {
+ case Partial | Complete =>
+ (buffer: MutableRow, row: InternalRow) => ae.update(buffer,
row)
+ case PartialMerge | Final =>
+ (buffer: MutableRow, row: InternalRow) => ae.merge(buffer,
row)
}
+ }
+ // This projection is used to merge buffer values for all
expression-based aggregates.
+ val aggregationBufferSchema =
functions.flatMap(_.aggBufferAttributes)
+ val updateProjection =
+ newMutableProjection(mergeExpressions, aggregationBufferSchema ++
inputAttributes)()
+
+ (currentBuffer: MutableRow, row: InternalRow) => {
+ // Process all expression-based aggregate functions.
+ updateProjection.target(currentBuffer)(joinedRow(currentBuffer,
row))
+ // Process all imperative aggregate functions.
+ var i = 0
+ while (i < updateFunctions.length) {
+ updateFunctions(i)(currentBuffer, row)
+ i += 1
}
-
+ }
+ } else {
// Grouping only.
- case (None, None) => (currentBuffer: MutableRow, row: InternalRow)
=> {}
-
- case other =>
- sys.error(
- s"Could not evaluate ${nonCompleteAggregateExpressions} because
we do not " +
- s"support evaluate modes $other in this iterator.")
+ (currentBuffer: MutableRow, row: InternalRow) => {}
}
}
- // Initializing the function used to generate the output row.
- protected val generateOutput: (InternalRow, MutableRow) => InternalRow =
{
- val rowToBeEvaluated = new JoinedRow
- val safeOutputRow = new
SpecificMutableRow(resultExpressions.map(_.dataType))
- val mutableOutput = if (outputsUnsafeRows) {
-
UnsafeProjection.create(resultExpressions.map(_.dataType).toArray).apply(safeOutputRow)
- } else {
- safeOutputRow
- }
-
- aggregationMode match {
- // Partial-only or PartialMerge-only: every output row is basically
the values of
- // the grouping expressions and the corresponding aggregation buffer.
- case (Some(Partial), None) | (Some(PartialMerge), None) =>
- // Because we cannot copy a joinedRow containing a UnsafeRow
(UnsafeRow does not
- // support generic getter), we create a mutable projection to
output the
- // JoinedRow(currentGroupingKey, currentBuffer)
- val bufferSchema =
nonCompleteAggregateFunctions.flatMap(_.aggBufferAttributes)
- val resultProjection =
- newMutableProjection(
- groupingKeyAttributes ++ bufferSchema,
- groupingKeyAttributes ++ bufferSchema)()
- resultProjection.target(mutableOutput)
+ protected val processRow: (MutableRow, InternalRow) => Unit =
+ generateProcessRow(aggregateExpressions, aggregateFunctions,
inputAttributes)
- (currentGroupingKey: InternalRow, currentBuffer: MutableRow) => {
- resultProjection(rowToBeEvaluated(currentGroupingKey,
currentBuffer))
- // rowToBeEvaluated(currentGroupingKey, currentBuffer)
- }
+ protected val groupingProjection: UnsafeProjection =
+ UnsafeProjection.create(groupingExpressions, inputAttributes)
+ protected val groupingAttributes = groupingExpressions.map(_.toAttribute)
- // Final-only, Complete-only and Final-Complete: every output row
contains values representing
- // resultExpressions.
- case (Some(Final), None) | (Some(Final) | None, Some(Complete)) =>
- val bufferSchemata =
- allAggregateFunctions.flatMap(_.aggBufferAttributes)
- val evalExpressions = allAggregateFunctions.map {
- case ae: DeclarativeAggregate => ae.evaluateExpression
- case agg: AggregateFunction => NoOp
- }
- val expressionAggEvalProjection =
newMutableProjection(evalExpressions, bufferSchemata)()
- val aggregateResultSchema = nonCompleteAggregateAttributes ++
completeAggregateAttributes
- // TODO: Use unsafe row.
- val aggregateResult = new
SpecificMutableRow(aggregateResultSchema.map(_.dataType))
- expressionAggEvalProjection.target(aggregateResult)
- val resultProjection =
- newMutableProjection(
- resultExpressions, groupingKeyAttributes ++
aggregateResultSchema)()
- resultProjection.target(mutableOutput)
-
- (currentGroupingKey: InternalRow, currentBuffer: MutableRow) => {
- // Generate results for all expression-based aggregate functions.
- expressionAggEvalProjection(currentBuffer)
- // Generate results for all imperative aggregate functions.
- var i = 0
- while (i < allImperativeAggregateFunctions.length) {
- aggregateResult.update(
- allImperativeAggregateFunctionPositions(i),
- allImperativeAggregateFunctions(i).eval(currentBuffer))
- i += 1
- }
- resultProjection(rowToBeEvaluated(currentGroupingKey,
aggregateResult))
+ // Initializing the function used to generate the output row.
+ protected def generateResultProjection(): (UnsafeRow, MutableRow) =>
UnsafeRow = {
+ val joinedRow = new JoinedRow
+ val modes = aggregateExpressions.map(_.mode).distinct
+ val bufferAttributes =
aggregateFunctions.flatMap(_.aggBufferAttributes)
+ if (modes.contains(Final) || modes.contains(Complete)) {
+ val evalExpressions = aggregateFunctions.map {
+ case ae: DeclarativeAggregate => ae.evaluateExpression
+ case agg: AggregateFunction => NoOp
+ }
+ val aggregateResult = new
SpecificMutableRow(aggregateAttributes.map(_.dataType))
+ val expressionAggEvalProjection =
newMutableProjection(evalExpressions, bufferAttributes)()
+ expressionAggEvalProjection.target(aggregateResult)
+
+ val resultProjection =
+ UnsafeProjection.create(resultExpressions, groupingAttributes ++
aggregateAttributes)
+
+ (currentGroupingKey: UnsafeRow, currentBuffer: MutableRow) => {
+ // Generate results for all expression-based aggregate functions.
+ expressionAggEvalProjection(currentBuffer)
+ // Generate results for all imperative aggregate functions.
+ var i = 0
+ while (i < allImperativeAggregateFunctions.length) {
+ aggregateResult.update(
+ allImperativeAggregateFunctionPositions(i),
+ allImperativeAggregateFunctions(i).eval(currentBuffer))
+ i += 1
}
-
+ resultProjection(joinedRow(currentGroupingKey, aggregateResult))
+ }
+ } else if (modes.contains(Partial) || modes.contains(PartialMerge)) {
--- End diff --
same here
---
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]
