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 b93b6b5bfb [core] Support nested sequence fields in
FieldNestedUpdateAgg operator (#8048)
b93b6b5bfb is described below
commit b93b6b5bfb6768fdfc1720cff5afa25c5643bf34
Author: AuroraVoyage <[email protected]>
AuthorDate: Sun May 31 21:21:47 2026 +0800
[core] Support nested sequence fields in FieldNestedUpdateAgg operator
(#8048)
---
.../primary-key-table/merge-engine/aggregation.mdx | 2 +
.../main/java/org/apache/paimon/CoreOptions.java | 14 ++
.../compact/aggregate/FieldNestedUpdateAgg.java | 69 +++++-
.../factory/FieldNestedUpdateAggFactory.java | 13 +-
.../compact/aggregate/FieldAggregatorTest.java | 255 +++++++++++++++++++++
5 files changed, 350 insertions(+), 3 deletions(-)
diff --git a/docs/docs/primary-key-table/merge-engine/aggregation.mdx
b/docs/docs/primary-key-table/merge-engine/aggregation.mdx
index cb88297205..96e6e4f3f0 100644
--- a/docs/docs/primary-key-table/merge-engine/aggregation.mdx
+++ b/docs/docs/primary-key-table/merge-engine/aggregation.mdx
@@ -311,6 +311,8 @@ 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-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
sequentially up to limit; but if nested-key is specified, it cannot
guarantee the correctness of the aggregation result. This option can be used to
avoid abnormal input.
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 24556d61c0..f09a6edb4a 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_SEQUENCE_FIELD = "nested-sequence-field";
+
public static final String COUNT_LIMIT = "count-limit";
public static final String DISTINCT = "distinct";
@@ -2740,6 +2742,18 @@ public class CoreOptions implements Serializable {
return
Arrays.stream(keyString.split(",")).map(String::trim).collect(Collectors.toList());
}
+ public List<String> fieldNestedUpdateAggNestedSequenceField(String
fieldName) {
+ String keyString =
+ options.get(
+ key(FIELDS_PREFIX + "." + fieldName + "." +
NESTED_SEQUENCE_FIELD)
+ .stringType()
+ .noDefaultValue());
+ if (keyString == null) {
+ return Collections.emptyList();
+ }
+ return
Arrays.stream(keyString.split(",")).map(String::trim).collect(Collectors.toList());
+ }
+
public int fieldNestedUpdateAggCountLimit(String fieldName) {
return options.get(
key(FIELDS_PREFIX + "." + fieldName + "." + COUNT_LIMIT)
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 b2848b35b4..13b19e6e9b 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
@@ -19,24 +19,29 @@
package org.apache.paimon.mergetree.compact.aggregate;
import org.apache.paimon.codegen.Projection;
+import org.apache.paimon.codegen.RecordComparator;
import org.apache.paimon.codegen.RecordEqualiser;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericArray;
import org.apache.paimon.data.InternalArray;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataType;
import org.apache.paimon.types.RowType;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Collections;
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;
/**
@@ -52,10 +57,23 @@ public class FieldNestedUpdateAgg extends FieldAggregator {
@Nullable private final Projection keyProjection;
@Nullable private final RecordEqualiser elementEqualiser;
+ @Nullable private final Projection sequenceProjection;
+ @Nullable private final RecordComparator sequenceComparator;
+ private final boolean hasSequenceField;
+
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,
+ List<String> nestedSequenceField,
+ int countLimit) {
super(name, dataType);
RowType nestedType = (RowType) dataType.getElementType();
this.nestedFields = nestedType.getFieldCount();
@@ -67,6 +85,31 @@ public class FieldNestedUpdateAgg extends FieldAggregator {
this.elementEqualiser = null;
}
+ // 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;
+
+ // Extract the data types of the sequence fields to generate a
native record comparator
+ int sequenceFields = nestedSequenceField.size();
+ List<DataType> seqTypes = new ArrayList<>(sequenceFields);
+ int[] sortFields = new int[sequenceFields];
+ for (int i = 0; i < sequenceFields; i++) {
+ String fieldName = nestedSequenceField.get(i);
+
seqTypes.add(nestedType.getTypeAt(nestedType.getFieldIndex(fieldName)));
+ sortFields[i] = i;
+ }
+ this.sequenceComparator = newRecordComparator(seqTypes,
sortFields);
+ } else {
+ this.sequenceProjection = null;
+ this.sequenceComparator = null;
+ this.hasSequenceField = false;
+ }
+
// If deduplicate key is set, we don't guarantee that the result is
exactly right
this.countLimit = countLimit;
}
@@ -94,7 +137,15 @@ public class FieldNestedUpdateAgg extends FieldAggregator {
Map<BinaryRow, InternalRow> map = new HashMap<>();
for (InternalRow row : rows) {
BinaryRow key = keyProjection.apply(row).copy();
- map.put(key, row);
+ if (hasSequenceField) {
+ // When sequence field is configured, only update if the
new sequence is greater
+ InternalRow existing = map.get(key);
+ if (existing == null || compareSequence(row, existing) >=
0) {
+ map.put(key, row);
+ }
+ } else {
+ map.put(key, row);
+ }
}
rows = new ArrayList<>(map.values());
@@ -146,6 +197,22 @@ public class FieldNestedUpdateAgg extends FieldAggregator {
}
}
+ private int compareSequence(InternalRow newRow, InternalRow oldRow) {
+ checkNotNull(
+ sequenceComparator,
+ "sequenceComparator should not be null when hasSequenceField
is true.");
+ checkNotNull(
+ sequenceProjection,
+ "sequenceProjection should not be null when hasSequenceField
is true.");
+
+ // Project the rows into sub-rows containing only sequence fields
+ BinaryRow newSeqRow = sequenceProjection.apply(newRow).copy();
+ BinaryRow oldSeqRow = sequenceProjection.apply(oldRow).copy();
+
+ // Triggers native CodeGen comparison (Nulls First by default)
+ return sequenceComparator.compare(newSeqRow, oldSeqRow);
+ }
+
private void addNonNullRows(InternalArray array, List<InternalRow> rows) {
for (int i = 0; i < array.size(); i++) {
if (array.isNullAt(i)) {
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 070931e011..ca0d961498 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
@@ -39,6 +39,7 @@ public class FieldNestedUpdateAggFactory implements
FieldAggregatorFactory {
return createFieldNestedUpdateAgg(
fieldType,
options.fieldNestedUpdateAggNestedKey(field),
+ options.fieldNestedUpdateAggNestedSequenceField(field),
options.fieldNestedUpdateAggCountLimit(field));
}
@@ -48,17 +49,25 @@ public class FieldNestedUpdateAggFactory implements
FieldAggregatorFactory {
}
private FieldNestedUpdateAgg createFieldNestedUpdateAgg(
- DataType fieldType, List<String> nestedKey, int countLimit) {
+ DataType fieldType,
+ List<String> nestedKey,
+ List<String> nestedSequenceField,
+ int countLimit) {
if (nestedKey == null) {
nestedKey = Collections.emptyList();
}
+ if (nestedSequenceField == null) {
+ nestedSequenceField = Collections.emptyList();
+ }
+
String typeErrorMsg =
"Data type for nested table column must be 'Array<Row>' but
was '%s'.";
checkArgument(fieldType instanceof ArrayType, typeErrorMsg, fieldType);
ArrayType arrayType = (ArrayType) fieldType;
checkArgument(arrayType.getElementType() instanceof RowType,
typeErrorMsg, fieldType);
- return new FieldNestedUpdateAgg(identifier(), arrayType, nestedKey,
countLimit);
+ return new FieldNestedUpdateAgg(
+ identifier(), arrayType, nestedKey, nestedSequenceField,
countLimit);
}
}
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 1bdddcb846..a6a2794c36 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
@@ -749,6 +749,252 @@ public class FieldAggregatorTest {
.containsExactlyInAnyOrderElementsOf(Arrays.asList(row(0, 1,
"B"), row(0, 1, "b")));
}
+ @Test
+ public void testFieldNestedUpdateAggWithSequenceField() {
+ 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"),
+ Collections.singletonList("seq"),
+ Integer.MAX_VALUE);
+
+ 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)));
+
+ current = row(0, 1, "b", 3);
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Arrays.asList(row(0, 0, "A", 1), row(0, 1, "b", 3)));
+
+ current = row(0, 1, "B_late", 2);
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Arrays.asList(row(0, 0, "A", 1), row(0, 1, "b", 3)));
+
+ current = row(0, 1, "b", 3);
+ accumulator = (InternalArray) agg.retract(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+
.containsExactlyInAnyOrderElementsOf(Collections.singletonList(row(0, 0, "A",
1)));
+ }
+
+ @Test
+ public void testFieldNestedUpdateAggWithMultipleSequenceFields() {
+ 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()),
+ DataTypes.FIELD(4, "ts", DataTypes.TIMESTAMP(3)));
+
+ FieldNestedUpdateAgg agg =
+ new FieldNestedUpdateAgg(
+ FieldNestedUpdateAggFactory.NAME,
+ DataTypes.ARRAY(elementRowType),
+ Arrays.asList("k0", "k1"),
+ Arrays.asList("seq", "ts"),
+ Integer.MAX_VALUE);
+
+ InternalArray accumulator = null;
+ InternalArray.ElementGetter elementGetter =
+ InternalArray.createElementGetter(elementRowType);
+
+ org.apache.paimon.data.Timestamp ts1 =
+ org.apache.paimon.data.Timestamp.fromEpochMillis(1000L);
+ org.apache.paimon.data.Timestamp ts2 =
+ org.apache.paimon.data.Timestamp.fromEpochMillis(2000L);
+ org.apache.paimon.data.Timestamp ts3 =
+ org.apache.paimon.data.Timestamp.fromEpochMillis(3000L);
+
+ InternalRow current = row(1, 0, "A", 1, ts2);
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ current = row(0, 1, "B", 2, ts1);
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Arrays.asList(row(1, 0, "A", 1, ts2), row(0, 1, "B",
2, ts1)));
+
+ current = row(1, 1, "C", 1, ts2);
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Arrays.asList(
+ row(1, 0, "A", 1, ts2),
+ row(0, 1, "B", 2, ts1),
+ row(1, 1, "C", 1, ts2)));
+
+ current = row(1, 0, "A_late_updated_by_ts", 1, ts1);
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Arrays.asList(
+ row(1, 0, "A", 1, ts2),
+ row(0, 1, "B", 2, ts1),
+ row(1, 1, "C", 1, ts2)));
+
+ current = row(1, 0, "A_updated_by_ts", 1, ts3);
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Arrays.asList(
+ row(1, 0, "A_updated_by_ts", 1, ts3),
+ row(0, 1, "B", 2, ts1),
+ row(1, 1, "C", 1, ts2)));
+
+ // Try to update with a smaller 1st seq, even if the 2nd seq (ts) is
larger
+ // Result: Should be IGNORED because the 1st seq field (1 < 2) takes
higher priority.
+ current = row(0, 1, "b_ignored", 1, ts3);
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Arrays.asList(
+ row(1, 0, "A_updated_by_ts", 1, ts3),
+ row(0, 1, "B", 2, ts1),
+ row(1, 1, "C", 1, ts2)));
+
+ // Update with the SAME 1st seq, but a larger 2nd seq (ts)
+ // Result: Should be SUCCESSFULLY UPDATED because seq (2 == 2) and ts
(ts2 > ts1).
+ current = row(0, 1, "B_updated_by_ts", 2, ts2);
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Arrays.asList(
+ row(1, 0, "A_updated_by_ts", 1, ts3),
+ row(0, 1, "B_updated_by_ts", 2, ts2),
+ row(1, 1, "C", 1, ts2)));
+
+ // Update with a larger 1st seq, even if the 2nd seq (ts) is smaller
+ // Result: Should be SUCCESSFULLY UPDATED because the 1st seq field (3
> 2) wins.
+ current = row(0, 1, "B_updated_by_seq", 3, ts1);
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Arrays.asList(
+ row(1, 0, "A_updated_by_ts", 1, ts3),
+ row(0, 1, "B_updated_by_seq", 3, ts1),
+ row(1, 1, "C", 1, ts2)));
+
+ // Retract the latest row matching the current state
+ current = row(0, 1, "B_updated_by_seq", 3, ts1);
+ accumulator = (InternalArray) agg.retract(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Arrays.asList(
+ 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 =
+ 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()));
+
+ // 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))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("nested-sequence-field requires nested-key to be
set.");
+ }
+
+ @Test
+ public void testFieldNestedUpdateAggWithCountLimitWithSequenceField() {
+ 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"),
+ Collections.singletonList("seq"),
+ 2); // Enforce count limit = 2
+
+ InternalArray accumulator = null;
+ InternalArray.ElementGetter elementGetter =
+ InternalArray.createElementGetter(elementRowType);
+
+ InternalRow current = row(0, 1, "B", 1);
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+
.containsExactlyInAnyOrderElementsOf(Collections.singletonList(row(0, 1, "B",
1)));
+
+ current = row(0, 1, "B_updated", 2);
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ // The existing row should be updated, and the total size remains 1
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Collections.singletonList(row(0, 1, "B_updated", 2)));
+
+ current = row(1, 2, "C", 3); // Different nested key (1, 2)
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Arrays.asList(row(0, 1, "B_updated", 2), row(1, 2,
"C", 3)));
+
+ current = row(0, 3, "D", 4); // Another different nested key (0, 3)
+ accumulator = (InternalArray) agg.agg(accumulator,
singletonArray(current));
+
+ // count limit is 2, so the third element will be dropped
+ assertThat(unnest(accumulator, elementGetter))
+ .containsExactlyInAnyOrderElementsOf(
+ Arrays.asList(row(0, 1, "B_updated", 2), row(1, 2,
"C", 3)));
+ }
+
private List<Object> unnest(InternalArray array,
InternalArray.ElementGetter elementGetter) {
return IntStream.range(0, array.size())
.mapToObj(i -> elementGetter.getElementOrNull(array, i))
@@ -763,6 +1009,15 @@ public class FieldAggregatorTest {
return GenericRow.of(k0, k1, BinaryString.fromString(v));
}
+ private InternalRow row(Integer k0, Integer k1, String v, Integer seq) {
+ return GenericRow.of(k0, k1, BinaryString.fromString(v), seq);
+ }
+
+ private InternalRow row(
+ Object k0, Object k1, String v, Object seq,
org.apache.paimon.data.Timestamp ts) {
+ return GenericRow.of(k0, k1, BinaryString.fromString(v), seq, ts);
+ }
+
@Test
public void testFieldCollectAggWithDistinct() {
FieldCollectAgg agg =
