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 5411d063c5 [core] Validate data types for max/min aggregate functions
(#9013)
5411d063c5 is described below
commit 5411d063c5d9b6b22c82eff3bde6e4d8839e507f
Author: jackylee <[email protected]>
AuthorDate: Fri Aug 7 13:17:21 2026 +0800
[core] Validate data types for max/min aggregate functions (#9013)
---
.../primary-key-table/merge-engine/aggregation.mdx | 4 +-
.../org/apache/paimon/utils/InternalRowUtils.java | 3 +
.../apache/paimon/utils/InternalRowUtilsTest.java | 5 +
.../aggregate/factory/FieldMaxAggFactory.java | 8 ++
.../aggregate/factory/FieldMinAggFactory.java | 8 ++
.../compact/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 823ae7b016..ca4a8cf8cf 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 1ff433f3e2..4c329797f3 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 class InternalRowUtils {
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 c085147c41..dec1ab4ad6 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 @@ public class InternalRowUtilsTest {
@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 4e3c33171a..22f22f371c 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 @@ package org.apache.paimon.mergetree.compact.aggregate.factory;
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 4ac7c08b19..e5ee8d9d16 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 @@ package org.apache.paimon.mergetree.compact.aggregate.factory;
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 5f9537e5a7..90ade26f71 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.GenericRow;
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.ArrayType;
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.types.VarCharType;
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.AbstractMap;
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 class FieldAggregatorTest {
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<DataType, Object> 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<DataType, Object> 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<DataTypeRoot> sampledRoots = new HashSet<>();
+ samples.keySet().forEach(type -> sampledRoots.add(type.getTypeRoot()));
+ Set<DataTypeRoot> 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);
