[core] Support nested-key-null-strategy in FieldNestedUpdateAgg operator

docs/docs/primary-key-table/merge-engine/aggregation.mdx

@@ -311,6 +311,12 @@ public static class BitmapContainsUDF extends ScalarFunction {
 
   Use `fields.<field-name>.nested-key=pk0,pk1,...` to specify the primary keys of the nested table. If no keys, row will be appended to array\<row>.
 
+  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.
+
   Use `fields.<field-name>.nested-sequence-field=seq0,seq1,...` to control the update sequence of a nested table, you must configure `fields.<field-name>.nested-key` when using it.
 
   Use `fields.<field-name>.count-limit=<Integer>` to specify the maximum number of rows in the nested table. When no nested-key, it will select data

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -80,6 +80,8 @@ public class CoreOptions implements Serializable {
 
     public static final String NESTED_KEY = "nested-key";
 
+    public static final String NESTED_KEY_NULL_STRATEGY = "nested-key-null-strategy";
+
     public static final String NESTED_SEQUENCE_FIELD = "nested-sequence-field";
 
     public static final String COUNT_LIMIT = "count-limit";
@@ -2943,6 +2945,13 @@ public List<String> fieldNestedUpdateAggNestedKey(String fieldName) {
         return Arrays.stream(keyString.split(",")).map(String::trim).collect(Collectors.toList());
     }
 
+    public NestedKeyNullStrategy fieldNestedUpdateAggNestedKeyNullStrategy(String fieldName) {
+        return options.get(
+                key(FIELDS_PREFIX + "." + fieldName + "." + NESTED_KEY_NULL_STRATEGY)
+                        .enumType(NestedKeyNullStrategy.class)
+                        .defaultValue(NestedKeyNullStrategy.MERGE));
+    }
+
     public List<String> fieldNestedUpdateAggNestedSequenceField(String fieldName) {
         String keyString =
                 options.get(
@@ -5046,4 +5055,37 @@ public InlineElement getDescription() {
             return text(description);
         }
     }
+
+    /** Strategy for handling rows whose nested-key contains null values. */
+    public enum NestedKeyNullStrategy implements DescribedEnum {
+        MERGE(
+                "merge",
+                "Merge rows even if the nested-key contains null values, without enforcing primary key semantics."),
+
+        IGNORE(
+                "ignore",
+                "Ignore rows whose nested-key contains null values because they do not satisfy primary key semantics."),
+
+        ERROR(
+                "error",
+                "Throw an exception if the nested-key contains null values, because primary key fields must not be null.");
+
+        private final String value;
+        private final String description;
+
+        NestedKeyNullStrategy(String value, String description) {
+            this.value = value;
+            this.description = description;
+        }
+
+        @Override
+        public InlineElement getDescription() {
+            return text(description);
+        }
+
+        @Override
+        public String toString() {
+            return value;
+        }
+    }
 }

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.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.codegen.RecordComparator;
 import org.apache.paimon.codegen.RecordEqualiser;
@@ -32,16 +33,14 @@
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
-import java.util.Collections;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 import static org.apache.paimon.codegen.CodeGenUtils.newProjection;
 import static org.apache.paimon.codegen.CodeGenUtils.newRecordComparator;
 import static org.apache.paimon.codegen.CodeGenUtils.newRecordEqualiser;
-import static org.apache.paimon.options.ConfigOptions.key;
-import static org.apache.paimon.utils.Preconditions.checkArgument;
 import static org.apache.paimon.utils.Preconditions.checkNotNull;
 
 /**
@@ -61,17 +60,14 @@ public class FieldNestedUpdateAgg extends FieldAggregator {
     @Nullable private final RecordComparator sequenceComparator;
     private final boolean hasSequenceField;
 
+    private final CoreOptions.NestedKeyNullStrategy nestedKeyNullStrategy;
     private final int countLimit;
 
-    public FieldNestedUpdateAgg(
-            String name, ArrayType dataType, List<String> nestedKey, int countLimit) {
-        this(name, dataType, nestedKey, Collections.emptyList(), countLimit);
-    }
-
     public FieldNestedUpdateAgg(
             String name,
             ArrayType dataType,
             List<String> nestedKey,
+            CoreOptions.NestedKeyNullStrategy nestedKeyNullStrategy,
             List<String> nestedSequenceField,
             int countLimit) {
         super(name, dataType);
@@ -85,12 +81,11 @@ public FieldNestedUpdateAgg(
             this.elementEqualiser = null;
         }
 
+        this.nestedKeyNullStrategy = nestedKeyNullStrategy;
+
         // If nestedSequenceField is set, we need to compare sequence fields to determine
         // whether to update. Only update when the new sequence is greater than the old one.
         if (!nestedSequenceField.isEmpty()) {
-            checkArgument(
-                    this.keyProjection != null,
-                    "nested-sequence-field requires nested-key to be set.");
             this.sequenceProjection = newProjection(nestedType, nestedSequenceField);
             this.hasSequenceField = true;
 
@@ -179,14 +174,22 @@ public Object retract(Object accumulator, Object retractField) {
                     continue;
                 }
                 InternalRow row = acc.getRow(i, nestedFields);
-                map.put(keyProjection.apply(row).copy(), row);
+                BinaryRow key = keyProjection.apply(row).copy();
+                if (!applyNestedKeyNullStrategy(key)) {
+                    continue;
+                }
+                map.put(key, row);
             }
 
             for (int i = 0; i < retract.size(); i++) {
                 if (retract.isNullAt(i)) {
                     continue;
                 }
-                map.remove(keyProjection.apply(retract.getRow(i, nestedFields)));
+                BinaryRow key = keyProjection.apply(retract.getRow(i, nestedFields)).copy();
+                if (!applyNestedKeyNullStrategy(key)) {
+                    continue;
+                }
+                map.remove(key);
             }
 
             return new GenericArray(new ArrayList<>(map.values()).toArray());
@@ -243,6 +246,11 @@ private void addNestedRows(
 
             InternalRow row = array.getRow(i, nestedFields);
             BinaryRow key = keyProjection.apply(row).copy();
+
+            if (!applyNestedKeyNullStrategy(key)) {
+                continue;
+            }
+
             InternalRow existing = rows.get(key);
             if (existing != null) {
                 if (!hasSequenceField || compareSequence(row, existing) >= 0) {
@@ -253,4 +261,27 @@ private void addNestedRows(
             }
         }
     }
+
+    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())));
+        }
+    }
 }

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldNestedUpdateAggFactory.java

@@ -24,9 +24,10 @@
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.RowType;
 
-import java.util.Collections;
 import java.util.List;
 
+import static org.apache.paimon.CoreOptions.FIELDS_PREFIX;
+import static org.apache.paimon.CoreOptions.NESTED_KEY_NULL_STRATEGY;
 import static org.apache.paimon.utils.Preconditions.checkArgument;
 
 /** Factory for #{@link FieldNestedUpdateAgg}. */
@@ -36,9 +37,19 @@ public class FieldNestedUpdateAggFactory implements FieldAggregatorFactory {
 
     @Override
     public FieldNestedUpdateAgg create(DataType fieldType, CoreOptions options, String field) {
-        return createFieldNestedUpdateAgg(
-                fieldType,
+        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);
+
+        checkOptionDependencies(options, field);
+
+        return new FieldNestedUpdateAgg(
+                identifier(),
+                arrayType,
                 options.fieldNestedUpdateAggNestedKey(field),
+                options.fieldNestedUpdateAggNestedKeyNullStrategy(field),
                 options.fieldNestedUpdateAggNestedSequenceField(field),
                 options.fieldNestedUpdateAggCountLimit(field));
     }
@@ -48,26 +59,22 @@ public String identifier() {
         return NAME;
     }
 
-    private FieldNestedUpdateAgg createFieldNestedUpdateAgg(
-            DataType fieldType,
-            List<String> nestedKey,
-            List<String> nestedSequenceField,
-            int countLimit) {
-        if (nestedKey == null) {
-            nestedKey = Collections.emptyList();
-        }
+    private void checkOptionDependencies(CoreOptions options, String field) {
+        List<String> nestedKey = options.fieldNestedUpdateAggNestedKey(field);
 
-        if (nestedSequenceField == null) {
-            nestedSequenceField = Collections.emptyList();
-        }
+        boolean strategyConfigured =
+                options.toConfiguration()
+                        .containsKey(FIELDS_PREFIX + "." + field + "." + NESTED_KEY_NULL_STRATEGY);
 
-        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);
+        checkArgument(
+                !strategyConfigured || !nestedKey.isEmpty(),
+                "Option 'fields.<field-name>.nested-key-null-strategy' requires "
+                        + "'fields.<field-name>.nested-key' to be configured.");
 
-        return new FieldNestedUpdateAgg(
-                identifier(), arrayType, nestedKey, nestedSequenceField, countLimit);
+        checkArgument(
+                options.fieldNestedUpdateAggNestedSequenceField(field).isEmpty()
+                        || !nestedKey.isEmpty(),
+                "Option 'fields.<field-name>.nested-sequence-field' requires "
+                        + "'fields.<field-name>.nested-key' to be configured.");
     }
 }