diff --git a/docs/docs/primary-key-table/merge-engine/aggregation.mdx b/docs/docs/primary-key-table/merge-engine/aggregation.mdx index 96e6e4f3f076..1f3cab934052 100644 --- a/docs/docs/primary-key-table/merge-engine/aggregation.mdx +++ b/docs/docs/primary-key-table/merge-engine/aggregation.mdx @@ -311,6 +311,12 @@ public static class BitmapContainsUDF extends ScalarFunction { Use `fields..nested-key=pk0,pk1,...` to specify the primary keys of the nested table. If no keys, row will be appended to array\. + Use `fields..nested-key-null-strategy=` 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..nested-sequence-field=seq0,seq1,...` to control the update sequence of a nested table, you must configure `fields..nested-key` when using it. Use `fields..count-limit=` to specify the maximum number of rows in the nested table. When no nested-key, it will select data diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index e3b6b01952cc..8c7805723e44 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/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 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 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; + } + } } diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java index 03c4c78500a0..78e473710776 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldNestedUpdateAgg.java +++ b/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,7 +33,7 @@ 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; @@ -40,8 +41,6 @@ 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 nestedKey, int countLimit) { - this(name, dataType, nestedKey, Collections.emptyList(), countLimit); - } - public FieldNestedUpdateAgg( String name, ArrayType dataType, List nestedKey, + CoreOptions.NestedKeyNullStrategy nestedKeyNullStrategy, List 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()))); + } + } } diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldNestedUpdateAggFactory.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldNestedUpdateAggFactory.java index ca0d9614986a..25228d4a2c55 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldNestedUpdateAggFactory.java +++ b/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' 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 nestedKey, - List nestedSequenceField, - int countLimit) { - if (nestedKey == null) { - nestedKey = Collections.emptyList(); - } + private void checkOptionDependencies(CoreOptions options, String field) { + List 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' 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..nested-key-null-strategy' requires " + + "'fields..nested-key' to be configured."); - return new FieldNestedUpdateAgg( - identifier(), arrayType, nestedKey, nestedSequenceField, countLimit); + checkArgument( + options.fieldNestedUpdateAggNestedSequenceField(field).isEmpty() + || !nestedKey.isEmpty(), + "Option 'fields..nested-sequence-field' requires " + + "'fields..nested-key' to be configured."); } } 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 c791fd149cdb..77cc470cfb50 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 @@ -734,6 +734,8 @@ public void testFieldNestedUpdateAgg() { DataTypes.FIELD(1, "k1", DataTypes.INT()), DataTypes.FIELD(2, "v", DataTypes.STRING()))), Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.MERGE, + Collections.emptyList(), Integer.MAX_VALUE); InternalArray accumulator; @@ -773,6 +775,8 @@ public void testFieldNestedAppendAgg() { FieldNestedUpdateAggFactory.NAME, DataTypes.ARRAY(elementRowType), Collections.emptyList(), + CoreOptions.NestedKeyNullStrategy.MERGE, + Collections.emptyList(), Integer.MAX_VALUE); InternalArray accumulator = null; @@ -795,6 +799,145 @@ public void testFieldNestedAppendAgg() { .containsExactlyInAnyOrderElementsOf(Collections.singletonList(row(0, 1, "B"))); } + @Test + public void testFieldNestedUpdateAggFactoryWithSequenceFieldPrerequisite() { + DataType elementRowType = + DataTypes.ROW( + DataTypes.FIELD(0, "k0", DataTypes.INT()), + DataTypes.FIELD(1, "k1", DataTypes.INT()), + DataTypes.FIELD(2, "v", DataTypes.STRING()), + DataTypes.FIELD(3, "seq", DataTypes.INT())); + + org.assertj.core.api.Assertions.assertThatThrownBy( + () -> + new FieldNestedUpdateAggFactory() + .create( + DataTypes.ARRAY(elementRowType), + CoreOptions.fromMap( + ImmutableMap.of( + "fields.fieldName.nested-sequence-field", + "seq")), + "fieldName")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Option 'fields..nested-sequence-field' requires " + + "'fields..nested-key' to be configured."); + + FieldNestedUpdateAgg seqAgg = + new FieldNestedUpdateAggFactory() + .create( + DataTypes.ARRAY(elementRowType), + CoreOptions.fromMap( + ImmutableMap.of( + "fields.fieldName.nested-key", + "k0,k1", + "fields.fieldName.nested-sequence-field", + "seq")), + "fieldName"); + + InternalArray.ElementGetter elementGetter = + InternalArray.createElementGetter(elementRowType); + + InternalArray accumulator = null; + accumulator = (InternalArray) seqAgg.agg(accumulator, singletonArray(row(0, 1, "A", 1))); + accumulator = (InternalArray) seqAgg.agg(accumulator, singletonArray(row(0, 1, "B", 2))); + + assertThat(unnest(accumulator, elementGetter)).containsExactly(row(0, 1, "B", 2)); + + accumulator = + (InternalArray) seqAgg.agg(accumulator, singletonArray(row(0, 1, "b_Late", 1))); + + assertThat(unnest(accumulator, elementGetter)).containsExactly(row(0, 1, "B", 2)); + } + + @Test + public void testFieldNestedUpdateAggFactoryWithNestedKeyNullStrategyPrerequisite() { + DataType elementRowType = + DataTypes.ROW( + DataTypes.FIELD(0, "k0", DataTypes.INT()), + DataTypes.FIELD(1, "k1", DataTypes.INT()), + DataTypes.FIELD(2, "v", DataTypes.STRING()), + DataTypes.FIELD(3, "seq", DataTypes.INT())); + + org.assertj.core.api.Assertions.assertThatThrownBy( + () -> + new FieldNestedUpdateAggFactory() + .create( + DataTypes.ARRAY(elementRowType), + CoreOptions.fromMap( + ImmutableMap.of( + "fields.fieldName.nested-key-null-strategy", + "merge")), + "fieldName")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Option 'fields..nested-key-null-strategy' requires " + + "'fields..nested-key' to be configured."); + + InternalArray.ElementGetter elementGetter = + InternalArray.createElementGetter(elementRowType); + + // verify merge behavior + FieldNestedUpdateAgg mergeAgg = + new FieldNestedUpdateAggFactory() + .create( + DataTypes.ARRAY(elementRowType), + CoreOptions.fromMap( + ImmutableMap.of( + "fields.fieldName.nested-key", + "k0,k1", + "fields.fieldName.nested-key-null-strategy", + "merge")), + "fieldName"); + + InternalArray mergeAccumulator = null; + mergeAccumulator = + (InternalArray) + mergeAgg.agg(mergeAccumulator, singletonArray(row(0, null, "A", 1))); + + assertThat(unnest(mergeAccumulator, elementGetter)).containsExactly(row(0, null, "A", 1)); + + // verify ignore behavior + FieldNestedUpdateAgg ignoreAgg = + new FieldNestedUpdateAggFactory() + .create( + DataTypes.ARRAY(elementRowType), + CoreOptions.fromMap( + ImmutableMap.of( + "fields.fieldName.nested-key", + "k0,k1", + "fields.fieldName.nested-key-null-strategy", + "ignore")), + "fieldName"); + + InternalArray ignoreAccumulator = null; + ignoreAccumulator = + (InternalArray) ignoreAgg.agg(ignoreAccumulator, singletonArray(row(0, 1, "A", 1))); + ignoreAccumulator = + (InternalArray) + ignoreAgg.agg(ignoreAccumulator, singletonArray(row(0, null, "B", 2))); + + assertThat(unnest(ignoreAccumulator, elementGetter)).containsExactly(row(0, 1, "A", 1)); + + // verify error behavior + FieldNestedUpdateAgg errorAgg = + new FieldNestedUpdateAggFactory() + .create( + DataTypes.ARRAY(elementRowType), + CoreOptions.fromMap( + ImmutableMap.of( + "fields.fieldName.nested-key", + "k0,k1", + "fields.fieldName.nested-key-null-strategy", + "error")), + "fieldName"); + + assertThatThrownBy(() -> errorAgg.agg(null, singletonArray(row(0, null, "B", 2)))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Nested key contains null values. Primary key fields must not be null."); + } + @Test public void testFieldNestedAppendAggWithCountLimit() { DataType elementRowType = @@ -807,6 +950,8 @@ public void testFieldNestedAppendAggWithCountLimit() { FieldNestedUpdateAggFactory.NAME, DataTypes.ARRAY(elementRowType), Collections.emptyList(), + CoreOptions.NestedKeyNullStrategy.MERGE, + Collections.emptyList(), 2); InternalArray accumulator = null; @@ -842,6 +987,8 @@ public void testFieldNestedAppendAggWithCountLimitOnFirstInputArray() { FieldNestedUpdateAggFactory.NAME, DataTypes.ARRAY(elementRowType), Collections.emptyList(), + CoreOptions.NestedKeyNullStrategy.MERGE, + Collections.emptyList(), 2); InternalArray.ElementGetter elementGetter = @@ -867,6 +1014,8 @@ public void testFieldNestedUpdateAggWithCountLimitUpdatesExistingKeyAtLimitWitho FieldNestedUpdateAggFactory.NAME, DataTypes.ARRAY(elementRowType), Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.MERGE, + Collections.emptyList(), 2); InternalArray accumulator = null; @@ -900,6 +1049,8 @@ public void testFieldNestedUpdateAggWithCountLimitOnFirstInputArrayWithoutSequen FieldNestedUpdateAggFactory.NAME, DataTypes.ARRAY(elementRowType), Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.MERGE, + Collections.emptyList(), 2); InternalArray.ElementGetter elementGetter = @@ -932,6 +1083,7 @@ public void testFieldNestedUpdateAggWithSequenceField() { FieldNestedUpdateAggFactory.NAME, DataTypes.ARRAY(elementRowType), Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.MERGE, Collections.singletonList("seq"), Integer.MAX_VALUE); @@ -983,6 +1135,7 @@ public void testFieldNestedUpdateAggWithMultipleSequenceFields() { FieldNestedUpdateAggFactory.NAME, DataTypes.ARRAY(elementRowType), Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.MERGE, Arrays.asList("seq", "ts"), Integer.MAX_VALUE); @@ -1074,27 +1227,6 @@ public void testFieldNestedUpdateAggWithMultipleSequenceFields() { row(1, 0, "A_updated_by_ts", 1, ts3), row(1, 1, "C", 1, ts2))); } - @Test - public void testFieldNestedUpdateAggWithSequenceFieldWithoutNestedKey() { - DataType elementRowType = - DataTypes.ROW( - DataTypes.FIELD(0, "k0", DataTypes.INT()), - DataTypes.FIELD(1, "k1", DataTypes.INT()), - DataTypes.FIELD(2, "v", DataTypes.STRING()), - DataTypes.FIELD(3, "seq", DataTypes.INT())); - - org.assertj.core.api.Assertions.assertThatThrownBy( - () -> - new FieldNestedUpdateAgg( - FieldNestedUpdateAggFactory.NAME, - DataTypes.ARRAY(elementRowType), - Collections.emptyList(), - Collections.singletonList("seq"), - Integer.MAX_VALUE)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("nested-sequence-field requires nested-key to be set."); - } - @Test public void testFieldNestedUpdateAggWithCountLimitWithSequenceFieldWithoutNestedKey() { DataType elementRowType = @@ -1107,14 +1239,45 @@ public void testFieldNestedUpdateAggWithCountLimitWithSequenceFieldWithoutNested // Verify that the same precondition check applies even when a count limit is specified org.assertj.core.api.Assertions.assertThatThrownBy( () -> - new FieldNestedUpdateAgg( - FieldNestedUpdateAggFactory.NAME, - DataTypes.ARRAY(elementRowType), - Collections.emptyList(), - Collections.singletonList("seq"), - 2)) + new FieldNestedUpdateAggFactory() + .create( + DataTypes.ARRAY(elementRowType), + CoreOptions.fromMap( + ImmutableMap.of( + "fields.fieldName.nested-sequence-field", + "seq", + "fields.fieldName.count-limit", + "2")), + "fieldName")) .isInstanceOf(IllegalArgumentException.class) - .hasMessage("nested-sequence-field requires nested-key to be set."); + .hasMessage( + "Option 'fields..nested-sequence-field' requires " + + "'fields..nested-key' to be configured."); + + FieldNestedUpdateAgg agg = + new FieldNestedUpdateAggFactory() + .create( + DataTypes.ARRAY(elementRowType), + CoreOptions.fromMap( + ImmutableMap.of( + "fields.fieldName.nested-key", "k0,k1", + "fields.fieldName.nested-sequence-field", "seq", + "fields.fieldName.count-limit", "2")), + "fieldName"); + + InternalArray.ElementGetter elementGetter = + InternalArray.createElementGetter(elementRowType); + + InternalArray accumulator = null; + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(row(0, 1, "A", 1))); + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(row(0, 2, "B", 2))); + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(row(0, 3, "C", 3))); + accumulator = + (InternalArray) agg.agg(accumulator, singletonArray(row(0, 1, "A_Update", 4))); + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(row(0, 2, "B_Late", 1))); + + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrder(row(0, 1, "A_Update", 4), row(0, 2, "B", 2)); } @Test @@ -1131,6 +1294,7 @@ public void testFieldNestedUpdateAggWithCountLimitWithSequenceField() { FieldNestedUpdateAggFactory.NAME, DataTypes.ARRAY(elementRowType), Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.MERGE, Collections.singletonList("seq"), 2); // Enforce count limit = 2 @@ -1179,6 +1343,7 @@ public void testFieldNestedUpdateAggWithCountLimitUpdatesExistingKeyAtLimit() { FieldNestedUpdateAggFactory.NAME, DataTypes.ARRAY(elementRowType), Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.MERGE, Collections.singletonList("seq"), 2); @@ -1215,6 +1380,7 @@ public void testFieldNestedUpdateAggWithCountLimitOnFirstInputArrayWithSequence( FieldNestedUpdateAggFactory.NAME, DataTypes.ARRAY(elementRowType), Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.MERGE, Collections.singletonList("seq"), 2); @@ -1235,6 +1401,462 @@ public void testFieldNestedUpdateAggWithCountLimitOnFirstInputArrayWithSequence( Arrays.asList(row(0, 1, "B_updated", 4), row(1, 2, "C", 3))); } + @Test + public void testFieldNestedUpdateAggWhenNestedKeyNullUseMergeStrategy() { + DataType elementRowType = + DataTypes.ROW( + DataTypes.FIELD(0, "k0", DataTypes.INT()), + DataTypes.FIELD(1, "k1", DataTypes.INT()), + DataTypes.FIELD(2, "v", DataTypes.STRING()), + DataTypes.FIELD(3, "seq", DataTypes.INT())); + + FieldNestedUpdateAgg agg = + new FieldNestedUpdateAgg( + FieldNestedUpdateAggFactory.NAME, + DataTypes.ARRAY(elementRowType), + Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.MERGE, + Collections.emptyList(), + Integer.MAX_VALUE); + + InternalArray.ElementGetter elementGetter = + InternalArray.createElementGetter(elementRowType); + + // Empty accumulator. + InternalArray accumulatorEmpty; + InternalRow current = row(0, null, "C", 3); + + // case 1: partially null nested key (some PK fields are null) + accumulatorEmpty = (InternalArray) agg.agg(null, singletonArray(current)); + assertThat(unnest(accumulatorEmpty, elementGetter)) + .containsExactlyInAnyOrderElementsOf(Collections.singletonList(current)); + + // case 2: fully null nested key (all PK fields are null) + current = row(null, null, "D", 4); + accumulatorEmpty = (InternalArray) agg.agg(null, singletonArray(current)); + assertThat(unnest(accumulatorEmpty, elementGetter)) + .containsExactlyInAnyOrderElementsOf(Collections.singletonList(current)); + + // Non-empty accumulator. + InternalArray accumulator; + + current = row(0, 0, "A", 1); + accumulator = (InternalArray) agg.agg(null, singletonArray(current)); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf(Collections.singletonList(current)); + + current = row(0, 1, "B", 2); + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(current)); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList(row(0, 0, "A", 1), row(0, 1, "B", 2))); + + // case 1: partially null nested key (some PK fields are null) + current = row(0, null, "C", 3); + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(current)); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList(row(0, 0, "A", 1), row(0, 1, "B", 2), row(0, null, "C", 3))); + + // case 2: fully null nested key (all PK fields are null) + current = row(null, null, "D", 4); + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(current)); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList( + row(0, 0, "A", 1), + row(0, 1, "B", 2), + row(0, null, "C", 3), + row(null, null, "D", 4))); + } + + @Test + public void testFieldNestedUpdateAggWhenNestedKeyNullUseIgnoreStrategy() { + DataType elementRowType = + DataTypes.ROW( + DataTypes.FIELD(0, "k0", DataTypes.INT()), + DataTypes.FIELD(1, "k1", DataTypes.INT()), + DataTypes.FIELD(2, "v", DataTypes.STRING()), + DataTypes.FIELD(3, "seq", DataTypes.INT())); + + FieldNestedUpdateAgg agg = + new FieldNestedUpdateAgg( + FieldNestedUpdateAggFactory.NAME, + DataTypes.ARRAY(elementRowType), + Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.IGNORE, // use ignore strategy + Collections.emptyList(), + Integer.MAX_VALUE); + + InternalArray.ElementGetter elementGetter = + InternalArray.createElementGetter(elementRowType); + + // Empty accumulator. + InternalArray accumulatorEmpty; + + // case 1: partially null nested key (some PK fields are null) + accumulatorEmpty = (InternalArray) agg.agg(null, singletonArray(row(0, null, "C", 3))); + assertThat(unnest(accumulatorEmpty, elementGetter)).isEmpty(); + + // case 2: fully null nested key (all PK fields are null) + accumulatorEmpty = (InternalArray) agg.agg(null, singletonArray(row(null, null, "D", 4))); + assertThat(unnest(accumulatorEmpty, elementGetter)).isEmpty(); + + // Non-empty accumulator. + InternalArray accumulator; + + InternalRow current = row(0, 0, "A", 1); + accumulator = (InternalArray) agg.agg(null, singletonArray(current)); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf(Collections.singletonList(current)); + + current = row(0, 1, "B", 2); + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(current)); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList(row(0, 0, "A", 1), row(0, 1, "B", 2))); + + // case 1: partially null nested key (some PK fields are null) + current = row(0, null, "C", 3); + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(current)); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList(row(0, 0, "A", 1), row(0, 1, "B", 2))); + + // case 2: fully null nested key (all PK fields are null) + current = row(null, null, "D", 4); + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(current)); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList(row(0, 0, "A", 1), row(0, 1, "B", 2))); + } + + @Test + public void testFieldNestedUpdateAggWhenNestedKeyNullUseThrowErrorStrategy() { + DataType elementRowType = + DataTypes.ROW( + DataTypes.FIELD(0, "k0", DataTypes.INT()), + DataTypes.FIELD(1, "k1", DataTypes.INT()), + DataTypes.FIELD(2, "v", DataTypes.STRING()), + DataTypes.FIELD(3, "seq", DataTypes.INT())); + + FieldNestedUpdateAgg agg = + new FieldNestedUpdateAgg( + FieldNestedUpdateAggFactory.NAME, + DataTypes.ARRAY(elementRowType), + Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.ERROR, // use error strategy + Collections.emptyList(), + Integer.MAX_VALUE); + + // Empty accumulator. + // case 1: partially null nested key (some PK fields are null) + assertThatThrownBy(() -> agg.agg(null, singletonArray(row(0, null, "C", 3)))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Nested key contains null values. Primary key fields must not be null."); + + // case 2: fully null nested key (all PK fields are null) + assertThatThrownBy(() -> agg.agg(null, singletonArray(row(null, null, "D", 4)))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Nested key contains null values. Primary key fields must not be null."); + + // Non-empty accumulator. + InternalArray accumulator; + InternalArray.ElementGetter elementGetter = + InternalArray.createElementGetter(elementRowType); + + InternalRow current = row(0, 0, "A", 1); + accumulator = (InternalArray) agg.agg(null, singletonArray(current)); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf(Collections.singletonList(current)); + + current = row(0, 1, "B", 2); + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(current)); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList(row(0, 0, "A", 1), row(0, 1, "B", 2))); + + // case 1: partially null nested key (some PK fields are null) + InternalArray finalAccumulator1 = accumulator; + assertThatThrownBy(() -> agg.agg(finalAccumulator1, singletonArray(row(0, null, "C", 3)))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Nested key contains null values. Primary key fields must not be null."); + + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList(row(0, 0, "A", 1), row(0, 1, "B", 2))); + + // case 2: fully null nested key (all PK fields are null) + InternalArray finalAccumulator2 = accumulator; + assertThatThrownBy( + () -> agg.agg(finalAccumulator2, singletonArray(row(null, null, "D", 4)))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Nested key contains null values. Primary key fields must not be null."); + + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList(row(0, 0, "A", 1), row(0, 1, "B", 2))); + } + + @Test + public void testFieldNestedUpdateAggWithCountLimitWhenNestedKeyNullUseMergeStrategy() { + DataType elementRowType = + DataTypes.ROW( + DataTypes.FIELD(0, "k0", DataTypes.INT()), + DataTypes.FIELD(1, "k1", DataTypes.INT()), + DataTypes.FIELD(2, "v", DataTypes.STRING()), + DataTypes.FIELD(3, "seq", DataTypes.INT())); + + FieldNestedUpdateAgg agg = + new FieldNestedUpdateAgg( + FieldNestedUpdateAggFactory.NAME, + DataTypes.ARRAY(elementRowType), + Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.MERGE, // use merge strategy + Collections.singletonList("seq"), + 3); + + InternalArray accumulator = null; + InternalArray.ElementGetter elementGetter = + InternalArray.createElementGetter(elementRowType); + + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(row(0, 1, "B", 1))); + accumulator = + (InternalArray) agg.agg(accumulator, singletonArray(row(null, 2, "NULL_2", 2))); + accumulator = + (InternalArray) + agg.agg(accumulator, singletonArray(row(null, null, "NULL_NULL", 3))); + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(row(1, 2, "C", 5))); + + accumulator = + (InternalArray) agg.agg(accumulator, singletonArray(row(0, 1, "B_updated", 4))); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList( + row(0, 1, "B_updated", 4), + row(null, 2, "NULL_2", 2), + row(null, null, "NULL_NULL", 3))); + } + + @Test + public void testFieldNestedUpdateAggWithCountLimitWhenNestedKeyNullUseIgnoreStrategy() { + DataType elementRowType = + DataTypes.ROW( + DataTypes.FIELD(0, "k0", DataTypes.INT()), + DataTypes.FIELD(1, "k1", DataTypes.INT()), + DataTypes.FIELD(2, "v", DataTypes.STRING()), + DataTypes.FIELD(3, "seq", DataTypes.INT())); + + FieldNestedUpdateAgg agg = + new FieldNestedUpdateAgg( + FieldNestedUpdateAggFactory.NAME, + DataTypes.ARRAY(elementRowType), + Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.IGNORE, // use ignore strategy + Collections.singletonList("seq"), + 3); + + InternalArray accumulator = null; + InternalArray.ElementGetter elementGetter = + InternalArray.createElementGetter(elementRowType); + + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(row(0, 1, "B", 1))); + accumulator = + (InternalArray) agg.agg(accumulator, singletonArray(row(null, 2, "NULL_2", 2))); + accumulator = + (InternalArray) + agg.agg(accumulator, singletonArray(row(null, null, "NULL_NULL", 3))); + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(row(1, 2, "C", 3))); + + accumulator = + (InternalArray) agg.agg(accumulator, singletonArray(row(0, 1, "B_updated", 4))); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList(row(0, 1, "B_updated", 4), row(1, 2, "C", 3))); + + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(row(2, 3, "D", 5))); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList( + row(0, 1, "B_updated", 4), row(1, 2, "C", 3), row(2, 3, "D", 5))); + } + + @Test + public void testFieldNestedUpdateAggWithCountLimitWhenNestedKeyNullUseThrowErrorStrategy() { + DataType elementRowType = + DataTypes.ROW( + DataTypes.FIELD(0, "k0", DataTypes.INT()), + DataTypes.FIELD(1, "k1", DataTypes.INT()), + DataTypes.FIELD(2, "v", DataTypes.STRING()), + DataTypes.FIELD(3, "seq", DataTypes.INT())); + + FieldNestedUpdateAgg agg = + new FieldNestedUpdateAgg( + FieldNestedUpdateAggFactory.NAME, + DataTypes.ARRAY(elementRowType), + Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.ERROR, // use error strategy + Collections.singletonList("seq"), + 3); + + InternalArray accumulator = null; + InternalArray.ElementGetter elementGetter = + InternalArray.createElementGetter(elementRowType); + + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(row(0, 1, "B", 1))); + + InternalArray finalAccumulator = accumulator; + assertThatThrownBy( + () -> agg.agg(finalAccumulator, singletonArray(row(null, 2, "NULL_2", 2)))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Nested key contains null values. Primary key fields must not be null."); + + assertThatThrownBy( + () -> + agg.agg( + finalAccumulator, + singletonArray(row(null, null, "NULL_NULL", 3)))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Nested key contains null values. Primary key fields must not be null."); + + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(row(1, 2, "C", 3))); + + accumulator = + (InternalArray) agg.agg(accumulator, singletonArray(row(0, 1, "B_updated", 4))); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList(row(0, 1, "B_updated", 4), row(1, 2, "C", 3))); + + accumulator = (InternalArray) agg.agg(accumulator, singletonArray(row(2, 3, "D", 5))); + assertThat(unnest(accumulator, elementGetter)) + .containsExactlyInAnyOrderElementsOf( + Arrays.asList( + row(0, 1, "B_updated", 4), row(1, 2, "C", 3), row(2, 3, "D", 5))); + } + + @Test + public void testFieldNestedUpdateAggRetractAppliesNestedKeyNullStrategyToAccumulator() { + DataType elementRowType = + DataTypes.ROW( + DataTypes.FIELD(0, "k0", DataTypes.INT()), + DataTypes.FIELD(1, "k1", DataTypes.INT()), + DataTypes.FIELD(2, "v", DataTypes.STRING())); + + // Build an accumulator containing a null nested key. + FieldNestedUpdateAgg mergeAgg = + new FieldNestedUpdateAgg( + FieldNestedUpdateAggFactory.NAME, + DataTypes.ARRAY(elementRowType), + Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.MERGE, + Collections.emptyList(), + Integer.MAX_VALUE); + + InternalArray accumulator = null; + accumulator = (InternalArray) mergeAgg.agg(accumulator, singletonArray(row(0, null, "A"))); + accumulator = (InternalArray) mergeAgg.agg(accumulator, singletonArray(row(1, 0, "B"))); + accumulator = (InternalArray) mergeAgg.agg(accumulator, singletonArray(row(1, 1, "C"))); + + // IGNORE behavior: rows with null nested keys in the accumulator should be ignored. + FieldNestedUpdateAgg ignoreAgg = + new FieldNestedUpdateAgg( + FieldNestedUpdateAggFactory.NAME, + DataTypes.ARRAY(elementRowType), + Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.IGNORE, + Collections.emptyList(), + Integer.MAX_VALUE); + + InternalArray result = + (InternalArray) ignoreAgg.retract(accumulator, singletonArray(row(1, 0, "B"))); + + InternalArray.ElementGetter elementGetter = + InternalArray.createElementGetter(elementRowType); + + assertThat(unnest(result, elementGetter)).containsExactly(row(1, 1, "C")); + + // ERROR behavior: rows with null nested keys in the accumulator should throw exception. + FieldNestedUpdateAgg errorAgg = + new FieldNestedUpdateAgg( + FieldNestedUpdateAggFactory.NAME, + DataTypes.ARRAY(elementRowType), + Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.ERROR, + Collections.emptyList(), + Integer.MAX_VALUE); + + final InternalArray acc = accumulator; + assertThatThrownBy(() -> errorAgg.retract(acc, singletonArray(row(1, 0, "B")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Nested key contains null values. Primary key fields must not be null."); + } + + @Test + public void testFieldNestedUpdateAggRetractAppliesNestedKeyNullStrategyToRetractInput() { + DataType elementRowType = + DataTypes.ROW( + DataTypes.FIELD(0, "k0", DataTypes.INT()), + DataTypes.FIELD(1, "k1", DataTypes.INT()), + DataTypes.FIELD(2, "v", DataTypes.STRING())); + + // Build an accumulator without null nested keys. + FieldNestedUpdateAgg mergeAgg = + new FieldNestedUpdateAgg( + FieldNestedUpdateAggFactory.NAME, + DataTypes.ARRAY(elementRowType), + Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.MERGE, + Collections.emptyList(), + Integer.MAX_VALUE); + + InternalArray accumulator = null; + accumulator = (InternalArray) mergeAgg.agg(accumulator, singletonArray(row(0, 0, "A"))); + accumulator = (InternalArray) mergeAgg.agg(accumulator, singletonArray(row(1, 1, "B"))); + + InternalArray.ElementGetter elementGetter = + InternalArray.createElementGetter(elementRowType); + + // IGNORE behavior: rows with null nested keys in the retract input should be ignored. + FieldNestedUpdateAgg ignoreAgg = + new FieldNestedUpdateAgg( + FieldNestedUpdateAggFactory.NAME, + DataTypes.ARRAY(elementRowType), + Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.IGNORE, + Collections.emptyList(), + Integer.MAX_VALUE); + + InternalArray result = + (InternalArray) ignoreAgg.retract(accumulator, singletonArray(row(0, null, "X"))); + + assertThat(unnest(result, elementGetter)) + .containsExactlyInAnyOrder(row(0, 0, "A"), row(1, 1, "B")); + + // ERROR behavior: rows with null nested keys in the retract input should throw exception. + FieldNestedUpdateAgg errorAgg = + new FieldNestedUpdateAgg( + FieldNestedUpdateAggFactory.NAME, + DataTypes.ARRAY(elementRowType), + Arrays.asList("k0", "k1"), + CoreOptions.NestedKeyNullStrategy.ERROR, + Collections.emptyList(), + Integer.MAX_VALUE); + + final InternalArray acc = accumulator; + + assertThatThrownBy(() -> errorAgg.retract(acc, singletonArray(row(0, null, "X")))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Nested key contains null values. Primary key fields must not be null."); + } + private List unnest(InternalArray array, InternalArray.ElementGetter elementGetter) { return IntStream.range(0, array.size()) .mapToObj(i -> elementGetter.getElementOrNull(array, i))