[core] Unify TransformPredicate and LeafPredicate to one class

Author: JingsongLi

paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexPredicate.java

@@ -31,7 +31,6 @@
 import org.apache.paimon.predicate.PredicateVisitor;
 import org.apache.paimon.predicate.SortValue;
 import org.apache.paimon.predicate.TopN;
-import org.apache.paimon.predicate.TransformPredicate;
 import org.apache.paimon.types.RowType;
 
 import org.slf4j.Logger;
@@ -42,11 +41,11 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 import static org.apache.paimon.fileindex.FileIndexResult.REMAIN;
@@ -121,7 +120,7 @@ private Set<String> getRequiredNames(Predicate filePredicate) {
 
                     @Override
                     public Set<String> visit(LeafPredicate predicate) {
-                        return Collections.singleton(predicate.fieldName());
+                        return new HashSet<>(predicate.fieldNames());
                     }
 
                     @Override
@@ -133,11 +132,6 @@ public Set<String> visit(CompoundPredicate predicate) {
                         }
                         return result;
                     }
-
-                    @Override
-                    public Set<String> visit(TransformPredicate predicate) {
-                        return new HashSet<>(predicate.fieldNames());
-                    }
                 });
     }
 
@@ -161,10 +155,14 @@ public FileIndexResult test(Predicate predicate) {
 
         @Override
         public FileIndexResult visit(LeafPredicate predicate) {
+            Optional<FieldRef> fieldRefOptional = predicate.fieldRefOptional();
+            if (!fieldRefOptional.isPresent()) {
+                return REMAIN;
+            }
+
             FileIndexResult compoundResult = REMAIN;
-            FieldRef fieldRef =
-                    new FieldRef(predicate.index(), predicate.fieldName(), predicate.type());
-            for (FileIndexReader fileIndexReader : columnIndexReaders.get(predicate.fieldName())) {
+            FieldRef fieldRef = fieldRefOptional.get();
+            for (FileIndexReader fileIndexReader : columnIndexReaders.get(fieldRef.name())) {
                 compoundResult =
                         compoundResult.and(
                                 predicate
@@ -204,10 +202,5 @@ public FileIndexResult visit(CompoundPredicate predicate) {
                 return compoundResult == null ? REMAIN : compoundResult;
             }
         }
-
-        @Override
-        public FileIndexResult visit(TransformPredicate predicate) {
-            return REMAIN;
-        }
     }
 }

paimon-common/src/main/java/org/apache/paimon/fileindex/FileIndexReader.java

@@ -20,8 +20,8 @@
 
 import org.apache.paimon.predicate.FieldRef;
 import org.apache.paimon.predicate.FunctionVisitor;
+import org.apache.paimon.predicate.LeafPredicate;
 import org.apache.paimon.predicate.TopN;
-import org.apache.paimon.predicate.TransformPredicate;
 
 import java.util.List;
 
@@ -132,7 +132,7 @@ public FileIndexResult visitTopN(TopN topN, FileIndexResult result) {
     }
 
     @Override
-    public FileIndexResult visit(TransformPredicate predicate) {
+    public FileIndexResult visitNonFieldLeaf(LeafPredicate predicate) {
         return REMAIN;
     }
 }

paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexEvaluator.java

@@ -24,7 +24,6 @@
 import org.apache.paimon.predicate.Or;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.predicate.PredicateVisitor;
-import org.apache.paimon.predicate.TransformPredicate;
 import org.apache.paimon.predicate.VectorSearch;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.IOUtils;
@@ -90,10 +89,13 @@ public Optional<GlobalIndexResult> evaluate(
 
     @Override
     public Optional<GlobalIndexResult> visit(LeafPredicate predicate) {
+        Optional<FieldRef> fieldRefOptional = predicate.fieldRefOptional();
+        if (!fieldRefOptional.isPresent()) {
+            return Optional.empty();
+        }
         Optional<GlobalIndexResult> compoundResult = Optional.empty();
-        FieldRef fieldRef =
-                new FieldRef(predicate.index(), predicate.fieldName(), predicate.type());
-        int fieldId = rowType.getField(predicate.fieldName()).id();
+        FieldRef fieldRef = fieldRefOptional.get();
+        int fieldId = rowType.getField(fieldRef.name()).id();
         Collection<GlobalIndexReader> readers =
                 indexReadersCache.computeIfAbsent(fieldId, readersFunction::apply);
         for (GlobalIndexReader fileIndexReader : readers) {
@@ -158,11 +160,6 @@ public Optional<GlobalIndexResult> visit(CompoundPredicate predicate) {
         }
     }
 
-    @Override
-    public Optional<GlobalIndexResult> visit(TransformPredicate predicate) {
-        return Optional.empty();
-    }
-
     public void close() {
         IOUtils.closeAllQuietly(
                 indexReadersCache.values().stream()

paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java

@@ -19,7 +19,7 @@
 package org.apache.paimon.globalindex;
 
 import org.apache.paimon.predicate.FunctionVisitor;
-import org.apache.paimon.predicate.TransformPredicate;
+import org.apache.paimon.predicate.LeafPredicate;
 import org.apache.paimon.predicate.VectorSearch;
 
 import java.io.Closeable;
@@ -40,7 +40,7 @@ default Optional<GlobalIndexResult> visitOr(List<Optional<GlobalIndexResult>> ch
     }
 
     @Override
-    default Optional<GlobalIndexResult> visit(TransformPredicate predicate) {
+    default Optional<GlobalIndexResult> visitNonFieldLeaf(LeafPredicate predicate) {
         throw new UnsupportedOperationException();
     }
 

paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeFileMetaSelector.java

@@ -22,7 +22,7 @@
 import org.apache.paimon.memory.MemorySlice;
 import org.apache.paimon.predicate.FieldRef;
 import org.apache.paimon.predicate.FunctionVisitor;
-import org.apache.paimon.predicate.TransformPredicate;
+import org.apache.paimon.predicate.LeafPredicate;
 import org.apache.paimon.utils.Pair;
 
 import java.util.ArrayList;
@@ -212,7 +212,7 @@ public Optional<List<GlobalIndexIOMeta>> visitOr(
     }
 
     @Override
-    public Optional<List<GlobalIndexIOMeta>> visit(TransformPredicate predicate) {
+    public Optional<List<GlobalIndexIOMeta>> visitNonFieldLeaf(LeafPredicate predicate) {
         return Optional.empty();
     }
 

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

@@ -26,6 +26,7 @@
 import java.util.Objects;
 
 import static org.apache.paimon.utils.InternalRowUtils.get;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
 
 /** Transform that extracts a field from a row. */
 public class FieldTransform implements Transform {
@@ -59,7 +60,7 @@ public Object transform(InternalRow row) {
 
     @Override
     public Transform copyWithNewInputs(List<Object> inputs) {
-        assert inputs.size() == 1;
+        checkArgument(inputs.size() == 1);
         return new FieldTransform((FieldRef) inputs.get(0));
     }
 

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

@@ -19,16 +19,23 @@
 package org.apache.paimon.predicate;
 
 import java.util.List;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 /** A {@link PredicateVisitor} to visit functions. */
 public interface FunctionVisitor<T> extends PredicateVisitor<T> {
 
     @Override
     default T visit(LeafPredicate predicate) {
-        return predicate.function().visit(this, predicate.fieldRef(), predicate.literals());
+        Optional<FieldRef> fieldRef = predicate.fieldRefOptional();
+        if (!fieldRef.isPresent()) {
+            return visitNonFieldLeaf(predicate);
+        }
+        return predicate.function().visit(this, fieldRef.get(), predicate.literals());
     }
 
+    T visitNonFieldLeaf(LeafPredicate leafPredicate);
+
     @Override
     default T visit(CompoundPredicate predicate) {
         return predicate