[core][spark] Support push down transform predicate

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

@@ -35,4 +35,9 @@ public ConcatTransform(List<Object> inputs) {
     public BinaryString transform(List<BinaryString> inputs) {
         return BinaryString.concat(inputs);
     }
+
+    @Override
+    public Transform withNewInputs(List<Object> inputs) {
+        return new ConcatTransform(inputs);
+    }
 }

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

@@ -39,4 +39,9 @@ public BinaryString transform(List<BinaryString> inputs) {
         BinaryString separator = inputs.get(0);
         return BinaryString.concatWs(separator, inputs.subList(1, inputs.size()));
     }
+
+    @Override
+    public Transform withNewInputs(List<Object> inputs) {
+        return new ConcatWsTransform(inputs);
+    }
 }

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

@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+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 {
+
+    private final FieldRef fieldRef;
+
+    public FieldTransform(FieldRef fieldRef) {
+        this.fieldRef = fieldRef;
+    }
+
+    public FieldRef fieldRef() {
+        return fieldRef;
+    }
+
+    @Override
+    public List<Object> inputs() {
+        return Collections.singletonList(fieldRef);
+    }
+
+    @Override
+    public DataType outputType() {
+        return fieldRef.type();
+    }
+
+    @Override
+    public Object transform(InternalRow row) {
+        return get(row, fieldRef.index(), fieldRef.type());
+    }
+
+    @Override
+    public Transform withNewInputs(List<Object> inputs) {
+        assert inputs.size() == 1;
+        return new FieldTransform((FieldRef) inputs.get(0));
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        FieldTransform that = (FieldTransform) o;
+        return Objects.equals(fieldRef, that.fieldRef);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(fieldRef);
+    }
+}

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

@@ -30,76 +30,70 @@
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
 
 import static org.apache.paimon.utils.InternalRowUtils.get;
 
 /** Leaf node of a {@link Predicate} tree. Compares a field in the row with literals. */
-public class LeafPredicate implements Predicate {
+public class LeafPredicate extends TransformPredicate {
 
     private static final long serialVersionUID = 1L;
 
-    private final LeafFunction function;
-    private final DataType type;
-    private final int fieldIndex;
-    private final String fieldName;
-
-    private transient List<Object> literals;
-
     public LeafPredicate(
             LeafFunction function,
             DataType type,
             int fieldIndex,
             String fieldName,
             List<Object> literals) {
-        this.function = function;
-        this.type = type;
-        this.fieldIndex = fieldIndex;
-        this.fieldName = fieldName;
-        this.literals = literals;
+        this(new FieldTransform(new FieldRef(fieldIndex, fieldName, type)), function, literals);
+    }
+
+    public LeafPredicate(
+            FieldTransform fieldTransform, LeafFunction function, List<Object> literals) {
+        super(fieldTransform, function, literals);
     }
 
     public LeafFunction function() {
         return function;
     }
 
     public DataType type() {
-        return type;
+        return fieldRef().type();
     }
 
     public int index() {
-        return fieldIndex;
+        return fieldRef().index();
     }
 
     public String fieldName() {
-        return fieldName;
+        return fieldRef().name();
+    }
+
+    public List<String> fieldNames() {
+        return Collections.singletonList(fieldRef().name());
     }
 
     public FieldRef fieldRef() {
-        return new FieldRef(fieldIndex, fieldName, type);
+        return ((FieldTransform) transform).fieldRef();
     }
 
     public List<Object> literals() {
         return literals;
     }
 
     public LeafPredicate copyWithNewIndex(int fieldIndex) {
-        return new LeafPredicate(function, type, fieldIndex, fieldName, literals);
-    }
-
-    @Override
-    public boolean test(InternalRow row) {
-        return function.test(type, get(row, fieldIndex, type), literals);
+        return new LeafPredicate(function, type(), fieldIndex, fieldName(), literals);
     }
 
     @Override
     public boolean test(
             long rowCount, InternalRow minValues, InternalRow maxValues, InternalArray nullCounts) {
-        Object min = get(minValues, fieldIndex, type);
-        Object max = get(maxValues, fieldIndex, type);
-        Long nullCount = nullCounts.isNullAt(fieldIndex) ? null : nullCounts.getLong(fieldIndex);
+        Object min = get(minValues, index(), type());
+        Object max = get(maxValues, index(), type());
+        Long nullCount = nullCounts.isNullAt(index()) ? null : nullCounts.getLong(index());
         if (nullCount == null || rowCount != nullCount) {
             // not all null
             // min or max is null
@@ -108,41 +102,20 @@ public boolean test(
                 return true;
             }
         }
-        return function.test(type, rowCount, min, max, nullCount, literals);
+        return function.test(type(), rowCount, min, max, nullCount, literals);
     }
 
     @Override
     public Optional<Predicate> negate() {
         return function.negate()
-                .map(negate -> new LeafPredicate(negate, type, fieldIndex, fieldName, literals));
+                .map(negate -> new LeafPredicate(negate, type(), index(), fieldName(), literals));
     }
 
     @Override
     public <T> T visit(PredicateVisitor<T> visitor) {
         return visitor.visit(this);
     }
 
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) {
-            return true;
-        }
-        if (o == null || getClass() != o.getClass()) {
-            return false;
-        }
-        LeafPredicate that = (LeafPredicate) o;
-        return fieldIndex == that.fieldIndex
-                && Objects.equals(fieldName, that.fieldName)
-                && Objects.equals(function, that.function)
-                && Objects.equals(type, that.type)
-                && Objects.equals(literals, that.literals);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(function, type, fieldIndex, fieldName, literals);
-    }
-
     @Override
     public String toString() {
         String literalsStr;
@@ -154,13 +127,13 @@ public String toString() {
             literalsStr = literals.toString();
         }
         return literalsStr.isEmpty()
-                ? function + "(" + fieldName + ")"
-                : function + "(" + fieldName + ", " + literalsStr + ")";
+                ? function + "(" + fieldName() + ")"
+                : function + "(" + fieldName() + ", " + literalsStr + ")";
     }
 
     private ListSerializer<Object> objectsSerializer() {
         return new ListSerializer<>(
-                NullableSerializer.wrapIfNullIsNotSupported(InternalSerializers.create(type)));
+                NullableSerializer.wrapIfNullIsNotSupported(InternalSerializers.create(type())));
     }
 
     private void writeObject(ObjectOutputStream out) throws IOException {