[spark] Support push down StringContains

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

@@ -51,6 +51,11 @@ public FileIndexResult visitEndsWith(FieldRef fieldRef, Object literal) {
         return REMAIN;
     }
 
+    @Override
+    public FileIndexResult visitContains(FieldRef fieldRef, Object literal) {
+        return REMAIN;
+    }
+
     @Override
     public FileIndexResult visitLessThan(FieldRef fieldRef, Object literal) {
         return REMAIN;

paimon-common/src/main/java/org/apache/paimon/fileindex/empty/EmptyFileIndexReader.java

@@ -52,6 +52,11 @@ public FileIndexResult visitEndsWith(FieldRef fieldRef, Object literal) {
         return SKIP;
     }
 
+    @Override
+    public FileIndexResult visitContains(FieldRef fieldRef, Object literal) {
+        return SKIP;
+    }
+
     @Override
     public FileIndexResult visitLessThan(FieldRef fieldRef, Object literal) {
         return SKIP;

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

@@ -0,0 +1,60 @@
+/*
+ * 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 org.apache.paimon.types.DataType;
+
+import java.util.List;
+import java.util.Optional;
+
+/** A {@link NullFalseLeafBinaryFunction} to evaluate {@code filter like '%abc%'}. */
+public class Contains extends NullFalseLeafBinaryFunction {
+
+    public static final Contains INSTANCE = new Contains();
+
+    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();
+    }
+
+    @Override
+    public <T> T visit(FunctionVisitor<T> visitor, FieldRef fieldRef, List<Object> literals) {
+        return visitor.visitContains(fieldRef, literals.get(0));
+    }
+}

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

@@ -52,6 +52,8 @@ default T visit(CompoundPredicate predicate) {
 
     T visitEndsWith(FieldRef fieldRef, Object literal);
 
+    T visitContains(FieldRef fieldRef, Object literal);
+
     T visitLessThan(FieldRef fieldRef, Object literal);
 
     T visitGreaterOrEqual(FieldRef fieldRef, Object literal);

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

@@ -62,6 +62,11 @@ public Boolean visitEndsWith(FieldRef fieldRef, Object literal) {
         return false;
     }
 
+    @Override
+    public Boolean visitContains(FieldRef fieldRef, Object literal) {
+        return false;
+    }
+
     @Override
     public Boolean visitLessThan(FieldRef fieldRef, Object literal) {
         return false;

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

@@ -112,6 +112,10 @@ public Predicate endsWith(int idx, Object patternLiteral) {
         return leaf(EndsWith.INSTANCE, idx, patternLiteral);
     }
 
+    public Predicate contains(int idx, Object patternLiteral) {
+        return leaf(Contains.INSTANCE, idx, patternLiteral);
+    }
+
     public Predicate leaf(NullFalseLeafBinaryFunction function, int idx, Object literal) {
         DataField field = rowType.getFields().get(idx);
         return new LeafPredicate(function, field.type(), idx, field.name(), singletonList(literal));

paimon-format/src/main/java/org/apache/paimon/format/orc/filter/OrcPredicateFunctionVisitor.java

@@ -69,6 +69,11 @@ public Optional<OrcFilters.Predicate> visitEndsWith(FieldRef fieldRef, Object li
         return Optional.empty();
     }
 
+    @Override
+    public Optional<OrcFilters.Predicate> visitContains(FieldRef fieldRef, Object literal) {
+        return Optional.empty();
+    }
+
     @Override
     public Optional<OrcFilters.Predicate> visitLessThan(FieldRef fieldRef, Object literal) {
         return convertBinary(fieldRef, literal, OrcFilters.LessThan::new);

paimon-format/src/main/java/org/apache/parquet/filter2/predicate/ParquetFilters.java

@@ -146,6 +146,11 @@ public FilterPredicate visitEndsWith(FieldRef fieldRef, Object literal) {
             throw new UnsupportedOperationException();
         }
 
+        @Override
+        public FilterPredicate visitContains(FieldRef fieldRef, Object literal) {
+            throw new UnsupportedOperationException();
+        }
+
         @Override
         public FilterPredicate visitIn(FieldRef fieldRef, List<Object> literals) {
             throw new UnsupportedOperationException();

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkFilterConverter.java

@@ -36,6 +36,7 @@
 import org.apache.spark.sql.sources.LessThanOrEqual;
 import org.apache.spark.sql.sources.Not;
 import org.apache.spark.sql.sources.Or;
+import org.apache.spark.sql.sources.StringContains;
 import org.apache.spark.sql.sources.StringEndsWith;
 import org.apache.spark.sql.sources.StringStartsWith;
 
@@ -63,7 +64,8 @@ public class SparkFilterConverter {
                     "Or",
                     "Not",
                     "StringStartsWith",
-                    "StringEndsWith");
+                    "StringEndsWith",
+                    "StringContains");
 
     private final RowType rowType;
     private final PredicateBuilder builder;
@@ -148,6 +150,11 @@ public Predicate convert(Filter filter) {
             int index = fieldIndex(endsWith.attribute());
             Object literal = convertLiteral(index, endsWith.value());
             return builder.endsWith(index, literal);
+        } else if (filter instanceof StringContains) {
+            StringContains contains = (StringContains) filter;
+            int index = fieldIndex(contains.attribute());
+            Object literal = convertLiteral(index, contains.value());
+            return builder.contains(index, literal);
         }
 
         // TODO: AlwaysTrue, AlwaysFalse

paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/SparkFilterConverterTest.java

@@ -40,6 +40,7 @@
 import org.apache.spark.sql.sources.LessThan;
 import org.apache.spark.sql.sources.LessThanOrEqual;
 import org.apache.spark.sql.sources.Not;
+import org.apache.spark.sql.sources.StringContains;
 import org.apache.spark.sql.sources.StringEndsWith;
 import org.apache.spark.sql.sources.StringStartsWith;
 import org.junit.jupiter.api.Test;
@@ -167,6 +168,13 @@ public void testAll() {
         boolean test1 = endsWithPre.test(10, min, max, new GenericArray(nullCount));
         assertThat(test).isEqualTo(true);
         assertThat(test1).isEqualTo(true);
+
+        // StringContains
+        StringContains stringContains = StringContains.apply("id", "aa");
+        Predicate contains = converter01.convert(stringContains);
+        assertThat(contains.test(row)).isEqualTo(true);
+        assertThat(contains.test(max)).isEqualTo(false);
+        assertThat(contains.test(min)).isEqualTo(true);
     }
 
     @Test