This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit fa1178fa96f8ec5bef5be60d0ff898e9c9d291e8 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)));
