[core] Introduce transform predicate for complex predicate

Author: JingsongLi

paimon-common/src/main/java/org/apache/paimon/data/BinaryString.java

@@ -1012,4 +1012,89 @@ public static int decodeUTF8Strict(MemorySegment segment, int sp, int len, char[
     public static String defaultDecodeUTF8(byte[] bytes, int offset, int len) {
         return new String(bytes, offset, len, StandardCharsets.UTF_8);
     }
+
+    /**
+     * Concatenates input strings together into a single string. Returns NULL if any argument is
+     * NULL.
+     */
+    public static BinaryString concat(Iterable<BinaryString> inputs) {
+        // Compute the total length of the result.
+        int totalLength = 0;
+        for (BinaryString input : inputs) {
+            if (input == null) {
+                return null;
+            }
+
+            totalLength += input.getSizeInBytes();
+        }
+
+        // Allocate a new byte array, and copy the inputs one by one into it.
+        final byte[] result = new byte[totalLength];
+        int offset = 0;
+        for (BinaryString input : inputs) {
+            if (input != null) {
+                int len = input.getSizeInBytes();
+                MemorySegmentUtils.copyToBytes(
+                        input.getSegments(), input.getOffset(), result, offset, len);
+                offset += len;
+            }
+        }
+        return fromBytes(result);
+    }
+
+    /**
+     * Concatenates input strings together into a single string using the separator. Returns NULL If
+     * the separator is NULL.
+     *
+     * <p>Note: CONCAT_WS() does not skip any empty strings, however it does skip any NULL values
+     * after the separator. For example, concat_ws(",", "a", null, "c") would yield "a,c".
+     */
+    public static BinaryString concatWs(BinaryString separator, Iterable<BinaryString> inputs) {
+        if (null == separator) {
+            return null;
+        }
+
+        int numInputBytes = 0; // total number of bytes from the inputs
+        int numInputs = 0; // number of non-null inputs
+        for (BinaryString input : inputs) {
+            if (input != null) {
+                numInputBytes += input.getSizeInBytes();
+                numInputs++;
+            }
+        }
+
+        if (numInputs == 0) {
+            // Return an empty string if there is no input, or all the inputs are null.
+            return EMPTY_UTF8;
+        }
+
+        // Allocate a new byte array, and copy the inputs one by one into it.
+        // The size of the new array is the size of all inputs, plus the separators.
+        final byte[] result =
+                new byte[numInputBytes + (numInputs - 1) * separator.getSizeInBytes()];
+        int offset = 0;
+
+        int j = 0;
+        for (BinaryString input : inputs) {
+            if (input != null) {
+                int len = input.getSizeInBytes();
+                MemorySegmentUtils.copyToBytes(
+                        input.getSegments(), input.getOffset(), result, offset, len);
+                offset += len;
+
+                j++;
+                // Add separator if this is not the last input.
+                if (j < numInputs) {
+                    MemorySegmentUtils.copyToBytes(
+                            separator.getSegments(),
+                            separator.getOffset(),
+                            result,
+                            offset,
+                            separator.getSizeInBytes());
+                    offset += separator.getSizeInBytes();
+                }
+            }
+        }
+        return fromBytes(result);
+    }
 }

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

@@ -31,6 +31,7 @@
 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;
@@ -131,6 +132,11 @@ public Set<String> visit(CompoundPredicate predicate) {
                         }
                         return names;
                     }
+
+                    @Override
+                    public Set<String> visit(TransformPredicate predicate) {
+                        throw new UnsupportedOperationException();
+                    }
                 });
     }
 
@@ -197,5 +203,10 @@ public FileIndexResult visit(CompoundPredicate predicate) {
                 return compoundResult == null ? REMAIN : compoundResult;
             }
         }
+
+        @Override
+        public FileIndexResult visit(TransformPredicate predicate) {
+            throw new UnsupportedOperationException();
+        }
     }
 }

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

@@ -21,6 +21,7 @@
 import org.apache.paimon.predicate.FieldRef;
 import org.apache.paimon.predicate.FunctionVisitor;
 import org.apache.paimon.predicate.TopN;
+import org.apache.paimon.predicate.TransformPredicate;
 
 import java.util.List;
 
@@ -124,4 +125,9 @@ public FileIndexResult visitOr(List<FileIndexResult> children) {
     public FileIndexResult visitTopN(TopN topN, FileIndexResult result) {
         return REMAIN;
     }
+
+    @Override
+    public FileIndexResult visit(TransformPredicate predicate) {
+        return REMAIN;
+    }
 }

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

@@ -0,0 +1,38 @@
+/*
+ * 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.BinaryString;
+
+import java.util.List;
+
+/** Concat {@link Transform}. */
+public class ConcatTransform extends StringTransform {
+
+    private static final long serialVersionUID = 1L;
+
+    public ConcatTransform(List<Object> inputs) {
+        super(inputs);
+    }
+
+    @Override
+    public BinaryString transform(List<BinaryString> inputs) {
+        return BinaryString.concat(inputs);
+    }
+}

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

@@ -0,0 +1,42 @@
+/*
+ * 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.BinaryString;
+
+import java.util.List;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** ConcatWs {@link Transform}. */
+public class ConcatWsTransform extends StringTransform {
+
+    private static final long serialVersionUID = 1L;
+
+    public ConcatWsTransform(List<Object> inputs) {
+        super(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()));
+    }
+}

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

@@ -41,4 +41,9 @@ public Map<String, LeafPredicate> visit(CompoundPredicate predicate) {
         }
         return Collections.emptyMap();
     }
+
+    @Override
+    public Map<String, LeafPredicate> visit(TransformPredicate predicate) {
+        throw new UnsupportedOperationException();
+    }
 }

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

@@ -121,4 +121,9 @@ public Boolean visitAnd(List<Boolean> children) {
     public Boolean visitOr(List<Boolean> children) {
         return false;
     }
+
+    @Override
+    public Boolean visit(TransformPredicate predicate) {
+        return false;
+    }
 }

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

@@ -45,4 +45,17 @@ public Boolean visit(CompoundPredicate predicate) {
         }
         return true;
     }
+
+    @Override
+    public Boolean visit(TransformPredicate predicate) {
+        Transform transform = predicate.transform();
+        for (Object input : transform.inputs()) {
+            if (input instanceof FieldRef) {
+                if (!partitionKeys.contains(((FieldRef) input).name())) {
+                    return false;
+                }
+            }
+        }
+        return true;
+    }
 }

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

@@ -63,4 +63,9 @@ public Optional<Predicate> visit(CompoundPredicate predicate) {
         }
         return Optional.of(new CompoundPredicate(predicate.function(), converted));
     }
+
+    @Override
+    public Optional<Predicate> visit(TransformPredicate predicate) {
+        throw new UnsupportedOperationException("TODO");
+    }
 }

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

@@ -24,4 +24,6 @@ public interface PredicateVisitor<T> {
     T visit(LeafPredicate predicate);
 
     T visit(CompoundPredicate predicate);
+
+    T visit(TransformPredicate predicate);
 }