[core] Support boolean for min/max aggregate functions

docs/docs/primary-key-table/merge-engine/aggregation.mdx

@@ -85,11 +85,13 @@ 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, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ data types.
+  For BOOLEAN, `false` is ordered before `true`, so max behaves like a logical OR.
 
 ### 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, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ data types.
+  For BOOLEAN, `false` is ordered before `true`, so min behaves like a logical AND.
 
 ### 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-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java

@@ -513,6 +513,28 @@ public void testFieldMinAgg() {
         assertThat(fieldMinAgg.agg(accumulator, inputField)).isEqualTo(1);
     }
 
+    @Test
+    public void testFieldMaxAggBoolean() {
+        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);
+    }
+
+    @Test
+    public void testFieldMinAggBoolean() {
+        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 testFieldSumIntAgg() {
         FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new IntType(), null, null);

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/MaxAggregationITCase.java

@@ -26,7 +26,7 @@
 import java.math.BigDecimal;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
-import java.util.Collections;
+import java.util.Arrays;
 import java.util.List;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -36,7 +36,7 @@
 public class MaxAggregationITCase extends CatalogITCaseBase {
     @Override
     protected List<String> ddl() {
-        return Collections.singletonList(
+        return Arrays.asList(
                 "CREATE TABLE IF NOT EXISTS T2 ("
                         + "j INT, k INT, "
                         + "a INT, "
@@ -63,9 +63,34 @@ protected List<String> ddl() {
                         + "'fields.l.aggregate-function'='max',"
                         + "'fields.m.aggregate-function'='max',"
                         + "'fields.n.aggregate-function'='max'"
+                        + ");",
+                "CREATE TABLE IF NOT EXISTS T_BOOL_MAX ("
+                        + "j INT, k INT, "
+                        + "o BOOLEAN, "
+                        + "PRIMARY KEY (j,k) NOT ENFORCED)"
+                        + " WITH ('merge-engine'='aggregation', "
+                        + "'fields.o.aggregate-function'='max'"
                         + ");");
     }
 
+    @Test
+    public void testBooleanMergeInMemory() {
+        batchSql(
+                "INSERT INTO T_BOOL_MAX VALUES (1, 2, CAST(NULL AS BOOLEAN)), (1, 2, false), "
+                        + "(1, 2, true), (1, 3, false), (1, 3, false)");
+        assertThat(batchSql("SELECT * FROM T_BOOL_MAX"))
+                .containsExactlyInAnyOrder(Row.of(1, 2, true), Row.of(1, 3, false));
+    }
+
+    @Test
+    public void testBooleanMergeRead() {
+        batchSql("INSERT INTO T_BOOL_MAX VALUES (1, 2, false)");
+        batchSql("INSERT INTO T_BOOL_MAX VALUES (1, 2, true)");
+        batchSql("INSERT INTO T_BOOL_MAX VALUES (1, 2, false)");
+        assertThat(batchSql("SELECT * FROM T_BOOL_MAX"))
+                .containsExactlyInAnyOrder(Row.of(1, 2, true));
+    }
+
     @Test
     public void testMergeInMemory() {
         batchSql(

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/aggregation/MinAggregationITCase.java

@@ -26,7 +26,7 @@
 import java.math.BigDecimal;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
-import java.util.Collections;
+import java.util.Arrays;
 import java.util.List;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -36,7 +36,7 @@
 public class MinAggregationITCase extends CatalogITCaseBase {
     @Override
     protected List<String> ddl() {
-        return Collections.singletonList(
+        return Arrays.asList(
                 "CREATE TABLE IF NOT EXISTS T3 ("
                         + "j INT, k INT, "
                         + "a INT, "
@@ -63,9 +63,34 @@ protected List<String> ddl() {
                         + "'fields.l.aggregate-function'='min',"
                         + "'fields.m.aggregate-function'='min',"
                         + "'fields.n.aggregate-function'='min'"
+                        + ");",
+                "CREATE TABLE IF NOT EXISTS T_BOOL_MIN ("
+                        + "j INT, k INT, "
+                        + "o BOOLEAN, "
+                        + "PRIMARY KEY (j,k) NOT ENFORCED)"
+                        + " WITH ('merge-engine'='aggregation', "
+                        + "'fields.o.aggregate-function'='min'"
                         + ");");
     }
 
+    @Test
+    public void testBooleanMergeInMemory() {
+        batchSql(
+                "INSERT INTO T_BOOL_MIN VALUES (1, 2, CAST(NULL AS BOOLEAN)), (1, 2, true), "
+                        + "(1, 2, false), (1, 3, true), (1, 3, true)");
+        assertThat(batchSql("SELECT * FROM T_BOOL_MIN"))
+                .containsExactlyInAnyOrder(Row.of(1, 2, false), Row.of(1, 3, true));
+    }
+
+    @Test
+    public void testBooleanMergeRead() {
+        batchSql("INSERT INTO T_BOOL_MIN VALUES (1, 2, true)");
+        batchSql("INSERT INTO T_BOOL_MIN VALUES (1, 2, false)");
+        batchSql("INSERT INTO T_BOOL_MIN VALUES (1, 2, true)");
+        assertThat(batchSql("SELECT * FROM T_BOOL_MIN"))
+                .containsExactlyInAnyOrder(Row.of(1, 2, false));
+    }
+
     @Test
     public void testMergeInMemory() {
         batchSql(