This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new ce9817b9e8 [core] Support nested-key-null-strategy in
FieldNestedPartialUpdateAgg operator. (#8461)
ce9817b9e8 is described below
commit ce9817b9e86079f7cd236c9c87c320dedc93e8d3
Author: AuroraVoyage <[email protected]>
AuthorDate: Sat Jul 4 22:27:07 2026 +0800
[core] Support nested-key-null-strategy in FieldNestedPartialUpdateAgg
operator. (#8461)
---
.../primary-key-table/merge-engine/aggregation.mdx | 7 ++
.../aggregate/FieldNestedPartialUpdateAgg.java | 51 +++++++++++--
.../FieldNestedPartialUpdateAggFactory.java | 11 ++-
.../compact/aggregate/FieldAggregatorTest.java | 89 +++++++++++++++++++++-
4 files changed, 148 insertions(+), 10 deletions(-)
diff --git a/docs/docs/primary-key-table/merge-engine/aggregation.mdx
b/docs/docs/primary-key-table/merge-engine/aggregation.mdx
index 1f3cab9340..823ae7b016 100644
--- a/docs/docs/primary-key-table/merge-engine/aggregation.mdx
+++ b/docs/docs/primary-key-table/merge-engine/aggregation.mdx
@@ -392,6 +392,13 @@ public static class BitmapContainsUDF extends
ScalarFunction {
ARRAY\<ROW\> data types. You need to use
`fields.<field-name>.nested-key=pk0,pk1,...` to specify the primary keys of the
nested table. The values in each row are written by partial updating some
columns.
+ Use `fields.<field-name>.nested-key-null-strategy=<merge|ignore|error>` to
specify how rows are handled when the nested-key does not satisfy primary key
semantics (e.g., primary key fields contain null values).
+
+ - `merge` (default): Merge rows even if the nested-key does not satisfy
primary key semantics. This is the same behavior as when this option is not
configured.
+ - `ignore`: Ignore rows whose nested-key contains null values because they
do not satisfy primary key semantics.
+ - `error`: Throw an exception if the nested-key contains null values,
because primary key fields must not be null.
+
+
### collect
The collect function collects elements into an Array. You can set
`fields.<field-name>.distinct=true` to deduplicate elements.
It only supports ARRAY type.
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedPartialUpdateAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedPartialUpdateAgg.java
index d8e058e839..06e9e64a2f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedPartialUpdateAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedPartialUpdateAgg.java
@@ -18,6 +18,7 @@
package org.apache.paimon.mergetree.compact.aggregate;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.codegen.Projection;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericArray;
@@ -29,6 +30,7 @@ import org.apache.paimon.types.ArrayType;
import org.apache.paimon.types.RowType;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -47,8 +49,13 @@ public class FieldNestedPartialUpdateAgg extends
FieldAggregator {
private final int nestedFields;
private final Projection keyProjection;
private final FieldGetter[] fieldGetters;
+ private final CoreOptions.NestedKeyNullStrategy nestedKeyNullStrategy;
- public FieldNestedPartialUpdateAgg(String name, ArrayType dataType,
List<String> nestedKey) {
+ public FieldNestedPartialUpdateAgg(
+ String name,
+ ArrayType dataType,
+ List<String> nestedKey,
+ CoreOptions.NestedKeyNullStrategy nestedKeyNullStrategy) {
super(name, dataType);
RowType nestedType = (RowType) dataType.getElementType();
this.nestedFields = nestedType.getFieldCount();
@@ -58,25 +65,34 @@ public class FieldNestedPartialUpdateAgg extends
FieldAggregator {
for (int i = 0; i < nestedFields; i++) {
fieldGetters[i] =
InternalRow.createFieldGetter(nestedType.getTypeAt(i), i);
}
+ this.nestedKeyNullStrategy = nestedKeyNullStrategy;
}
@Override
public Object agg(Object accumulator, Object inputField) {
- if (accumulator == null || inputField == null) {
- return accumulator == null ? inputField : accumulator;
+ if (inputField == null) {
+ return accumulator;
}
- InternalArray acc = (InternalArray) accumulator;
InternalArray input = (InternalArray) inputField;
- List<InternalRow> rows = new ArrayList<>(acc.size() + input.size());
- addNonNullRows(acc, rows);
+ List<InternalRow> rows;
+ if (accumulator == null) {
+ rows = new ArrayList<>(input.size());
+ } else {
+ InternalArray acc = (InternalArray) accumulator;
+ rows = new ArrayList<>(acc.size() + input.size());
+ addNonNullRows(acc, rows);
+ }
addNonNullRows(input, rows);
if (keyProjection != null) {
Map<BinaryRow, GenericRow> map = new HashMap<>();
for (InternalRow row : rows) {
BinaryRow key = keyProjection.apply(row).copy();
+ if (!applyNestedKeyNullStrategy(key)) {
+ continue;
+ }
GenericRow toUpdate = map.computeIfAbsent(key, k -> new
GenericRow(nestedFields));
partialUpdate(toUpdate, row);
}
@@ -105,4 +121,27 @@ public class FieldNestedPartialUpdateAgg extends
FieldAggregator {
}
}
}
+
+ private boolean applyNestedKeyNullStrategy(BinaryRow key) {
+ if (!key.anyNull()) {
+ // The nested-key satisfies primary key semantics.
+ return true;
+ }
+ switch (nestedKeyNullStrategy) {
+ case MERGE:
+ // Preserve the previous behavior.
+ return true;
+ case IGNORE:
+ return false;
+ case ERROR:
+ throw new IllegalArgumentException(
+ "Nested key contains null values. Primary key fields
must not be null.");
+ default:
+ throw new UnsupportedOperationException(
+ String.format(
+ "Unsupported nested-key-null-strategy '%s'.
Supported values are: %s.",
+ nestedKeyNullStrategy,
+
Arrays.toString(CoreOptions.NestedKeyNullStrategy.values())));
+ }
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldNestedPartialUpdateAggFactory.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldNestedPartialUpdateAggFactory.java
index 80b899a8b2..9a22dc5cd8 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldNestedPartialUpdateAggFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldNestedPartialUpdateAggFactory.java
@@ -37,7 +37,9 @@ public class FieldNestedPartialUpdateAggFactory implements
FieldAggregatorFactor
public FieldNestedPartialUpdateAgg create(
DataType fieldType, CoreOptions options, String field) {
return createFieldNestedPartialUpdateAgg(
- fieldType, options.fieldNestedUpdateAggNestedKey(field));
+ fieldType,
+ options.fieldNestedUpdateAggNestedKey(field),
+ options.fieldNestedUpdateAggNestedKeyNullStrategy(field));
}
@Override
@@ -46,13 +48,16 @@ public class FieldNestedPartialUpdateAggFactory implements
FieldAggregatorFactor
}
private FieldNestedPartialUpdateAgg createFieldNestedPartialUpdateAgg(
- DataType fieldType, List<String> nestedKey) {
+ DataType fieldType,
+ List<String> nestedKey,
+ CoreOptions.NestedKeyNullStrategy nestedKeyNullStrategy) {
checkArgument(!nestedKey.isEmpty());
String typeErrorMsg =
"Data type for nested table column must be 'Array<Row>' but
was '%s'.";
checkArgument(fieldType instanceof ArrayType, typeErrorMsg, fieldType);
ArrayType arrayType = (ArrayType) fieldType;
checkArgument(arrayType.getElementType() instanceof RowType,
typeErrorMsg, fieldType);
- return new FieldNestedPartialUpdateAgg(identifier(), arrayType,
nestedKey);
+ return new FieldNestedPartialUpdateAgg(
+ identifier(), arrayType, nestedKey, nestedKeyNullStrategy);
}
}
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 77cc470cfb..183a1b63ef 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
@@ -2423,7 +2423,8 @@ public class FieldAggregatorTest {
DataTypes.FIELD(0, "k",
DataTypes.INT()),
DataTypes.FIELD(1, "v1",
DataTypes.INT()),
DataTypes.FIELD(2, "v2",
DataTypes.STRING()))),
- Collections.singletonList("k"));
+ Collections.singletonList("k"),
+ CoreOptions.NestedKeyNullStrategy.MERGE);
InternalArray accumulator;
InternalArray.ElementGetter elementGetter =
@@ -2448,6 +2449,92 @@ public class FieldAggregatorTest {
accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
assertThat(unnest(accumulator, elementGetter))
.containsExactlyInAnyOrderElementsOf(Arrays.asList(row(0, 1,
"B"), row(1, 2, "C")));
+
+ // Verify MERGE strategy keeps rows with null nested keys.
+ current = row(null, 0, "D");
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Arrays.asList(row(0, 1, "B"), row(1, 2, "C"),
row(null, 0, "D")));
+ }
+
+ @Test
+ public void
testFieldNestedPartialUpdateAggWithNestedKeyNullUseIgnoreStrategy() {
+ DataType elementRowType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "k", DataTypes.INT()),
+ DataTypes.FIELD(1, "v1", DataTypes.INT()),
+ DataTypes.FIELD(2, "v2", DataTypes.STRING()));
+ FieldNestedPartialUpdateAgg agg =
+ new FieldNestedPartialUpdateAgg(
+ FieldNestedPartialUpdateAggFactory.NAME,
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "k",
DataTypes.INT()),
+ DataTypes.FIELD(1, "v1",
DataTypes.INT()),
+ DataTypes.FIELD(2, "v2",
DataTypes.STRING()))),
+ Collections.singletonList("k"),
+ CoreOptions.NestedKeyNullStrategy.IGNORE);
+
+ InternalArray accumulator;
+ InternalArray.ElementGetter elementGetter =
+ InternalArray.createElementGetter(elementRowType);
+
+ InternalRow current = row(0, 0, null);
+ accumulator = (InternalArray) agg.agg(null, singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+
.containsExactlyInAnyOrderElementsOf(Collections.singletonList(current));
+
+ // Verify rows with null nested keys are ignored.
+ current = row(null, null, "A_ignore");
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+
.containsExactlyInAnyOrderElementsOf(Collections.singletonList(row(0, 0,
null)));
+
+ // Verify IGNORE strategy is also applied during the first aggregation.
+ current = row(null, null, "FirstInput");
+ InternalArray result = (InternalArray) agg.agg(null,
singletonArray(current));
+ assertThat(unnest(result, elementGetter)).isEmpty();
+ }
+
+ @Test
+ public void
testFieldNestedPartialUpdateAggWithNestedKeyNullUseThrowErrorStrategy() {
+ DataType elementRowType =
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "k", DataTypes.INT()),
+ DataTypes.FIELD(1, "v1", DataTypes.INT()),
+ DataTypes.FIELD(2, "v2", DataTypes.STRING()));
+ FieldNestedPartialUpdateAgg agg =
+ new FieldNestedPartialUpdateAgg(
+ FieldNestedPartialUpdateAggFactory.NAME,
+ DataTypes.ARRAY(
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "k",
DataTypes.INT()),
+ DataTypes.FIELD(1, "v1",
DataTypes.INT()),
+ DataTypes.FIELD(2, "v2",
DataTypes.STRING()))),
+ Collections.singletonList("k"),
+ CoreOptions.NestedKeyNullStrategy.ERROR);
+
+ InternalArray accumulator;
+ InternalArray.ElementGetter elementGetter =
+ InternalArray.createElementGetter(elementRowType);
+
+ InternalRow current = row(0, 0, null);
+ accumulator = (InternalArray) agg.agg(null, singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+
.containsExactlyInAnyOrderElementsOf(Collections.singletonList(current));
+
+ // Verify ERROR strategy rejects rows with null nested keys.
+ assertThatThrownBy(() -> agg.agg(accumulator, singletonArray(row(null,
0, "A", 2))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Nested key contains null values. Primary key fields
must not be null.");
+
+ // Verify ERROR strategy is also applied during the first aggregation.
+ assertThatThrownBy(() -> agg.agg(null, singletonArray(row(null, null,
"FirstInput"))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Nested key contains null values. Primary key fields
must not be null.");
}
private Map<Object, Object> toMap(Object... kvs) {
