[core] Move predicate API to paimon-api module

paimon-common/src/main/java/org/apache/paimon/predicate/And.java → paimon-api/src/main/java/org/apache/paimon/predicate/And.java

@@ -18,9 +18,6 @@
 
 package org.apache.paimon.predicate;
 
-import org.apache.paimon.data.InternalArray;
-import org.apache.paimon.data.InternalRow;
-
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
@@ -34,31 +31,6 @@ public class And extends CompoundPredicate.Function {
 
     private And() {}
 
-    @Override
-    public boolean test(InternalRow row, List<Predicate> children) {
-        for (Predicate child : children) {
-            if (!child.test(row)) {
-                return false;
-            }
-        }
-        return true;
-    }
-
-    @Override
-    public boolean test(
-            long rowCount,
-            InternalRow minValues,
-            InternalRow maxValues,
-            InternalArray 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<>();

paimon-common/src/main/java/org/apache/paimon/predicate/CastTransform.java → paimon-api/src/main/java/org/apache/paimon/predicate/CastTransform.java

@@ -18,47 +18,35 @@
 
 package org.apache.paimon.predicate;
 
-import org.apache.paimon.casting.CastExecutor;
-import org.apache.paimon.casting.CastExecutors;
-import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.types.DataType;
 
-import java.io.IOException;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
 
-import static org.apache.paimon.utils.InternalRowUtils.get;
-
 /** Transform that casts a field to a new type. */
 public class CastTransform implements Transform {
 
     private static final long serialVersionUID = 1L;
 
     private final FieldRef fieldRef;
     private final DataType type;
-    private transient CastExecutor<Object, Object> cast;
 
-    private CastTransform(FieldRef fieldRef, DataType type, CastExecutor<Object, Object> cast) {
+    private CastTransform(FieldRef fieldRef, DataType type) {
         this.fieldRef = fieldRef;
         this.type = type;
-        this.cast = cast;
+    }
+
+    public static CastTransform of(FieldRef fieldRef, DataType type) {
+        return new CastTransform(fieldRef, type);
     }
 
     public static Optional<Transform> tryCreate(FieldRef fieldRef, DataType type) {
         if (fieldRef.type().equals(type)) {
             return Optional.of(new FieldTransform(fieldRef));
         }
-
-        @SuppressWarnings("unchecked")
-        CastExecutor<Object, Object> cast =
-                (CastExecutor<Object, Object>) CastExecutors.resolve(fieldRef.type(), type);
-        if (cast == null) {
-            return Optional.empty();
-        } else {
-            return Optional.of(new CastTransform(fieldRef, type, cast));
-        }
+        return Optional.of(new CastTransform(fieldRef, type));
     }
 
     @Override
@@ -71,15 +59,10 @@ public DataType outputType() {
         return type;
     }
 
-    @Override
-    public Object transform(InternalRow row) {
-        return cast.cast(get(row, fieldRef.index(), fieldRef.type()));
-    }
-
     @Override
     public Transform copyWithNewInputs(List<Object> inputs) {
         assert inputs.size() == 1;
-        return new CastTransform((FieldRef) inputs.get(0), type, cast);
+        return new CastTransform((FieldRef) inputs.get(0), type);
     }
 
     @Override
@@ -100,13 +83,4 @@ public int hashCode() {
     public String toString() {
         return "CAST( " + fieldRef + " AS " + type + ")";
     }
-
-    private void readObject(java.io.ObjectInputStream in)
-            throws IOException, ClassNotFoundException {
-        in.defaultReadObject();
-        @SuppressWarnings("unchecked")
-        CastExecutor<Object, Object> resolved =
-                (CastExecutor<Object, Object>) CastExecutors.resolve(fieldRef.type(), type);
-        this.cast = resolved;
-    }
 }

paimon-common/src/main/java/org/apache/paimon/predicate/CompoundPredicate.java → paimon-api/src/main/java/org/apache/paimon/predicate/CompoundPredicate.java

@@ -18,9 +18,6 @@
 
 package org.apache.paimon.predicate;
 
-import org.apache.paimon.data.InternalArray;
-import org.apache.paimon.data.InternalRow;
-
 import java.io.Serializable;
 import java.util.List;
 import java.util.Objects;
@@ -48,17 +45,6 @@ public List<Predicate> children() {
         return children;
     }
 
-    @Override
-    public boolean test(InternalRow row) {
-        return function.test(row, children);
-    }
-
-    @Override
-    public boolean test(
-            long rowCount, InternalRow minValues, InternalRow maxValues, InternalArray nullCounts) {
-        return function.test(rowCount, minValues, maxValues, nullCounts, children);
-    }
-
     @Override
     public Optional<Predicate> negate() {
         return function.negate(children);
@@ -91,15 +77,6 @@ public String toString() {
     /** Evaluate the predicate result based on multiple {@link Predicate}s. */
     public abstract static class Function implements Serializable {
 
-        public abstract boolean test(InternalRow row, List<Predicate> children);
-
-        public abstract boolean test(
-                long rowCount,
-                InternalRow minValues,
-                InternalRow maxValues,
-                InternalArray nullCounts,
-                List<Predicate> children);
-
         public abstract Optional<Predicate> negate(List<Predicate> children);
 
         public abstract <T> T visit(FunctionVisitor<T> visitor, List<T> children);

paimon-common/src/main/java/org/apache/paimon/predicate/ConcatTransform.java → paimon-api/src/main/java/org/apache/paimon/predicate/ConcatTransform.java

@@ -18,8 +18,6 @@
 
 package org.apache.paimon.predicate;
 
-import org.apache.paimon.data.BinaryString;
-
 import java.util.List;
 
 /** Concat {@link Transform}. */
@@ -31,11 +29,6 @@ public ConcatTransform(List<Object> inputs) {
         super(inputs);
     }
 
-    @Override
-    public BinaryString transform(List<BinaryString> inputs) {
-        return BinaryString.concat(inputs);
-    }
-
     @Override
     public Transform copyWithNewInputs(List<Object> inputs) {
         return new ConcatTransform(inputs);

paimon-common/src/main/java/org/apache/paimon/predicate/ConcatWsTransform.java → paimon-api/src/main/java/org/apache/paimon/predicate/ConcatWsTransform.java

@@ -18,8 +18,6 @@
 
 package org.apache.paimon.predicate;
 
-import org.apache.paimon.data.BinaryString;
-
 import java.util.List;
 
 import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -34,12 +32,6 @@ public ConcatWsTransform(List<Object> inputs) {
         checkArgument(inputs.size() >= 2);
     }
 
-    @Override
-    public BinaryString transform(List<BinaryString> inputs) {
-        BinaryString separator = inputs.get(0);
-        return BinaryString.concatWs(separator, inputs.subList(1, inputs.size()));
-    }
-
     @Override
     public Transform copyWithNewInputs(List<Object> inputs) {
         return new ConcatWsTransform(inputs);

paimon-common/src/main/java/org/apache/paimon/predicate/Contains.java → paimon-api/src/main/java/org/apache/paimon/predicate/Contains.java

@@ -18,9 +18,6 @@
 
 package org.apache.paimon.predicate;
 
-import org.apache.paimon.data.BinaryString;
-import org.apache.paimon.types.DataType;
-
 import java.util.List;
 import java.util.Optional;
 
@@ -31,23 +28,6 @@ public class Contains extends NullFalseLeafBinaryFunction {
 
     private Contains() {}
 
-    @Override
-    public boolean test(DataType type, Object field, Object patternLiteral) {
-        BinaryString fieldString = (BinaryString) field;
-        return fieldString.contains((BinaryString) patternLiteral);
-    }
-
-    @Override
-    public boolean test(
-            DataType type,
-            long rowCount,
-            Object min,
-            Object max,
-            Long nullCount,
-            Object patternLiteral) {
-        return true;
-    }
-
     @Override
     public Optional<LeafFunction> negate() {
         return Optional.empty();

paimon-common/src/main/java/org/apache/paimon/predicate/EndsWith.java → paimon-api/src/main/java/org/apache/paimon/predicate/EndsWith.java

@@ -18,9 +18,6 @@
 
 package org.apache.paimon.predicate;
 
-import org.apache.paimon.data.BinaryString;
-import org.apache.paimon.types.DataType;
-
 import java.util.List;
 import java.util.Optional;
 
@@ -34,23 +31,6 @@ public class EndsWith extends NullFalseLeafBinaryFunction {
 
     private EndsWith() {}
 
-    @Override
-    public boolean test(DataType type, Object field, Object patternLiteral) {
-        BinaryString fieldString = (BinaryString) field;
-        return fieldString.endsWith((BinaryString) patternLiteral);
-    }
-
-    @Override
-    public boolean test(
-            DataType type,
-            long rowCount,
-            Object min,
-            Object max,
-            Long nullCount,
-            Object patternLiteral) {
-        return true;
-    }
-
     @Override
     public Optional<LeafFunction> negate() {
         return Optional.empty();

paimon-common/src/main/java/org/apache/paimon/predicate/Equal.java → paimon-api/src/main/java/org/apache/paimon/predicate/Equal.java

@@ -18,31 +18,16 @@
 
 package org.apache.paimon.predicate;
 
-import org.apache.paimon.types.DataType;
-
 import java.util.List;
 import java.util.Optional;
 
-import static org.apache.paimon.predicate.CompareUtils.compareLiteral;
-
 /** A {@link NullFalseLeafBinaryFunction} to eval equal. */
 public class Equal extends NullFalseLeafBinaryFunction {
 
     public static final Equal INSTANCE = new Equal();
 
     private Equal() {}
 
-    @Override
-    public boolean test(DataType type, Object field, Object literal) {
-        return compareLiteral(type, literal, field) == 0;
-    }
-
-    @Override
-    public boolean test(
-            DataType type, long rowCount, Object min, Object max, Long nullCount, Object literal) {
-        return compareLiteral(type, literal, min) >= 0 && compareLiteral(type, literal, max) <= 0;
-    }
-
     @Override
     public Optional<LeafFunction> negate() {
         return Optional.of(NotEqual.INSTANCE);

paimon-common/src/main/java/org/apache/paimon/predicate/FieldTransform.java → paimon-api/src/main/java/org/apache/paimon/predicate/FieldTransform.java

@@ -18,15 +18,12 @@
 
 package org.apache.paimon.predicate;
 
-import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.types.DataType;
 
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 
-import static org.apache.paimon.utils.InternalRowUtils.get;
-
 /** Transform that extracts a field from a row. */
 public class FieldTransform implements Transform {
 
@@ -52,11 +49,6 @@ public DataType outputType() {
         return fieldRef.type();
     }
 
-    @Override
-    public Object transform(InternalRow row) {
-        return get(row, fieldRef.index(), fieldRef.type());
-    }
-
     @Override
     public Transform copyWithNewInputs(List<Object> inputs) {
         assert inputs.size() == 1;

paimon-common/src/main/java/org/apache/paimon/predicate/GreaterOrEqual.java → paimon-api/src/main/java/org/apache/paimon/predicate/GreaterOrEqual.java

@@ -18,31 +18,16 @@
 
 package org.apache.paimon.predicate;
 
-import org.apache.paimon.types.DataType;
-
 import java.util.List;
 import java.util.Optional;
 
-import static org.apache.paimon.predicate.CompareUtils.compareLiteral;
-
 /** A {@link NullFalseLeafBinaryFunction} to eval greater or equal. */
 public class GreaterOrEqual extends NullFalseLeafBinaryFunction {
 
     public static final GreaterOrEqual INSTANCE = new GreaterOrEqual();
 
     private GreaterOrEqual() {}
 
-    @Override
-    public boolean test(DataType type, Object field, Object literal) {
-        return compareLiteral(type, literal, field) <= 0;
-    }
-
-    @Override
-    public boolean test(
-            DataType type, long rowCount, Object min, Object max, Long nullCount, Object literal) {
-        return compareLiteral(type, literal, max) <= 0;
-    }
-
     @Override
     public Optional<LeafFunction> negate() {
         return Optional.of(LessThan.INSTANCE);