[core] Validate data types for max/min aggregate functions

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.

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;

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);

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);
     }
 

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);
     }
 

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<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);