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 81a5682208 [core] Fix remaining byte[] identity comparisons in collect
and merge_map aggregators (#9870)
81a5682208 is described below
commit 81a5682208fccc73132ae480a160e9dd40966fa3
Author: Eunbin Son <[email protected]>
AuthorDate: Thu Sep 17 14:09:48 2026 +0900
[core] Fix remaining byte[] identity comparisons in collect and merge_map
aggregators (#9870)
---
.../mergetree/compact/aggregate/BinaryMapKeys.java | 19 ++-
.../compact/aggregate/FieldCollectAgg.java | 18 +--
.../compact/aggregate/FieldAggregatorTest.java | 132 +++++++++++++++++++++
3 files changed, 156 insertions(+), 13 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/BinaryMapKeys.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/BinaryMapKeys.java
index 5fc1eb68e4..4a2e1df388 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/BinaryMapKeys.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/BinaryMapKeys.java
@@ -20,14 +20,15 @@ package org.apache.paimon.mergetree.compact.aggregate;
import org.apache.paimon.data.GenericMap;
import org.apache.paimon.types.DataType;
-import org.apache.paimon.types.DataTypeFamily;
+import org.apache.paimon.types.DataTypeRoot;
import org.apache.paimon.utils.ByteArrayKey;
import java.util.HashMap;
import java.util.Map;
/**
- * Value semantics for map keys of type {@code BINARY} or {@code VARBINARY}.
+ * Value semantics for map keys of type {@code BINARY}, {@code VARBINARY},
{@code GEOMETRY} or
+ * {@code GEOGRAPHY}.
*
* <p>Such a key arrives as a {@code byte[]}, which inherits identity equality
from {@link Object}.
* Used directly as a hash key, two keys with the same content occupy two
entries, a lookup never
@@ -39,8 +40,18 @@ final class BinaryMapKeys {
private BinaryMapKeys() {}
- static boolean isBinary(DataType keyType) {
- return
keyType.getTypeRoot().getFamilies().contains(DataTypeFamily.BINARY_STRING);
+ /**
+ * Whether values of this type are held as a {@code byte[]}. This is the
set of roots that
+ * {@link org.apache.paimon.data.InternalArray#createElementGetter} reads
with {@code
+ * getBinary}: {@code BINARY} and {@code VARBINARY}, plus {@code GEOMETRY}
and {@code GEOGRAPHY}
+ * whose in-memory value is WKB.
+ */
+ static boolean isBinary(DataType type) {
+ return type.isAnyOf(
+ DataTypeRoot.BINARY,
+ DataTypeRoot.VARBINARY,
+ DataTypeRoot.GEOMETRY,
+ DataTypeRoot.GEOGRAPHY);
}
/** Wrap a key for storage in a hash collection; a no-op for every
non-binary key type. */
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
index 368ae685a9..60f60e82ce 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
@@ -36,7 +36,6 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
-import java.util.Set;
import java.util.function.BiFunction;
import static org.apache.paimon.codegen.CodeGenUtils.newRecordEqualiser;
@@ -55,7 +54,9 @@ public class FieldCollectAgg extends FieldAggregator {
this.distinct = distinct;
this.elementGetter =
InternalArray.createElementGetter(dataType.getElementType());
- if (distinct && needsEqualiser(dataType.getElementType())) {
+ // The equaliser is built from the element type alone: retract() needs
it whether or not
+ // the array is distinct, and agg() gates de-duplication on 'distinct'
separately.
+ if (needsEqualiser(dataType.getElementType())) {
DataType elementType = dataType.getElementType();
List<DataType> fieldTypes =
elementType instanceof RowType
@@ -83,14 +84,13 @@ public class FieldCollectAgg extends FieldAggregator {
* Whether elements of this type need the generated equaliser rather than
{@link Object#equals}.
*
* <p>Constructed types need it because two rows holding the same values
are not necessarily
- * equal objects. Binary types need it for a blunter reason: an element of
{@code BINARY} or
- * {@code VARBINARY} is a {@code byte[]}, which inherits identity equality
from {@link Object},
- * so two arrays with the same content never compare equal and never share
a hash bucket.
+ * equal objects. Binary types need it for a blunter reason: an element of
{@code BINARY},
+ * {@code VARBINARY}, {@code GEOMETRY} or {@code GEOGRAPHY} is a {@code
byte[]}, which inherits
+ * identity equality from {@link Object}, so two arrays with the same
content never compare
+ * equal and never share a hash bucket.
*/
private static boolean needsEqualiser(DataType elementType) {
- Set<DataTypeFamily> families = elementType.getTypeRoot().getFamilies();
- return families.contains(DataTypeFamily.CONSTRUCTED)
- || families.contains(DataTypeFamily.BINARY_STRING);
+ return elementType.is(DataTypeFamily.CONSTRUCTED) ||
BinaryMapKeys.isBinary(elementType);
}
@Override
@@ -111,7 +111,7 @@ public class FieldCollectAgg extends FieldAggregator {
return accumulator == null ? inputField : accumulator;
}
- if (equaliser != null) {
+ if (distinct && equaliser != null) {
List<Object> collection = new ArrayList<>();
// do not need to distinct accumulator, because the accumulator is
always distinct, no
// need to distinct it every time
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 6cc3c73a01..2dc2eb7890 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
@@ -2158,6 +2158,90 @@ public class FieldAggregatorTest {
.containsExactly(new byte[] {3, 4});
}
+ /**
+ * A {@code GEOMETRY} or {@code GEOGRAPHY} element is a WKB {@code byte[]}
just like a binary
+ * one, so distinct collection has to compare it by content as well.
+ */
+ @Test
+ public void testFieldCollectAggWithDistinctGeospatial() {
+ for (DataType elementType : Arrays.asList(DataTypes.GEOMETRY(),
DataTypes.GEOGRAPHY())) {
+ FieldCollectAgg agg =
+ new FieldCollectAggFactory()
+ .create(
+ DataTypes.ARRAY(elementType),
+ CoreOptions.fromMap(
+
ImmutableMap.of("fields.fieldName.distinct", "true")),
+ "fieldName");
+ InternalArray.ElementGetter elementGetter =
+ InternalArray.createElementGetter(elementType);
+
+ InternalArray result =
+ (InternalArray)
+ agg.agg(
+ new GenericArray(new Object[] {new byte[]
{1, 2}}),
+ new GenericArray(
+ new Object[] {new byte[] {1, 2},
new byte[] {3, 4}}));
+
+ assertThat(unnest(result, elementGetter))
+ .as(elementType.toString())
+ .usingRecursiveFieldByFieldElementComparator()
+ .containsExactlyInAnyOrder(new byte[] {1, 2}, new byte[]
{3, 4});
+ }
+ }
+
+ /**
+ * Retraction compares by content even when the array is not distinct: a
retracted binary
+ * element must still be removed from the accumulator.
+ */
+ @Test
+ public void testFieldCollectAggRetractWithBinary() {
+ FieldCollectAgg agg =
+ new FieldCollectAggFactory()
+ .create(
+ DataTypes.ARRAY(DataTypes.VARBINARY(10)),
+ CoreOptions.fromMap(
+
ImmutableMap.of("fields.fieldName.distinct", "false")),
+ "fieldName");
+ InternalArray.ElementGetter elementGetter =
+ InternalArray.createElementGetter(DataTypes.VARBINARY(10));
+
+ InternalArray result =
+ (InternalArray)
+ agg.retract(
+ new GenericArray(
+ new Object[] {new byte[] {1, 2}, new
byte[] {3, 4}}),
+ new GenericArray(new Object[] {new byte[] {1,
2}}));
+
+ assertThat(unnest(result, elementGetter))
+ .usingRecursiveFieldByFieldElementComparator()
+ .containsExactly(new byte[] {3, 4});
+ }
+
+ /** Without distinct, a binary array keeps duplicates: the equaliser must
not de-duplicate. */
+ @Test
+ public void testFieldCollectAggKeepsDuplicatesWithBinary() {
+ FieldCollectAgg agg =
+ new FieldCollectAggFactory()
+ .create(
+ DataTypes.ARRAY(DataTypes.VARBINARY(10)),
+ CoreOptions.fromMap(
+
ImmutableMap.of("fields.fieldName.distinct", "false")),
+ "fieldName");
+ InternalArray.ElementGetter elementGetter =
+ InternalArray.createElementGetter(DataTypes.VARBINARY(10));
+
+ InternalArray result =
+ (InternalArray)
+ agg.agg(
+ new GenericArray(new Object[] {new byte[] {1,
2}}),
+ new GenericArray(
+ new Object[] {new byte[] {1, 2}, new
byte[] {3, 4}}));
+
+ assertThat(unnest(result, elementGetter))
+ .usingRecursiveFieldByFieldElementComparator()
+ .containsExactly(new byte[] {1, 2}, new byte[] {1, 2}, new
byte[] {3, 4});
+ }
+
@Test
public void testFiledCollectAggWithRowType() {
RowType rowType = RowType.of(DataTypes.INT(), DataTypes.STRING());
@@ -2594,6 +2678,25 @@ public class FieldAggregatorTest {
assertThat(binaryKeyed(result)).containsOnlyKeys("0304");
}
+ /** A {@code GEOMETRY} key is a WKB {@code byte[]} and must be merged by
content as well. */
+ @Test
+ public void testFieldMergeMapAggWithGeospatialKey() {
+ FieldMergeMapAgg agg =
+ new FieldMergeMapAggFactory()
+ .create(DataTypes.MAP(DataTypes.GEOMETRY(),
DataTypes.INT()), null, null);
+
+ Map<Object, Object> first = new HashMap<>();
+ first.put(new byte[] {1, 2}, 1);
+ Map<Object, Object> second = new HashMap<>();
+ second.put(new byte[] {1, 2}, 2);
+ second.put(new byte[] {3, 4}, 3);
+
+ InternalMap merged = (InternalMap) agg.agg(new GenericMap(first), new
GenericMap(second));
+
+ assertThat(merged.size()).isEqualTo(2);
+ assertThat(binaryKeyed(merged)).containsOnlyKeys("0102",
"0304").containsValues(2, 3);
+ }
+
/** Render an {@code InternalMap} with binary keys as hex so it can be
asserted by value. */
private Map<String, Object> binaryKeyed(InternalMap map) {
InternalArray.ElementGetter keyGetter =
@@ -2943,6 +3046,35 @@ public class FieldAggregatorTest {
assertThat(((InternalMap) acc).size()).isEqualTo(0);
}
+ /** The same walk with a {@code GEOMETRY} key, which is a WKB {@code
byte[]} as well. */
+ @Test
+ public void testFieldMergeMapWithKeyTimeAggWithGeospatialKey() {
+ MapType mapType =
+ DataTypes.MAP(
+ DataTypes.GEOMETRY(),
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "actual_value",
DataTypes.STRING()),
+ DataTypes.FIELD(1, "dbsync_ts",
DataTypes.STRING())));
+ FieldMergeMapWithKeyTimeAgg agg = new
FieldMergeMapWithKeyTimeAgg("test", mapType, 1);
+
+ Object acc = agg.agg(null, binaryKeyedMap(new byte[] {1, 2}, "A",
"100"));
+
+ acc = agg.agg(acc, binaryKeyedMap(new byte[] {1, 2}, "A1", "200"));
+ InternalMap merged = (InternalMap) acc;
+ assertThat(merged.size()).isEqualTo(1);
+ assertThat(firstRowValue(merged)).isEqualTo("A1");
+
+ acc = agg.agg(acc, binaryKeyedMap(new byte[] {1, 2}, "A0", "050"));
+ merged = (InternalMap) acc;
+ assertThat(merged.size()).isEqualTo(1);
+ assertThat(firstRowValue(merged)).isEqualTo("A1");
+
+ Map<Object, Object> tombstone = new HashMap<>();
+ tombstone.put(new byte[] {1, 2}, null);
+ acc = agg.agg(acc, new GenericMap(tombstone));
+ assertThat(((InternalMap) acc).size()).isEqualTo(0);
+ }
+
private GenericMap binaryKeyedMap(byte[] key, String value, String ts) {
Map<Object, Object> map = new HashMap<>();
map.put(key, GenericRow.of(BinaryString.fromString(value),
BinaryString.fromString(ts)));
