Skip calling encoder.shift many times in update and do not need to do shift when delta is zero.

Author: viirya

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala

@@ -339,9 +339,13 @@ case class ExpressionEncoder[T](
    * Returns a new encoder with input columns shifted by `delta` ordinals
    */
   def shift(delta: Int): ExpressionEncoder[T] = {
-    copy(deserializer = deserializer transform {
-      case r: BoundReference => r.copy(ordinal = r.ordinal + delta)
-    })
+    if (delta == 0) {
+      this
+    } else {
+      copy(deserializer = deserializer transform {
+        case r: BoundReference => r.copy(ordinal = r.ordinal + delta)
+      })
+    }
   }
 
   protected val attrs = serializer.flatMap(_.collect {

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TypedAggregateExpression.scala

@@ -85,6 +85,9 @@ case class TypedAggregateExpression(
     .resolve(aggBufferAttributes, OuterScopes.outerScopes)
     .bind(aggBufferAttributes)
 
+  val bEncoderForMutableAggBuffer = bEncoder.shift(mutableAggBufferOffset)
+  val bEncoderForInputAggBuffer = bEncoder.shift(inputAggBufferOffset)
+
   // Note: although this simply copies aggBufferAttributes, this common code can not be placed
   // in the superclass because that will lead to initialization ordering issues.
   override val inputAggBufferAttributes: Seq[AttributeReference] =
@@ -118,24 +121,24 @@ case class TypedAggregateExpression(
 
   override def update(buffer: MutableRow, input: InternalRow): Unit = {
     val inputA = boundA.fromRow(input)
-    val currentB = bEncoder.shift(mutableAggBufferOffset).fromRow(buffer)
+    val currentB = bEncoderForMutableAggBuffer.fromRow(buffer)
     val merged = aggregator.reduce(currentB, inputA)
     val returned = bEncoder.toRow(merged)
 
     updateBuffer(buffer, returned)
   }
 
   override def merge(buffer1: MutableRow, buffer2: InternalRow): Unit = {
-    val b1 = bEncoder.shift(mutableAggBufferOffset).fromRow(buffer1)
-    val b2 = bEncoder.shift(inputAggBufferOffset).fromRow(buffer2)
+    val b1 = bEncoderForMutableAggBuffer.fromRow(buffer1)
+    val b2 = bEncoderForInputAggBuffer.fromRow(buffer2)
     val merged = aggregator.merge(b1, b2)
     val returned = bEncoder.toRow(merged)
 
     updateBuffer(buffer1, returned)
   }
 
   override def eval(buffer: InternalRow): Any = {
-    val b = bEncoder.shift(mutableAggBufferOffset).fromRow(buffer)
+    val b = bEncoderForMutableAggBuffer.fromRow(buffer)
     val result = cEncoder.toRow(aggregator.finish(b))
     dataType match {
       case _: StructType => result