You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2025/01/22 07:26:25 UTC
(paimon) 21/24: [core] Fix that sequence fields are mistakenly aggregated by default aggregator in AggregateMergeFunction (#4977)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-1.0
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 15d2b943902df15c59b6e899fe444120ed08c530
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Wed Jan 22 10:58:14 2025 +0800
[core] Fix that sequence fields are mistakenly aggregated by default aggregator in AggregateMergeFunction (#4977)
---
.../compact/PartialUpdateMergeFunction.java | 50 ++++++++++++----------
.../compact/aggregate/AggregateMergeFunction.java | 40 +++++++++++------
.../aggregate/factory/FieldAggregatorFactory.java | 32 ++++----------
.../compact/aggregate/FieldAggregatorTest.java | 6 +--
.../apache/paimon/flink/PreAggregationITCase.java | 18 ++++++++
5 files changed, 82 insertions(+), 64 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
index 357461f74d..a28ac52df4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
@@ -24,6 +24,7 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
import org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory;
+import org.apache.paimon.mergetree.compact.aggregate.factory.FieldPrimaryKeyAggFactory;
import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
@@ -54,6 +55,7 @@ import static org.apache.paimon.CoreOptions.FIELDS_SEPARATOR;
import static org.apache.paimon.CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE;
import static org.apache.paimon.CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP;
import static org.apache.paimon.utils.InternalRowUtils.createFieldGetters;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
/**
* A {@link MergeFunction} where key is primary key (unique) and value is the partial record, update
@@ -352,10 +354,6 @@ public class PartialUpdateMergeFunction implements MergeFunction<KeyValue> {
this.fieldAggregators =
createFieldAggregators(
rowType, primaryKeys, allSequenceFields, new CoreOptions(options));
- if (!fieldAggregators.isEmpty() && fieldSeqComparators.isEmpty()) {
- throw new IllegalArgumentException(
- "Must use sequence group for aggregation functions.");
- }
removeRecordOnDelete = options.get(PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE);
@@ -526,41 +524,47 @@ public class PartialUpdateMergeFunction implements MergeFunction<KeyValue> {
List<String> fieldNames = rowType.getFieldNames();
List<DataType> fieldTypes = rowType.getFieldTypes();
Map<Integer, Supplier<FieldAggregator>> fieldAggregators = new HashMap<>();
- 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);
- boolean ignoreRetract = options.fieldAggIgnoreRetract(fieldName);
- if (strAggFunc != null) {
+ if (allSequenceFields.contains(fieldName)) {
+ // no agg for sequence fields
+ continue;
+ }
+
+ if (primaryKeys.contains(fieldName)) {
+ // aggregate by primary keys, so they do not aggregate
fieldAggregators.put(
i,
() ->
FieldAggregatorFactory.create(
fieldType,
- strAggFunc,
- ignoreRetract,
- isPrimaryKey,
- options,
- fieldName));
- } else if (defaultAggFunc != null && !allSequenceFields.contains(fieldName)) {
- // no agg for sequence fields
+ fieldName,
+ FieldPrimaryKeyAggFactory.NAME,
+ options));
+ continue;
+ }
+
+ String aggFuncName = getAggFuncName(options, fieldName);
+ if (aggFuncName != null) {
+ checkArgument(
+ !fieldSeqComparators.isEmpty(),
+ "Must use sequence group for aggregation functions.");
fieldAggregators.put(
i,
() ->
FieldAggregatorFactory.create(
- fieldType,
- defaultAggFunc,
- ignoreRetract,
- isPrimaryKey,
- options,
- fieldName));
+ fieldType, fieldName, aggFuncName, options));
}
}
return fieldAggregators;
}
+
+ @Nullable
+ private String getAggFuncName(CoreOptions options, String fieldName) {
+ String aggFunc = options.fieldAggFunc(fieldName);
+ return aggFunc == null ? options.fieldsDefaultFunc() : aggFunc;
+ }
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java
index bad77ba91d..ca380d2778 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java
@@ -25,6 +25,8 @@ import org.apache.paimon.data.InternalRow;
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;
@@ -132,27 +134,39 @@ public class AggregateMergeFunction implements MergeFunction<KeyValue> {
}
FieldAggregator[] fieldAggregators = new FieldAggregator[fieldNames.size()];
- String defaultAggFunc = options.fieldsDefaultFunc();
+ List<String> sequenceFields = options.sequenceField();
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;
- boolean ignoreRetract = options.fieldAggIgnoreRetract(fieldName);
+ String aggFuncName = getAggFuncName(fieldName, sequenceFields);
fieldAggregators[i] =
- FieldAggregatorFactory.create(
- fieldType,
- strAggFunc,
- ignoreRetract,
- isPrimaryKey,
- options,
- fieldName);
+ FieldAggregatorFactory.create(fieldType, fieldName, aggFuncName, options);
}
return new AggregateMergeFunction(createFieldGetters(fieldTypes), fieldAggregators);
}
+
+ private String getAggFuncName(String fieldName, List<String> sequenceFields) {
+ if (sequenceFields.contains(fieldName)) {
+ // no agg for sequence fields, use last_non_null_value to do cover
+ return FieldLastNonNullValueAggFactory.NAME;
+ }
+
+ if (primaryKeys.contains(fieldName)) {
+ // aggregate by primary keys, so they do not aggregate
+ return FieldPrimaryKeyAggFactory.NAME;
+ }
+
+ String aggFuncName = options.fieldAggFunc(fieldName);
+ if (aggFuncName == null) {
+ aggFuncName = options.fieldsDefaultFunc();
+ }
+ if (aggFuncName == null) {
+ // final default agg func
+ aggFuncName = FieldLastNonNullValueAggFactory.NAME;
+ }
+ return aggFuncName;
+ }
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldAggregatorFactory.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldAggregatorFactory.java
index d2ce0e4760..4801609914 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldAggregatorFactory.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldAggregatorFactory.java
@@ -25,8 +25,6 @@ import org.apache.paimon.mergetree.compact.aggregate.FieldAggregator;
import org.apache.paimon.mergetree.compact.aggregate.FieldIgnoreRetractAgg;
import org.apache.paimon.types.DataType;
-import javax.annotation.Nullable;
-
/** Factory for {@link FieldAggregator}. */
public interface FieldAggregatorFactory extends Factory {
@@ -35,37 +33,23 @@ public interface FieldAggregatorFactory extends Factory {
String identifier();
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;
- }
-
+ DataType fieldType, String fieldName, String aggFuncName, CoreOptions options) {
FieldAggregatorFactory fieldAggregatorFactory =
FactoryUtil.discoverFactory(
FieldAggregator.class.getClassLoader(),
FieldAggregatorFactory.class,
- strAgg);
+ aggFuncName);
if (fieldAggregatorFactory == null) {
throw new RuntimeException(
String.format(
"Use unsupported aggregation: %s or spell aggregate function incorrectly!",
- strAgg));
- }
-
- fieldAggregator = fieldAggregatorFactory.create(fieldType, options, field);
-
- if (ignoreRetract) {
- fieldAggregator = new FieldIgnoreRetractAgg(fieldAggregator);
+ aggFuncName));
}
- return fieldAggregator;
+ FieldAggregator fieldAggregator =
+ fieldAggregatorFactory.create(fieldType, options, fieldName);
+ return options.fieldAggIgnoreRetract(fieldName)
+ ? new FieldIgnoreRetractAgg(fieldAggregator)
+ : fieldAggregator;
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java
index d32098b80f..492532703c 100644
--- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java
@@ -936,10 +936,8 @@ public class FieldAggregatorTest {
FieldAggregatorFactory.create(
DataTypes.STRING(),
"custom",
- false,
- false,
- CoreOptions.fromMap(new HashMap<>()),
- "custom");
+ "custom",
+ CoreOptions.fromMap(new HashMap<>()));
Object agg = fieldAggregator.agg("test", "test");
assertThat(agg).isEqualTo("test");
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
index 954c1455d4..7b8ce3904e 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PreAggregationITCase.java
@@ -1195,6 +1195,24 @@ public class PreAggregationITCase {
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}. */