[common] Never compare null literals when merging range predicates

paimon-common/src/main/java/org/apache/paimon/predicate/Between.java

@@ -27,6 +27,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 
 import static org.apache.paimon.predicate.CompareUtils.compareLiteral;
@@ -133,7 +134,7 @@ private static List<LeafPredicate> mergeLessAndGreaterToBetween(
             if (leafPredicate.function() == LessOrEqual.INSTANCE) {
                 if (lessOrEqual == null) {
                     lessOrEqual = leafPredicate;
-                } else {
+                } else if (!hasNullLiteral(lessOrEqual) && !hasNullLiteral(leafPredicate)) {
                     lessOrEqual =
                             compareLiteral(
                                                     type,
@@ -142,11 +143,14 @@ private static List<LeafPredicate> mergeLessAndGreaterToBetween(
                                             < 0
                                     ? lessOrEqual
                                     : leafPredicate;
+                } else {
+                    // A null bound matches nothing; keep the null-bearing predicate.
+                    lessOrEqual = hasNullLiteral(lessOrEqual) ? lessOrEqual : leafPredicate;
                 }
             } else if (leafPredicate.function() == GreaterOrEqual.INSTANCE) {
                 if (greaterOrEqual == null) {
                     greaterOrEqual = leafPredicate;
-                } else {
+                } else if (!hasNullLiteral(greaterOrEqual) && !hasNullLiteral(leafPredicate)) {
                     greaterOrEqual =
                             compareLiteral(
                                                     type,
@@ -155,6 +159,9 @@ private static List<LeafPredicate> mergeLessAndGreaterToBetween(
                                             > 0
                                     ? greaterOrEqual
                                     : leafPredicate;
+                } else {
+                    greaterOrEqual =
+                            hasNullLiteral(greaterOrEqual) ? greaterOrEqual : leafPredicate;
                 }
             } else {
                 result.add(leafPredicate);
@@ -166,7 +173,12 @@ private static List<LeafPredicate> mergeLessAndGreaterToBetween(
             // Determine which is the lower bound and which is the upper bound
             Object lowerBound = greaterOrEqual.literals().get(0);
             Object upperBound = lessOrEqual.literals().get(0);
-            if (compareLiteral(type, lowerBound, upperBound) >= 0) {
+            if (lowerBound == null || upperBound == null) {
+                // A null bound makes the conjunction match nothing; never compare
+                // nulls (SQL null literals are unordered).
+                result.add(lessOrEqual);
+                result.add(greaterOrEqual);
+            } else if (compareLiteral(type, lowerBound, upperBound) >= 0) {
                 // No valid intersection, keep all original predicates
                 result.add(lessOrEqual);
                 result.add(greaterOrEqual);
@@ -190,6 +202,10 @@ private static List<LeafPredicate> mergeLessAndGreaterToBetween(
         return result;
     }
 
+    private static boolean hasNullLiteral(LeafPredicate predicate) {
+        return predicate.literals().stream().anyMatch(Objects::isNull);
+    }
+
     private static List<LeafPredicate> mergeMultipleBetweens(
             FieldTransform field, List<LeafPredicate> predicates) {
         List<LeafPredicate> results = new ArrayList<>();
@@ -211,17 +227,27 @@ private static List<LeafPredicate> mergeMultipleBetweens(
         Object maxLower = null;
         Object minUpper = null;
 
+        boolean anyNullLiteral = false;
         for (LeafPredicate between : betweens) {
             Object lower = between.literals().get(0);
             Object upper = between.literals().get(1);
 
+            if (lower == null || upper == null) {
+                anyNullLiteral = true;
+                continue;
+            }
             if (maxLower == null || compareLiteral(fieldType, lower, maxLower) > 0) {
                 maxLower = lower;
             }
             if (minUpper == null || compareLiteral(fieldType, upper, minUpper) < 0) {
                 minUpper = upper;
             }
         }
+        if (anyNullLiteral) {
+            // A null bound makes the conjunction match nothing; leave the predicates
+            // unmerged instead of comparing nulls.
+            return predicates;
+        }
 
         // Check if intersection is valid
         if (maxLower != null

paimon-common/src/main/java/org/apache/paimon/predicate/CompareUtils.java

@@ -27,6 +27,11 @@ public class CompareUtils {
     private CompareUtils() {}
 
     public static int compareLiteral(DataType type, Object v1, Object v2) {
+        // SQL null literals are unordered, so a caller that merges range predicates has to
+        // handle a null bound rather than ask for its order.
+        if (v1 == null || v2 == null) {
+            throw new IllegalArgumentException("Null literal cannot be compared for type: " + type);
+        }
         if (v1 instanceof Comparable) {
             return ((Comparable<Object>) v1).compareTo(v2);
         } else if (v1 instanceof byte[]) {

paimon-common/src/test/java/org/apache/paimon/predicate/BetweenTest.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.predicate;
 
+import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.types.IntType;
 import org.apache.paimon.types.RowType;
 
@@ -47,6 +48,44 @@ public void testOneLessOrEqualNotRewrite() {
         assertThat(compoundResult.children().get(0)).isEqualTo(isNotNull);
     }
 
+    @Test
+    public void testNullLiteralBoundsDoNotCrash() {
+        PredicateBuilder builder = new PredicateBuilder(RowType.of(new IntType()));
+        // x <= NULL AND x >= 1: Flink pushdown keeps null literals; optimize()
+        // previously crashed comparing null (Unsupported type / NPE).
+        Predicate lteNull = builder.lessOrEqual(0, null);
+        Predicate gte = builder.greaterOrEqual(0, 1);
+        Predicate and = PredicateBuilder.and(Arrays.asList(lteNull, gte));
+        assertThat(and).isNotNull();
+
+        // Two <= bounds where one is null: keeps the null-bearing predicate.
+        Predicate lte10 = builder.lessOrEqual(0, 10);
+        Predicate andNulls = PredicateBuilder.and(Arrays.asList(lteNull, lte10, gte));
+        assertThat(andNulls).isNotNull();
+
+        // Two >= bounds where one is null.
+        Predicate gteNull = builder.greaterOrEqual(0, null);
+        Predicate andGteNulls =
+                PredicateBuilder.and(Arrays.asList(gteNull, builder.greaterOrEqual(0, 5), lte10));
+        assertThat(andGteNulls).isNotNull();
+
+        // Two BETWEENs where one has a null bound stay unmerged: dropping the
+        // null-bound BETWEEN would wrongly match rows 1..4.
+        Predicate betweenNull = builder.between(0, null, 5);
+        Predicate between = builder.between(0, 1, 4);
+        Predicate andBetweens = PredicateBuilder.and(Arrays.asList(betweenNull, between));
+        assertThat(andBetweens).isInstanceOf(CompoundPredicate.class);
+        CompoundPredicate compound = (CompoundPredicate) andBetweens;
+        assertThat(compound.function()).isInstanceOf(And.class);
+        assertThat(compound.children()).hasSize(2);
+
+        // The null-bearing predicate retained by the merge still evaluates to false.
+        Predicate merged = PredicateBuilder.and(Arrays.asList(lteNull, gte));
+        GenericRow row = new GenericRow(1);
+        row.setField(0, 3);
+        assertThat(merged.test(row)).isFalse();
+    }
+
     @Test
     public void testTryRewriteBetweenPredicateBasic() {
         // Test basic case: AND(a>=1, a<=10, a is not null) should be rewritten to BETWEEN