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 1fc7506cd5 [core] Support nested-key-null-strategy in
FieldNestedUpdateAgg operator (#8374)
1fc7506cd5 is described below
commit 1fc7506cd51d5b6adda2b02ea7151f1221d37506
Author: AuroraVoyage <[email protected]>
AuthorDate: Fri Jul 3 22:25:49 2026 +0800
[core] Support nested-key-null-strategy in FieldNestedUpdateAgg operator
(#8374)
Add `fields.{fieldName}.nested-key-null-strategy`
configuration for `nested_update` function
[Feature] Support nested-key null handling strategies in
FieldNestedUpdateAgg operator to control behavior when nested-key does
not satisfy primary key semantics
---
.../primary-key-table/merge-engine/aggregation.mdx | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 42 ++
.../compact/aggregate/FieldNestedUpdateAgg.java | 57 +-
.../factory/FieldNestedUpdateAggFactory.java | 49 +-
.../compact/aggregate/FieldAggregatorTest.java | 678 ++++++++++++++++++++-
5 files changed, 770 insertions(+), 62 deletions(-)
diff --git a/docs/docs/primary-key-table/merge-engine/aggregation.mdx
b/docs/docs/primary-key-table/merge-engine/aggregation.mdx
index 96e6e4f3f0..1f3cab9340 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.<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
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 ce15c207ec..23721c1f44 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";
@@ -2944,6 +2946,13 @@ public class CoreOptions implements Serializable {
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(
@@ -5033,4 +5042,37 @@ public class CoreOptions implements Serializable {
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 03c4c78500..78e4737107 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 org.apache.paimon.types.RowType;
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 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 class FieldNestedUpdateAgg extends FieldAggregator {
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 class FieldNestedUpdateAgg extends FieldAggregator
{
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 @@ public class FieldNestedUpdateAgg extends FieldAggregator {
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 @@ public class FieldNestedUpdateAgg 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/FieldNestedUpdateAggFactory.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldNestedUpdateAggFactory.java
index ca0d961498..25228d4a2c 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.ArrayType;
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 class FieldNestedUpdateAggFactory implements
FieldAggregatorFactory {
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.");
}
}
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 c791fd149c..77cc470cfb 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 class FieldAggregatorTest {
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 class FieldAggregatorTest {
FieldNestedUpdateAggFactory.NAME,
DataTypes.ARRAY(elementRowType),
Collections.emptyList(),
+ CoreOptions.NestedKeyNullStrategy.MERGE,
+ Collections.emptyList(),
Integer.MAX_VALUE);
InternalArray accumulator = null;
@@ -795,6 +799,145 @@ public class FieldAggregatorTest {
.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.<field-name>.nested-sequence-field'
requires "
+ + "'fields.<field-name>.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.<field-name>.nested-key-null-strategy'
requires "
+ + "'fields.<field-name>.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 class FieldAggregatorTest {
FieldNestedUpdateAggFactory.NAME,
DataTypes.ARRAY(elementRowType),
Collections.emptyList(),
+ CoreOptions.NestedKeyNullStrategy.MERGE,
+ Collections.emptyList(),
2);
InternalArray accumulator = null;
@@ -842,6 +987,8 @@ public class FieldAggregatorTest {
FieldNestedUpdateAggFactory.NAME,
DataTypes.ARRAY(elementRowType),
Collections.emptyList(),
+ CoreOptions.NestedKeyNullStrategy.MERGE,
+ Collections.emptyList(),
2);
InternalArray.ElementGetter elementGetter =
@@ -867,6 +1014,8 @@ public class FieldAggregatorTest {
FieldNestedUpdateAggFactory.NAME,
DataTypes.ARRAY(elementRowType),
Arrays.asList("k0", "k1"),
+ CoreOptions.NestedKeyNullStrategy.MERGE,
+ Collections.emptyList(),
2);
InternalArray accumulator = null;
@@ -900,6 +1049,8 @@ public class FieldAggregatorTest {
FieldNestedUpdateAggFactory.NAME,
DataTypes.ARRAY(elementRowType),
Arrays.asList("k0", "k1"),
+ CoreOptions.NestedKeyNullStrategy.MERGE,
+ Collections.emptyList(),
2);
InternalArray.ElementGetter elementGetter =
@@ -932,6 +1083,7 @@ public class FieldAggregatorTest {
FieldNestedUpdateAggFactory.NAME,
DataTypes.ARRAY(elementRowType),
Arrays.asList("k0", "k1"),
+ CoreOptions.NestedKeyNullStrategy.MERGE,
Collections.singletonList("seq"),
Integer.MAX_VALUE);
@@ -983,6 +1135,7 @@ public class FieldAggregatorTest {
FieldNestedUpdateAggFactory.NAME,
DataTypes.ARRAY(elementRowType),
Arrays.asList("k0", "k1"),
+ CoreOptions.NestedKeyNullStrategy.MERGE,
Arrays.asList("seq", "ts"),
Integer.MAX_VALUE);
@@ -1074,27 +1227,6 @@ public class FieldAggregatorTest {
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 class FieldAggregatorTest {
// 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.<field-name>.nested-sequence-field'
requires "
+ + "'fields.<field-name>.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 class FieldAggregatorTest {
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 class FieldAggregatorTest {
FieldNestedUpdateAggFactory.NAME,
DataTypes.ARRAY(elementRowType),
Arrays.asList("k0", "k1"),
+ CoreOptions.NestedKeyNullStrategy.MERGE,
Collections.singletonList("seq"),
2);
@@ -1215,6 +1380,7 @@ public class FieldAggregatorTest {
FieldNestedUpdateAggFactory.NAME,
DataTypes.ARRAY(elementRowType),
Arrays.asList("k0", "k1"),
+ CoreOptions.NestedKeyNullStrategy.MERGE,
Collections.singletonList("seq"),
2);
@@ -1235,6 +1401,462 @@ public class FieldAggregatorTest {
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<Object> unnest(InternalArray array,
InternalArray.ElementGetter elementGetter) {
return IntStream.range(0, array.size())
.mapToObj(i -> elementGetter.getElementOrNull(array, i))
