From 760a5f7877a89b0f0e7f415a9cb48e708c7a12e5 Mon Sep 17 00:00:00 2001 From: jackylee Date: Wed, 5 Aug 2026 13:01:47 +0800 Subject: [PATCH] [core] Validate data types for max/min aggregate functions ### Purpose `FieldMaxAggFactory` / `FieldMinAggFactory` create the aggregator without checking the column type. `FieldMaxAgg#agg` delegates to `InternalRowUtils.compare`, which only handles ordered types, so `fields..aggregate-function = max` on an ARRAY / MAP / MULTISET / ROW / VARIANT / BLOB / VECTOR column is accepted and only fails later during merging with `Incomparable type: ARRAY`, naming neither the field nor the function. #4446 moved this kind of check into the factories and #7485 did the same for `listagg`; max/min were never migrated. `compare` also had no BOOLEAN case, although `TypeCheckUtils.isComparable` treats BOOLEAN as comparable and the codegen comparator already implements it, so BOOLEAN is added rather than rejected. With that, the set of types the factory admits equals the set `compare` can order; a test pins the invariant. ### Tests `FieldAggregatorTest`, `InternalRowUtilsTest`. --- .../merge-engine/aggregation.mdx | 4 +- .../apache/paimon/utils/InternalRowUtils.java | 3 + .../paimon/utils/InternalRowUtilsTest.java | 5 + .../aggregate/factory/FieldMaxAggFactory.java | 8 ++ .../aggregate/factory/FieldMinAggFactory.java | 8 ++ .../aggregate/FieldAggregatorTest.java | 103 ++++++++++++++++++ 6 files changed, 129 insertions(+), 2 deletions(-) diff --git a/docs/docs/primary-key-table/merge-engine/aggregation.mdx b/docs/docs/primary-key-table/merge-engine/aggregation.mdx index 823ae7b0169a..ca4a8cf8cff2 100644 --- a/docs/docs/primary-key-table/merge-engine/aggregation.mdx +++ b/docs/docs/primary-key-table/merge-engine/aggregation.mdx @@ -85,11 +85,11 @@ Current supported aggregate functions and data types are: ### max The max function identifies and retains the maximum value. - It supports CHAR, VARCHAR, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ data types. + It supports BOOLEAN, CHAR, VARCHAR, BINARY, VARBINARY, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ data types. ### min The min function identifies and retains the minimum value. - It supports CHAR, VARCHAR, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ data types. + It supports BOOLEAN, CHAR, VARCHAR, BINARY, VARBINARY, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ data types. ### last_value The last_value function replaces the previous value with the most recently imported value. diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/InternalRowUtils.java b/paimon-common/src/main/java/org/apache/paimon/utils/InternalRowUtils.java index 1ff433f3e24d..4c329797f39e 100644 --- a/paimon-common/src/main/java/org/apache/paimon/utils/InternalRowUtils.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/InternalRowUtils.java @@ -446,6 +446,9 @@ public static InternalRow.FieldGetter createNullCheckingFieldGetter( public static int compare(Object x, Object y, DataTypeRoot type) { int ret; switch (type) { + case BOOLEAN: + ret = Boolean.compare((boolean) x, (boolean) y); + break; case DECIMAL: Decimal xDD = (Decimal) x; Decimal yDD = (Decimal) y; diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/InternalRowUtilsTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/InternalRowUtilsTest.java index c085147c4106..dec1ab4ad66f 100644 --- a/paimon-common/src/test/java/org/apache/paimon/utils/InternalRowUtilsTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/utils/InternalRowUtilsTest.java @@ -120,6 +120,11 @@ private BinaryRow toBinary(InternalRow row) { @Test public void testCompare() { + // test BOOLEAN data type + assertThat(InternalRowUtils.compare(false, true, DataTypeRoot.BOOLEAN)).isLessThan(0); + assertThat(InternalRowUtils.compare(true, false, DataTypeRoot.BOOLEAN)).isGreaterThan(0); + assertThat(InternalRowUtils.compare(true, true, DataTypeRoot.BOOLEAN)).isEqualTo(0); + // test DECIMAL data type Decimal xDecimalData = Decimal.fromBigDecimal(new BigDecimal("12.34"), 4, 2); Decimal yDecimalData = Decimal.fromBigDecimal(new BigDecimal("13.14"), 4, 2); diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMaxAggFactory.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMaxAggFactory.java index 4e3c33171a89..22f22f371caa 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMaxAggFactory.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMaxAggFactory.java @@ -21,6 +21,9 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.mergetree.compact.aggregate.FieldMaxAgg; import org.apache.paimon.types.DataType; +import org.apache.paimon.utils.TypeCheckUtils; + +import static org.apache.paimon.utils.Preconditions.checkArgument; /** Factory for #{@link FieldMaxAgg}. */ public class FieldMaxAggFactory implements FieldAggregatorFactory { @@ -29,6 +32,11 @@ public class FieldMaxAggFactory implements FieldAggregatorFactory { @Override public FieldMaxAgg create(DataType fieldType, CoreOptions options, String field) { + checkArgument( + TypeCheckUtils.isComparable(fieldType), + "Data type for max column '%s' must be comparable but was '%s'.", + field, + fieldType); return new FieldMaxAgg(identifier(), fieldType); } diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMinAggFactory.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMinAggFactory.java index 4ac7c08b1904..e5ee8d9d16cd 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMinAggFactory.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldMinAggFactory.java @@ -21,6 +21,9 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.mergetree.compact.aggregate.FieldMinAgg; import org.apache.paimon.types.DataType; +import org.apache.paimon.utils.TypeCheckUtils; + +import static org.apache.paimon.utils.Preconditions.checkArgument; /** Factory for #{@link FieldMinAgg}. */ public class FieldMinAggFactory implements FieldAggregatorFactory { @@ -29,6 +32,11 @@ public class FieldMinAggFactory implements FieldAggregatorFactory { @Override public FieldMinAgg create(DataType fieldType, CoreOptions options, String field) { + checkArgument( + TypeCheckUtils.isComparable(fieldType), + "Data type for min column '%s' must be comparable but was '%s'.", + field, + fieldType); return new FieldMinAgg(identifier(), fieldType); } 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 5f9537e5a75d..90ade26f7129 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 @@ -27,6 +27,7 @@ import org.apache.paimon.data.InternalArray; import org.apache.paimon.data.InternalMap; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.Timestamp; import org.apache.paimon.mergetree.compact.aggregate.factory.FieldAggregatorFactory; import org.apache.paimon.mergetree.compact.aggregate.factory.FieldBoolAndAggFactory; import org.apache.paimon.mergetree.compact.aggregate.factory.FieldBoolOrAggFactory; @@ -51,6 +52,7 @@ import org.apache.paimon.types.BigIntType; import org.apache.paimon.types.BooleanType; import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypeRoot; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.DecimalType; import org.apache.paimon.types.DoubleType; @@ -64,6 +66,7 @@ import org.apache.paimon.utils.HllSketchUtil; import org.apache.paimon.utils.RoaringBitmap32; import org.apache.paimon.utils.RoaringBitmap64; +import org.apache.paimon.utils.TypeCheckUtils; import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap; @@ -75,8 +78,11 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.stream.Collectors; import java.util.stream.IntStream; import java.util.stream.Stream; @@ -513,6 +519,103 @@ public void testFieldMinAgg() { assertThat(fieldMinAgg.agg(accumulator, inputField)).isEqualTo(1); } + @Test + public void testFieldMaxMinAggWithBooleanType() { + FieldMaxAgg fieldMaxAgg = new FieldMaxAggFactory().create(new BooleanType(), null, null); + assertThat(fieldMaxAgg.agg(null, true)).isEqualTo(true); + assertThat(fieldMaxAgg.agg(false, null)).isEqualTo(false); + assertThat(fieldMaxAgg.agg(false, false)).isEqualTo(false); + assertThat(fieldMaxAgg.agg(false, true)).isEqualTo(true); + assertThat(fieldMaxAgg.agg(true, false)).isEqualTo(true); + assertThat(fieldMaxAgg.agg(true, true)).isEqualTo(true); + + FieldMinAgg fieldMinAgg = new FieldMinAggFactory().create(new BooleanType(), null, null); + assertThat(fieldMinAgg.agg(null, false)).isEqualTo(false); + assertThat(fieldMinAgg.agg(true, null)).isEqualTo(true); + assertThat(fieldMinAgg.agg(true, true)).isEqualTo(true); + assertThat(fieldMinAgg.agg(true, false)).isEqualTo(false); + assertThat(fieldMinAgg.agg(false, true)).isEqualTo(false); + assertThat(fieldMinAgg.agg(false, false)).isEqualTo(false); + } + + @Test + public void testFieldMaxMinAggComparableTypesAreAllSupported() { + // The factory admits a field iff TypeCheckUtils.isComparable, so every admitted type must + // be one InternalRowUtils.compare can actually order. Keep the two in lockstep: a new + // comparable type must be added to compare() in the same change. + Map samples = new LinkedHashMap<>(); + samples.put(DataTypes.BOOLEAN(), true); + samples.put(DataTypes.TINYINT(), (byte) 1); + samples.put(DataTypes.SMALLINT(), (short) 1); + samples.put(DataTypes.INT(), 1); + samples.put(DataTypes.BIGINT(), 1L); + samples.put(DataTypes.FLOAT(), 1.0f); + samples.put(DataTypes.DOUBLE(), 1.0d); + samples.put(DataTypes.DECIMAL(4, 2), Decimal.fromUnscaledLong(1, 4, 2)); + samples.put(DataTypes.CHAR(1), BinaryString.fromString("a")); + samples.put(DataTypes.VARCHAR(1), BinaryString.fromString("a")); + samples.put(DataTypes.BINARY(1), new byte[] {1}); + samples.put(DataTypes.VARBINARY(1), new byte[] {1}); + samples.put(DataTypes.DATE(), 1); + samples.put(DataTypes.TIME(), 1); + samples.put(DataTypes.TIMESTAMP(), Timestamp.fromEpochMillis(1)); + samples.put(DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(), Timestamp.fromEpochMillis(1)); + + for (Map.Entry entry : samples.entrySet()) { + DataType type = entry.getKey(); + Object value = entry.getValue(); + assertThat(TypeCheckUtils.isComparable(type)).as("isComparable(%s)", type).isTrue(); + assertThat(new FieldMaxAggFactory().create(type, null, "f").agg(value, value)) + .as("max on %s", type) + .isEqualTo(value); + assertThat(new FieldMinAggFactory().create(type, null, "f").agg(value, value)) + .as("min on %s", type) + .isEqualTo(value); + } + + // Guard against a new comparable type root slipping in without being covered above: the + // sampled roots must be exactly the roots that are not excluded by isComparable. + Set sampledRoots = new HashSet<>(); + samples.keySet().forEach(type -> sampledRoots.add(type.getTypeRoot())); + Set expectedRoots = new HashSet<>(Arrays.asList(DataTypeRoot.values())); + expectedRoots.removeAll( + Arrays.asList( + DataTypeRoot.MAP, + DataTypeRoot.MULTISET, + DataTypeRoot.ROW, + DataTypeRoot.ARRAY, + DataTypeRoot.VECTOR, + DataTypeRoot.VARIANT, + DataTypeRoot.BLOB)); + assertThat(sampledRoots) + .as("a comparable type root must be covered here and in InternalRowUtils.compare") + .isEqualTo(expectedRoots); + } + + @Test + public void testFieldMaxMinAggWithIncomparableTypeShouldFail() { + // These types have no ordering, so max/min must be rejected when the aggregator is + // created rather than failing later during merging. + for (DataType incomparable : + Arrays.asList( + DataTypes.ARRAY(DataTypes.INT()), + DataTypes.MAP(DataTypes.INT(), DataTypes.INT()), + DataTypes.MULTISET(DataTypes.INT()), + DataTypes.ROW(DataTypes.FIELD(0, "f0", DataTypes.INT())), + DataTypes.VARIANT(), + DataTypes.BLOB(), + DataTypes.VECTOR(3, DataTypes.FLOAT()))) { + assertThatThrownBy(() -> new FieldMaxAggFactory().create(incomparable, null, "label")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "Data type for max column 'label' must be comparable but was"); + assertThatThrownBy(() -> new FieldMinAggFactory().create(incomparable, null, "label")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "Data type for min column 'label' must be comparable but was"); + } + } + @Test public void testFieldSumIntAgg() { FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new IntType(), null, null);