[common] Support statistic-based Predicate interface

wuchong · wuchong · commit 74ab1f130a00 · 2025-08-07T14:14:19.000+08:00
diff --git a/LICENSE b/LICENSE
@@ -359,6 +359,33 @@ Apache Kafka
 ./fluss-server/src/main/java/com/alibaba/fluss/server/utils/timer/TimingWheel.java
 
 Apache Paimon
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/And.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/CompareUtils.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/CompoundPredicate.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/Contains.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/EndsWith.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/Equal.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/FieldRef.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/FunctionVisitor.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/GreaterOrEqual.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/GreaterThan.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/In.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/IsNotNull.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/IsNull.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafFunction.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafPredicate.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafUnaryFunction.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/LessOrEqual.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/LessThan.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/NotEqual.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/NotIn.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/NullFalseLeafBinaryFunction.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/Or.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/PartitionPredicateVisitor.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/Predicate.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/PredicateBuilder.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/PredicateVisitor.java
+./fluss-common/src/main/java/com/alibaba/fluss/predicate/StartsWith.java
 ./fluss-common/src/main/java/com/alibaba/fluss/row/encode/paimon/PaimonBinaryRowWriter.java
 ./fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/lakehouse/paimon/reader/PaimonSnapshotScanner.java
 
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/And.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/And.java
@@ -46,6 +46,21 @@ public boolean test(InternalRow row, List<Predicate> children) {
         return true;
     }
 
+    @Override
+    public boolean test(
+            long rowCount,
+            InternalRow minValues,
+            InternalRow maxValues,
+            Long[] nullCounts,
+            List<Predicate> children) {
+        for (Predicate child : children) {
+            if (!child.test(rowCount, minValues, maxValues, nullCounts)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     @Override
     public Optional<Predicate> negate(List<Predicate> children) {
         List<Predicate> negatedChildren = new ArrayList<>();
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/CompareUtils.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/CompareUtils.java
@@ -17,7 +17,6 @@
 
 package com.alibaba.fluss.predicate;
 
-import com.alibaba.fluss.row.BinaryString;
 import com.alibaba.fluss.types.DataType;
 
 import static java.lang.Math.min;
@@ -30,16 +29,9 @@
 public class CompareUtils {
     private CompareUtils() {}
 
+    @SuppressWarnings("unchecked")
     public static int compareLiteral(DataType type, Object v1, Object v2) {
         if (v1 instanceof Comparable) {
-            // because BinaryString can not serialize so v1 or v2 may be BinaryString convert to
-            // String for compare
-            if (v1 instanceof BinaryString) {
-                v1 = ((BinaryString) v1).toString();
-            }
-            if (v2 instanceof BinaryString) {
-                v2 = ((BinaryString) v2).toString();
-            }
             return ((Comparable<Object>) v1).compareTo(v2);
         } else if (v1 instanceof byte[]) {
             return compare((byte[]) v1, (byte[]) v2);
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/CompoundPredicate.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/CompoundPredicate.java
@@ -34,6 +34,7 @@
  */
 public class CompoundPredicate implements Predicate {
 
+    private static final long serialVersionUID = 1L;
     private final Function function;
     private final List<Predicate> children;
 
@@ -55,6 +56,12 @@ public boolean test(InternalRow row) {
         return function.test(row, children);
     }
 
+    @Override
+    public boolean test(
+            long rowCount, InternalRow minValues, InternalRow maxValues, Long[] nullCounts) {
+        return function.test(rowCount, minValues, maxValues, nullCounts, children);
+    }
+
     @Override
     public Optional<Predicate> negate() {
         return function.negate(children);
@@ -87,8 +94,17 @@ public String toString() {
     /** Evaluate the predicate result based on multiple {@link Predicate}s. */
     public abstract static class Function implements Serializable {
 
+        private static final long serialVersionUID = 1L;
+
         public abstract boolean test(InternalRow row, List<Predicate> children);
 
+        public abstract boolean test(
+                long rowCount,
+                InternalRow minValues,
+                InternalRow maxValues,
+                Long[] nullCounts,
+                List<Predicate> children);
+
         public abstract Optional<Predicate> negate(List<Predicate> children);
 
         public abstract <T> T visit(FunctionVisitor<T> visitor, List<T> children);
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/Contains.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/Contains.java
@@ -29,6 +29,8 @@
 /** A {@link NullFalseLeafBinaryFunction} to evaluate {@code filter like '%abc%'}. */
 public class Contains extends NullFalseLeafBinaryFunction {
 
+    private static final long serialVersionUID = 1L;
+
     public static final Contains INSTANCE = new Contains();
 
     private Contains() {}
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/EndsWith.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/EndsWith.java
@@ -32,6 +32,8 @@
  */
 public class EndsWith extends NullFalseLeafBinaryFunction {
 
+    private static final long serialVersionUID = 1L;
+
     public static final EndsWith INSTANCE = new EndsWith();
 
     private EndsWith() {}
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/Equal.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/Equal.java
@@ -31,6 +31,8 @@
 /** A {@link NullFalseLeafBinaryFunction} to eval equal. */
 public class Equal extends NullFalseLeafBinaryFunction {
 
+    private static final long serialVersionUID = 1L;
+
     public static final Equal INSTANCE = new Equal();
 
     private Equal() {}
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/FieldRef.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/FieldRef.java
@@ -29,7 +29,7 @@
 /** A reference to a field in an input. */
 public class FieldRef implements Serializable {
 
-    private static final long serialVersionUID = 4982103776651292199L;
+    private static final long serialVersionUID = 1L;
 
     private final int index;
     private final String name;
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/GreaterOrEqual.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/GreaterOrEqual.java
@@ -31,6 +31,8 @@
 /** A {@link NullFalseLeafBinaryFunction} to eval greater or equal. */
 public class GreaterOrEqual extends NullFalseLeafBinaryFunction {
 
+    private static final long serialVersionUID = 1L;
+
     public static final GreaterOrEqual INSTANCE = new GreaterOrEqual();
 
     private GreaterOrEqual() {}
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/GreaterThan.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/GreaterThan.java
@@ -31,6 +31,8 @@
 /** A {@link LeafFunction} to eval greater. */
 public class GreaterThan extends NullFalseLeafBinaryFunction {
 
+    private static final long serialVersionUID = 1L;
+
     public static final GreaterThan INSTANCE = new GreaterThan();
 
     private GreaterThan() {}
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/In.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/In.java
@@ -31,7 +31,7 @@
 /** A {@link LeafFunction} to eval in. */
 public class In extends LeafFunction {
 
-    private static final long serialVersionUID = -9115697441080586485L;
+    private static final long serialVersionUID = 1L;
 
     public static final In INSTANCE = new In();
 
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/IsNotNull.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/IsNotNull.java
@@ -29,6 +29,8 @@
 /** A {@link NullFalseLeafBinaryFunction} to eval is not null. */
 public class IsNotNull extends LeafUnaryFunction {
 
+    private static final long serialVersionUID = 1L;
+
     public static final IsNotNull INSTANCE = new IsNotNull();
 
     private IsNotNull() {}
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/IsNull.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/IsNull.java
@@ -29,6 +29,8 @@
 /** A {@link NullFalseLeafBinaryFunction} to eval is null. */
 public class IsNull extends LeafUnaryFunction {
 
+    private static final long serialVersionUID = 1L;
+
     public static final IsNull INSTANCE = new IsNull();
 
     private IsNull() {}
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafFunction.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafFunction.java
@@ -30,8 +30,29 @@
 /** Function to test a field with literals. */
 public abstract class LeafFunction implements Serializable {
 
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * Tests whether a field satisfies the condition based on the provided literals.
+     *
+     * @param type the data type of the field being tested
+     * @param field the value of the field to test
+     * @param literals the list of literals to test against the field
+     * @return true if the field satisfies the condition, false otherwise
+     */
     public abstract boolean test(DataType type, Object field, List<Object> literals);
 
+    /**
+     * Tests whether a set of rows satisfies the condition based on the provided statistics.
+     *
+     * @param type the data type of the field being tested
+     * @param rowCount the total number of rows
+     * @param min the minimum value of the field in the rows
+     * @param max the maximum value of the field in the rows
+     * @param nullCount the number of null values in the field, or null if unknown
+     * @param literals the literals to test against the field
+     * @return true if there is any row satisfies the condition, false otherwise
+     */
     public abstract boolean test(
             DataType type,
             long rowCount,
@@ -42,6 +63,9 @@ public abstract boolean test(
 
     public abstract Optional<LeafFunction> negate();
 
+    public abstract <T> T visit(
+            FunctionVisitor<T> visitor, FieldRef fieldRef, List<Object> literals);
+
     @Override
     public int hashCode() {
         return this.getClass().getName().hashCode();
@@ -55,9 +79,6 @@ public boolean equals(Object o) {
         return o != null && getClass() == o.getClass();
     }
 
-    public abstract <T> T visit(
-            FunctionVisitor<T> visitor, FieldRef fieldRef, List<Object> literals);
-
     @Override
     public String toString() {
         return getClass().getSimpleName();
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafPredicate.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafPredicate.java
@@ -34,14 +34,14 @@
 /** Leaf node of a {@link Predicate} tree. Compares a field in the row with literals. */
 public class LeafPredicate implements Predicate {
 
-    private static final long serialVersionUID = -9033842253303772188L;
+    private static final long serialVersionUID = 1L;
 
     private final LeafFunction function;
     private final DataType type;
     private final int fieldIndex;
     private final String fieldName;
 
-    private List<Object> literals;
+    private final List<Object> literals;
 
     public LeafPredicate(
             LeafFunction function,
@@ -89,6 +89,23 @@ public boolean test(InternalRow row) {
         return function.test(type, get(row, fieldIndex, type), literals);
     }
 
+    @Override
+    public boolean test(
+            long rowCount, InternalRow minValues, InternalRow maxValues, Long[] nullCounts) {
+        Object min = get(minValues, fieldIndex, type);
+        Object max = get(maxValues, fieldIndex, type);
+        Long nullCount = nullCounts != null ? nullCounts[fieldIndex] : null;
+        if (nullCount == null || rowCount != nullCount) {
+            // not all null
+            // min or max is null
+            // unknown stats
+            if (min == null || max == null) {
+                return true;
+            }
+        }
+        return function.test(type, rowCount, min, max, nullCount, literals);
+    }
+
     @Override
     public Optional<Predicate> negate() {
         return function.negate()
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafUnaryFunction.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafUnaryFunction.java
@@ -28,7 +28,7 @@
 /** Function to test a field. */
 public abstract class LeafUnaryFunction extends LeafFunction {
 
-    private static final long serialVersionUID = -155104972966998013L;
+    private static final long serialVersionUID = 1L;
 
     public abstract boolean test(DataType type, Object value);
 
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/LessOrEqual.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/LessOrEqual.java
@@ -31,6 +31,8 @@
 /** A {@link NullFalseLeafBinaryFunction} to eval less or equal. */
 public class LessOrEqual extends NullFalseLeafBinaryFunction {
 
+    private static final long serialVersionUID = 1L;
+
     public static final LessOrEqual INSTANCE = new LessOrEqual();
 
     private LessOrEqual() {}
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/LessThan.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/LessThan.java
@@ -31,6 +31,8 @@
 /** A {@link NullFalseLeafBinaryFunction} to eval less or equal. */
 public class LessThan extends NullFalseLeafBinaryFunction {
 
+    private static final long serialVersionUID = 1L;
+
     public static final LessThan INSTANCE = new LessThan();
 
     private LessThan() {}
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/NotEqual.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/NotEqual.java
@@ -31,6 +31,8 @@
 /** A {@link NullFalseLeafBinaryFunction} to eval not equal. */
 public class NotEqual extends NullFalseLeafBinaryFunction {
 
+    private static final long serialVersionUID = 1L;
+
     public static final NotEqual INSTANCE = new NotEqual();
 
     private NotEqual() {}
@@ -43,6 +45,8 @@ public boolean test(DataType type, Object field, Object literal) {
     @Override
     public boolean test(
             DataType type, long rowCount, Object min, Object max, Long nullCount, Object literal) {
+        // ony when max == min == literal, the result is false,
+        // otherwise, the row set MAY contain the literal.
         return compareLiteral(type, literal, min) != 0 || compareLiteral(type, literal, max) != 0;
     }
 
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/NotIn.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/NotIn.java
@@ -31,7 +31,7 @@
 /** A {@link LeafFunction} to eval not in. */
 public class NotIn extends LeafFunction {
 
-    private static final long serialVersionUID = 8953845894700582887L;
+    private static final long serialVersionUID = 1L;
     public static final NotIn INSTANCE = new NotIn();
 
     private NotIn() {}
@@ -62,6 +62,9 @@ public boolean test(
         }
         for (Object literal : literals) {
             if (literal == null
+                    // only if max == min == literal, the row set are all IN the literal, return
+                    // false; other cases, the row set MAY contain elements NOT IN the literal,
+                    // return true
                     || (compareLiteral(type, literal, min) == 0
                             && compareLiteral(type, literal, max) == 0)) {
                 return false;
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/NullFalseLeafBinaryFunction.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/NullFalseLeafBinaryFunction.java
@@ -28,7 +28,7 @@
 /** Function to test a field with a literal. */
 public abstract class NullFalseLeafBinaryFunction extends LeafFunction {
 
-    private static final long serialVersionUID = 5617091663961558170L;
+    private static final long serialVersionUID = 1L;
 
     public abstract boolean test(DataType type, Object field, Object literal);
 
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/Or.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/Or.java
@@ -30,7 +30,7 @@
 /** A {@link CompoundPredicate.Function} to eval or. */
 public class Or extends CompoundPredicate.Function {
 
-    private static final long serialVersionUID = -2110346319473699418L;
+    private static final long serialVersionUID = 1L;
 
     public static final Or INSTANCE = new Or();
 
@@ -46,6 +46,21 @@ public boolean test(InternalRow row, List<Predicate> children) {
         return false;
     }
 
+    @Override
+    public boolean test(
+            long rowCount,
+            InternalRow minValues,
+            InternalRow maxValues,
+            Long[] nullCounts,
+            List<Predicate> children) {
+        for (Predicate child : children) {
+            if (child.test(rowCount, minValues, maxValues, nullCounts)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     @Override
     public Optional<Predicate> negate(List<Predicate> children) {
         List<Predicate> negatedChildren = new ArrayList<>();
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/Predicate.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/Predicate.java
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/PredicateBuilder.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/PredicateBuilder.java
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/PredicateVisitor.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/PredicateVisitor.java
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/StartsWith.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/StartsWith.java
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/UnsupportedExpression.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/UnsupportedExpression.java
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/types/RowType.java b/fluss-common/src/main/java/com/alibaba/fluss/types/RowType.java
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/utils/BinaryStringUtils.java b/fluss-common/src/main/java/com/alibaba/fluss/utils/BinaryStringUtils.java
diff --git a/fluss-common/src/test/java/com/alibaba/fluss/predicate/PredicateBuilderTest.java b/fluss-common/src/test/java/com/alibaba/fluss/predicate/PredicateBuilderTest.java
diff --git a/fluss-common/src/test/java/com/alibaba/fluss/predicate/PredicateTest.java b/fluss-common/src/test/java/com/alibaba/fluss/predicate/PredicateTest.java
