Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10228#discussion_r47276008
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggregationIterator.scala
---
@@ -165,155 +134,52 @@ abstract class AggregationIterator(
// 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
- }
- }
-
- // 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 joinedRow = new JoinedRow
+ val aggregationBufferSchema =
aggregateFunctions.flatMap(_.aggBufferAttributes)
+ val modes = aggregateExpressions.map(_.mode).distinct
+ if (aggregateExpressions.nonEmpty) {
+ val inputAggregationBufferSchema = if (initialInputBufferOffset ==
0) {
--- End diff --
Why doesn't this just check groupingKeyAttributes.nonEmpty?
---
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]
