[core] Fix that sequence fields are mistakenly aggregated by default …

…aggregator in AggregateMergeFunction
yuzelin · Jan 21, 2025 · 30e1fc9 · 30e1fc9
1 parent cfb0075
commit 30e1fc9
Show file tree

Hide file tree

Showing 5 changed files with 45 additions and 30 deletions.
diff --git a/...on-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java b/...on-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
@@ -543,7 +543,6 @@ private Map<Integer, Supplier<FieldAggregator>> createFieldAggregators(
                                             fieldType,
                                             strAggFunc,
                                             ignoreRetract,
-                                            isPrimaryKey,
                                             options,
                                             fieldName));
                 } else if (defaultAggFunc != null && !allSequenceFields.contains(fieldName)) {
@@ -555,7 +554,6 @@ private Map<Integer, Supplier<FieldAggregator>> createFieldAggregators(
                                             fieldType,
                                             defaultAggFunc,
                                             ignoreRetract,
-                                            isPrimaryKey,
                                             options,
                                             fieldName));
                 }

diff --git a/...e/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java b/...e/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java
@@ -25,6 +25,8 @@
 import org.apache.paimon.mergetree.compact.MergeFunction;
 import org.apache.paimon.mergetree.compact.MergeFunctionFactory;
 import org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
+import org.apache.paimon.mergetree.compact.aggregate.factory.FieldLastNonNullValueAggFactory;
+import org.apache.paimon.mergetree.compact.aggregate.factory.FieldPrimaryKeyAggFactory;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.RowKind;
@@ -109,6 +111,8 @@ private static class Factory implements MergeFunctionFactory<KeyValue> {
         private final List<String> tableNames;
         private final List<DataType> tableTypes;
         private final List<String> primaryKeys;
+        private final List<String> sequenceFields;
+        @Nullable private final String defaultAggFunc;
 
         private Factory(
                 Options conf,
@@ -119,6 +123,11 @@ private Factory(
             this.tableNames = tableNames;
             this.tableTypes = tableTypes;
             this.primaryKeys = primaryKeys;
+            this.sequenceFields = options.sequenceField();
+
+            String defaultAggFunc = options.fieldsDefaultFunc();
+            this.defaultAggFunc =
+                    defaultAggFunc == null ? FieldLastNonNullValueAggFactory.NAME : defaultAggFunc;
         }
 
         @Override
@@ -132,27 +141,31 @@ public MergeFunction<KeyValue> create(@Nullable int[][] projection) {
             }
 
             FieldAggregator[] fieldAggregators = new FieldAggregator[fieldNames.size()];
-            String defaultAggFunc = options.fieldsDefaultFunc();
             for (int i = 0; i < fieldNames.size(); i++) {
                 String fieldName = fieldNames.get(i);
                 DataType fieldType = fieldTypes.get(i);
-                // aggregate by primary keys, so they do not aggregate
-                boolean isPrimaryKey = primaryKeys.contains(fieldName);
-                String strAggFunc = options.fieldAggFunc(fieldName);
-                strAggFunc = strAggFunc == null ? defaultAggFunc : strAggFunc;
-
+                String strAggFunc = getAggFunc(fieldName);
                 boolean ignoreRetract = options.fieldAggIgnoreRetract(fieldName);
                 fieldAggregators[i] =
                         FieldAggregatorFactory.create(
-                                fieldType,
-                                strAggFunc,
-                                ignoreRetract,
-                                isPrimaryKey,
-                                options,
-                                fieldName);
+                                fieldType, strAggFunc, ignoreRetract, options, fieldName);
             }
 
             return new AggregateMergeFunction(createFieldGetters(fieldTypes), fieldAggregators);
         }
+
+        private String getAggFunc(String fieldName) {
+            if (primaryKeys.contains(fieldName)) {
+                // aggregate by primary keys, so they do not aggregate
+                return FieldPrimaryKeyAggFactory.NAME;
+            }
+
+            if (sequenceFields.contains(fieldName)) {
+                return FieldLastNonNullValueAggFactory.NAME;
+            }
+
+            String strAggFunc = options.fieldAggFunc(fieldName);
+            return strAggFunc == null ? defaultAggFunc : strAggFunc;
+        }
     }
 }
diff --git a/...in/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldAggregatorFactory.java b/...in/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldAggregatorFactory.java
@@ -38,16 +38,8 @@ static FieldAggregator create(
             DataType fieldType,
             @Nullable String strAgg,
             boolean ignoreRetract,
-            boolean isPrimaryKey,
             CoreOptions options,
             String field) {
-        FieldAggregator fieldAggregator;
-        if (isPrimaryKey) {
-            strAgg = FieldPrimaryKeyAggFactory.NAME;
-        } else if (strAgg == null) {
-            strAgg = FieldLastNonNullValueAggFactory.NAME;
-        }
-
         FieldAggregatorFactory fieldAggregatorFactory =
                 FactoryUtil.discoverFactory(
                         FieldAggregator.class.getClassLoader(),
@@ -60,12 +52,7 @@ static FieldAggregator create(
                             strAgg));
         }
 
-        fieldAggregator = fieldAggregatorFactory.create(fieldType, options, field);
-
-        if (ignoreRetract) {
-            fieldAggregator = new FieldIgnoreRetractAgg(fieldAggregator);
-        }
-
-        return fieldAggregator;
+        FieldAggregator fieldAggregator = fieldAggregatorFactory.create(fieldType, options, field);
+        return ignoreRetract ? new FieldIgnoreRetractAgg(fieldAggregator) : fieldAggregator;
     }
 }
diff --git a/...core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java b/...core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java
@@ -937,7 +937,6 @@ public void testCustomAgg() throws IOException {
                         DataTypes.STRING(),
                         "custom",
                         false,
-                        false,
                         CoreOptions.fromMap(new HashMap<>()),
                         "custom");
 

diff --git a/...flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java b/...flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
@@ -1195,6 +1195,24 @@ public void testMergeRead() {
             assertThat(batchSql("SELECT * FROM T where v = 1"))
                     .containsExactlyInAnyOrder(Row.of(2, 1, 1));
         }
+
+        @Test
+        public void testSequenceFieldWithDefaultAgg() {
+            sql(
+                    "CREATE TABLE seq_default_agg ("
+                            + " pk INT PRIMARY KEY NOT ENFORCED,"
+                            + " seq INT,"
+                            + " v INT) WITH ("
+                            + " 'merge-engine'='aggregation',"
+                            + " 'sequence.field'='seq',"
+                            + " 'fields.default-aggregate-function'='sum'"
+                            + ")");
+
+            sql("INSERT INTO seq_default_agg VALUES (0, 1, 1)");
+            sql("INSERT INTO seq_default_agg VALUES (0, 2, 2)");
+
+            assertThat(sql("SELECT * FROM seq_default_agg")).containsExactly(Row.of(0, 2, 3));
+        }
     }
 
     /** ITCase for {@link FieldNestedUpdateAgg}. */