improve performance

Author: platinumhamburg

fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMerger.java

@@ -17,10 +17,6 @@
 
 package org.apache.fluss.server.kv.rowmerger;
 
-/* This file is based on source code of Apache Paimon Project (https://paimon.apache.org/), licensed by the Apache
- * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
- * additional information regarding copyright ownership. */
-
 import org.apache.fluss.config.TableConfig;
 import org.apache.fluss.metadata.DeleteBehavior;
 import org.apache.fluss.metadata.KvFormat;
@@ -31,7 +27,6 @@
 import org.apache.fluss.row.BinaryRow;
 import org.apache.fluss.row.GenericRow;
 import org.apache.fluss.row.InternalRow;
-import org.apache.fluss.row.encode.RowEncoder;
 import org.apache.fluss.server.kv.rowmerger.aggregate.AggregationContext;
 import org.apache.fluss.server.kv.rowmerger.aggregate.AggregationContextCache;
 import org.apache.fluss.server.kv.rowmerger.aggregate.FieldAggregator;
@@ -94,40 +89,29 @@ private void aggregateFields(
             BinaryRow newRow,
             AggregationContext newContext,
             AggregationContext oldContext) {
-        for (int i = 0; i < newContext.getFieldCount(); i++) {
-            FieldAggregator aggregator = newContext.getAggregators()[i];
+        // Cache loop-invariant values and array references for better performance
+        int newFieldCount = newContext.getFieldCount();
+        int oldFieldCount = oldContext.getFieldCount();
+        FieldAggregator[] newAggregators = newContext.getAggregators();
+        InternalRow.FieldGetter[] newFieldGetters = newContext.getFieldGetters();
+        InternalRow.FieldGetter[] oldFieldGetters = oldContext.getFieldGetters();
+
+        for (int i = 0; i < newFieldCount; i++) {
+            FieldAggregator aggregator = newAggregators[i];
 
             // Get accumulator value from oldRow (null if field doesn't exist in old schema)
             Object accumulator =
-                    (i < oldContext.getFieldCount())
-                            ? oldContext.getFieldGetters()[i].getFieldOrNull(oldRow)
-                            : null;
+                    (i < oldFieldCount) ? oldFieldGetters[i].getFieldOrNull(oldRow) : null;
 
             // Get input value from newRow (null if field doesn't exist in new schema)
-            Object inputField = newContext.getFieldGetters()[i].getFieldOrNull(newRow);
+            Object inputField = newFieldGetters[i].getFieldOrNull(newRow);
 
             // Apply aggregate function and store result
             Object mergedField = aggregator.agg(accumulator, inputField);
             accumulatorRow.setField(i, mergedField);
         }
     }
 
-    /**
-     * Encode a GenericRow to BinaryRow using the new schema encoder.
-     *
-     * @param accumulatorRow the GenericRow to encode
-     * @param newContext context for the new schema
-     * @return the encoded BinaryRow
-     */
-    private BinaryRow encodeRow(GenericRow accumulatorRow, AggregationContext newContext) {
-        RowEncoder encoder = newContext.getRowEncoder();
-        encoder.startNewRow();
-        for (int i = 0; i < newContext.getFieldCount(); i++) {
-            encoder.encodeField(i, accumulatorRow.getField(i));
-        }
-        return encoder.finishRow();
-    }
-
     @Override
     public BinaryValue merge(BinaryValue oldValue, BinaryValue newValue) {
         // First write: no existing row
@@ -158,7 +142,7 @@ public BinaryValue merge(BinaryValue oldValue, BinaryValue newValue) {
         aggregateFields(accumulatorRow, oldValue.row, newValue.row, newContext, oldContext);
 
         // Encode aggregated row to BinaryRow
-        BinaryRow mergedRow = encodeRow(accumulatorRow, newContext);
+        BinaryRow mergedRow = newContext.encodeRow(accumulatorRow);
 
         return new BinaryValue(newValue.schemaId, mergedRow);
     }
@@ -346,12 +330,7 @@ public BinaryValue merge(BinaryValue oldValue, BinaryValue newValue) {
             mergeRows(accumulatorRow, oldValue.row, newValue.row, oldContext, newContext);
 
             // Encode and return merged row using new schema encoder
-            RowEncoder encoder = newContext.getRowEncoder();
-            encoder.startNewRow();
-            for (int i = 0; i < newFieldCount; i++) {
-                encoder.encodeField(i, accumulatorRow.getField(i));
-            }
-            BinaryRow mergedRow = encoder.finishRow();
+            BinaryRow mergedRow = newContext.encodeRow(accumulatorRow);
 
             return new BinaryValue(newValue.schemaId, mergedRow);
         }

fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregationContext.java

@@ -21,6 +21,8 @@
 import org.apache.fluss.metadata.AggFunction;
 import org.apache.fluss.metadata.KvFormat;
 import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.row.BinaryRow;
+import org.apache.fluss.row.GenericRow;
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.row.encode.RowEncoder;
 import org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldAggregatorFactory;
@@ -75,6 +77,23 @@ public int getFieldCount() {
         return fieldCount;
     }
 
+    /**
+     * Encode a GenericRow to BinaryRow using this context's encoder.
+     *
+     * <p>This method is optimized to reuse the encoder from the context and minimize overhead
+     * during row encoding.
+     *
+     * @param accumulatorRow the GenericRow to encode
+     * @return the encoded BinaryRow
+     */
+    public BinaryRow encodeRow(GenericRow accumulatorRow) {
+        rowEncoder.startNewRow();
+        for (int i = 0; i < fieldCount; i++) {
+            rowEncoder.encodeField(i, accumulatorRow.getField(i));
+        }
+        return rowEncoder.finishRow();
+    }
+
     /**
      * Create an aggregation context for a given schema.
      *

fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/FieldAggregator.java

@@ -23,6 +23,7 @@
  * additional information regarding copyright ownership. */
 
 import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.DataTypeRoot;
 
 import java.io.Serializable;
 
@@ -32,11 +33,13 @@ public abstract class FieldAggregator implements Serializable {
     private static final long serialVersionUID = 1L;
 
     protected final DataType fieldType;
+    protected final DataTypeRoot typeRoot;
     protected final String name;
 
     public FieldAggregator(String name, DataType dataType) {
         this.name = name;
         this.fieldType = dataType;
+        this.typeRoot = dataType.getTypeRoot();
     }
 
     /**

fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/FieldMaxAgg.java

@@ -23,7 +23,6 @@
  * additional information regarding copyright ownership. */
 
 import org.apache.fluss.types.DataType;
-import org.apache.fluss.types.DataTypeRoot;
 import org.apache.fluss.utils.InternalRowUtils;
 
 /** Max aggregator - keeps the maximum value. */
@@ -40,8 +39,7 @@ public Object agg(Object accumulator, Object inputField) {
         if (accumulator == null || inputField == null) {
             return accumulator == null ? inputField : accumulator;
         }
-        DataTypeRoot type = fieldType.getTypeRoot();
-        return InternalRowUtils.compare(accumulator, inputField, type) < 0
+        return InternalRowUtils.compare(accumulator, inputField, typeRoot) < 0
                 ? inputField
                 : accumulator;
     }

fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/FieldMinAgg.java

@@ -23,7 +23,6 @@
  * additional information regarding copyright ownership. */
 
 import org.apache.fluss.types.DataType;
-import org.apache.fluss.types.DataTypeRoot;
 import org.apache.fluss.utils.InternalRowUtils;
 
 /** Min aggregator - keeps the minimum value. */
@@ -41,8 +40,7 @@ public Object agg(Object accumulator, Object inputField) {
             return accumulator == null ? inputField : accumulator;
         }
 
-        DataTypeRoot type = fieldType.getTypeRoot();
-        return InternalRowUtils.compare(accumulator, inputField, type) < 0
+        return InternalRowUtils.compare(accumulator, inputField, typeRoot) < 0
                 ? accumulator
                 : inputField;
     }

fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/FieldProductAgg.java

@@ -47,7 +47,7 @@ public Object agg(Object accumulator, Object inputField) {
         Object product;
 
         // ordered by type root definition
-        switch (fieldType.getTypeRoot()) {
+        switch (typeRoot) {
             case DECIMAL:
                 Decimal mergeFieldDD = (Decimal) accumulator;
                 Decimal inFieldDD = (Decimal) inputField;
@@ -82,7 +82,7 @@ public Object agg(Object accumulator, Object inputField) {
                 String msg =
                         String.format(
                                 "type %s not support in %s",
-                                fieldType.getTypeRoot().toString(), this.getClass().getName());
+                                typeRoot.toString(), this.getClass().getName());
                 throw new IllegalArgumentException(msg);
         }
         return product;

fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/FieldSumAgg.java

@@ -43,7 +43,7 @@ public Object agg(Object accumulator, Object inputField) {
         Object sum;
 
         // ordered by type root definition
-        switch (fieldType.getTypeRoot()) {
+        switch (typeRoot) {
             case DECIMAL:
                 Decimal mergeFieldDD = (Decimal) accumulator;
                 Decimal inFieldDD = (Decimal) inputField;
@@ -80,7 +80,7 @@ public Object agg(Object accumulator, Object inputField) {
                 String msg =
                         String.format(
                                 "type %s not support in %s",
-                                fieldType.getTypeRoot().toString(), this.getClass().getName());
+                                typeRoot.toString(), this.getClass().getName());
                 throw new IllegalArgumentException(msg);
         }
         return sum;