[core] Fix that sequence group fields are mistakenly aggregated by default aggregator in partial update (apache#4897)

Author: yuzelin

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java

@@ -184,6 +184,7 @@ private void updateWithSequenceGroup(KeyValue kv) {
                             row.setField(
                                     fieldIndex, getters[fieldIndex].getFieldOrNull(kv.value()));
                         }
+                        continue;
                     }
                     row.setField(
                             i, aggregator == null ? field : aggregator.agg(accumulator, field));
@@ -304,6 +305,7 @@ private Factory(Options options, RowType rowType, List<String> primaryKeys) {
             List<String> fieldNames = rowType.getFieldNames();
             this.fieldSeqComparators = new HashMap<>();
             Map<String, Integer> sequenceGroupMap = new HashMap<>();
+            List<String> allSequenceFields = new ArrayList<>();
             for (Map.Entry<String, String> entry : options.toMap().entrySet()) {
                 String k = entry.getKey();
                 String v = entry.getValue();
@@ -318,6 +320,7 @@ private Factory(Options options, RowType rowType, List<String> primaryKeys) {
                                                     .split(FIELDS_SEPARATOR))
                                     .map(fieldName -> validateFieldName(fieldName, fieldNames))
                                     .collect(Collectors.toList());
+                    allSequenceFields.addAll(sequenceFields);
 
                     Supplier<FieldsComparator> userDefinedSeqComparator =
                             () -> UserDefinedSeqComparator.create(rowType, sequenceFields, true);
@@ -347,7 +350,8 @@ private Factory(Options options, RowType rowType, List<String> primaryKeys) {
                 }
             }
             this.fieldAggregators =
-                    createFieldAggregators(rowType, primaryKeys, new CoreOptions(options));
+                    createFieldAggregators(
+                            rowType, primaryKeys, allSequenceFields, new CoreOptions(options));
             if (!fieldAggregators.isEmpty() && fieldSeqComparators.isEmpty()) {
                 throw new IllegalArgumentException(
                         "Must use sequence group for aggregation functions.");
@@ -514,7 +518,10 @@ private String validateFieldName(String fieldName, List<String> fieldNames) {
          * @return The aggregators for each column.
          */
         private Map<Integer, Supplier<FieldAggregator>> createFieldAggregators(
-                RowType rowType, List<String> primaryKeys, CoreOptions options) {
+                RowType rowType,
+                List<String> primaryKeys,
+                List<String> allSequenceFields,
+                CoreOptions options) {
 
             List<String> fieldNames = rowType.getFieldNames();
             List<DataType> fieldTypes = rowType.getFieldTypes();
@@ -539,7 +546,8 @@ private Map<Integer, Supplier<FieldAggregator>> createFieldAggregators(
                                             isPrimaryKey,
                                             options,
                                             fieldName));
-                } else if (defaultAggFunc != null) {
+                } else if (defaultAggFunc != null && !allSequenceFields.contains(fieldName)) {
+                    // no agg for sequence fields
                     fieldAggregators.put(
                             i,
                             () ->

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java

@@ -723,4 +723,22 @@ public void testRemoveRecordOnDeleteLookup() throws Exception {
                         Row.ofKind(RowKind.UPDATE_AFTER, 1, "A", "apache"));
         iterator.close();
     }
+
+    @Test
+    public void testSequenceGroupWithDefaultAgg() {
+        sql(
+                "CREATE TABLE seq_default_agg ("
+                        + " pk INT PRIMARY KEY NOT ENFORCED,"
+                        + " seq INT,"
+                        + " v INT) WITH ("
+                        + " 'merge-engine'='partial-update',"
+                        + " 'fields.seq.sequence-group'='v',"
+                        + " '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));
+    }
 }