[Improve][Connector-V2] Support like predicate pushdown in paimon (#9653)

Author: xiaochen-zhou

docs/en/connector-v2/source/Paimon.md

@@ -85,7 +85,7 @@ The file path of `hdfs-site.xml`
 ### query [string]
 
 The filter condition of the table read. For example: `select * from st_test where id > 100`. If not specified, all rows are read.
-Currently, where conditions only support <, <=, >, >=, =, !=, or, and,is null, is not null, between...and, in, not in, like(pattern matching with prefix only) ,and others are not supported.
+Currently, where conditions only support <, <=, >, >=, =, !=, or, and,is null, is not null, between...and, in, not in, like, and others are not supported.
 The Having, Group By, Order By clauses are currently unsupported, because these clauses are not supported by Paimon.
 you can also project specific columns, for example: select id, name from st_test where id > 100.
 The limit will be supported in the future.

docs/zh/connector-v2/source/Paimon.md

@@ -86,7 +86,7 @@ Paimon 的 catalog uri，仅当 catalog_type 为 hive 时需要
 
 读取表格的筛选条件，例如：`select * from st_test where id > 100`。如果未指定，则将读取所有记录。 
 
-目前，`where` 支持`<, <=, >, >=, =, !=, or, and,is null, is not null, between...and, in , not in, like(pattern matching with prefix only)`，其他暂不支持。 
+目前，`where` 支持`<, <=, >, >=, =, !=, or, and,is null, is not null, between...and, in , not in, like`，其他暂不支持。 
 
 Projection 已支持,你可以选择特定的列，例如：select id, name from st_test where id > 100。
 

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/converter/SqlToPaimonPredicateConverter.java

@@ -252,16 +252,31 @@ private static Predicate parseExpressionToPredicate(
             Object rightVal =
                     convertValueByPaimonDataType(rowType, column.getColumnName(), rightPredicate);
 
-            Pattern BEGIN_PATTERN = Pattern.compile("([^%]+)%");
-            Matcher matcher = BEGIN_PATTERN.matcher(rightVal.toString());
-            if (matcher.matches()) {
-                return builder.startsWith(columnIndex, BinaryString.fromString(matcher.group(1)));
-            } else {
-                throw new IllegalArgumentException(
-                        "Unsupported expression type: "
-                                + expression.getClass().getSimpleName()
-                                + ", only support like pattern matching with prefix");
+            Pattern BEGIN_PATTERN = Pattern.compile("([^%]+)%$");
+            Matcher beginMatcher = BEGIN_PATTERN.matcher(rightVal.toString());
+            if (beginMatcher.matches()) {
+                return builder.startsWith(
+                        columnIndex, BinaryString.fromString(beginMatcher.group(1)));
+            }
+
+            Pattern END_PATTERN = Pattern.compile("^%([^%]+)");
+            Matcher endMatcher = END_PATTERN.matcher(rightVal.toString());
+            if (endMatcher.matches()) {
+                return builder.endsWith(columnIndex, BinaryString.fromString(endMatcher.group(1)));
             }
+
+            Pattern CONTAINS_PATTERN = Pattern.compile("^%([^%]+)%$");
+            Matcher containsMatcher = CONTAINS_PATTERN.matcher(rightVal.toString());
+            if (containsMatcher.matches()) {
+                return builder.contains(
+                        columnIndex, BinaryString.fromString(containsMatcher.group(1)));
+            }
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Invalid LIKE pattern: '%s'. Supported patterns are: 'prefix%%', '%%suffix', and '%%substring%%'. "
+                                    + "Please ensure your pattern matches one of these formats.",
+                            rightVal.toString()));
+
         } else if (expression instanceof Parenthesis) {
             Parenthesis parenthesis = (Parenthesis) expression;
             return parseExpressionToPredicate(builder, rowType, parenthesis.getExpression());

seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/source/converter/SqlToPaimonConverterTest.java

@@ -250,7 +250,7 @@ public void testConvertSqlSelectToPaimonProjectionArrayWithStar() {
     }
 
     @Test
-    public void testConvertSqlWhereToPaimonPredicateWithStartWith() {
+    public void testConvertSqlWhereToPaimonLikePredicate() {
         String query = "SELECT * FROM table WHERE varchar_col like 'te%'";
 
         PlainSelect plainSelect = convertToPlainSelect(query);
@@ -264,5 +264,32 @@ public void testConvertSqlWhereToPaimonPredicateWithStartWith() {
         Predicate expectedPredicate = PredicateBuilder.or(builder.startsWith(1, "te"));
 
         assertEquals(expectedPredicate.toString(), predicate.toString());
+
+        query = "SELECT * FROM table WHERE varchar_col like '%st'";
+
+        plainSelect = convertToPlainSelect(query);
+        predicate =
+                SqlToPaimonPredicateConverter.convertSqlWhereToPaimonPredicate(
+                        rowType, plainSelect);
+
+        assertNotNull(predicate);
+
+        builder = new PredicateBuilder(rowType);
+        expectedPredicate = PredicateBuilder.or(builder.endsWith(1, "st"));
+
+        assertEquals(expectedPredicate.toString(), predicate.toString());
+
+        query = "SELECT * FROM table WHERE varchar_col like '%es%'";
+        plainSelect = convertToPlainSelect(query);
+        predicate =
+                SqlToPaimonPredicateConverter.convertSqlWhereToPaimonPredicate(
+                        rowType, plainSelect);
+
+        assertNotNull(predicate);
+
+        builder = new PredicateBuilder(rowType);
+        expectedPredicate = PredicateBuilder.or(builder.contains(1, "es"));
+
+        assertEquals(expectedPredicate.toString(), predicate.toString());
     }
 }

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paimon/PaimonSinkCDCIT.java

@@ -477,6 +477,12 @@ public void testFakeSinkPaimonWithFullTypeAndReadWithFilter(TestContainer contai
         Container.ExecResult readResult8 =
                 container.executeJob("/paimon_to_assert_with_filter8.conf");
         Assertions.assertEquals(0, readResult8.getExitCode());
+        Container.ExecResult readResult9 =
+                container.executeJob("/paimon_to_assert_with_filter9.conf");
+        Assertions.assertEquals(0, readResult9.getExitCode());
+        Container.ExecResult readResult10 =
+                container.executeJob("/paimon_to_assert_with_filter10.conf");
+        Assertions.assertEquals(0, readResult10.getExitCode());
     }
 
     @TestTemplate

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/resources/paimon_to_assert_with_filter10.conf

@@ -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.
+#
+
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+}
+
+source {
+  Paimon {
+    warehouse = "/tmp/seatunnel_mnt/paimon"
+    database = "full_type"
+    table = "st_test"
+    query = "select * from st_test where c_string like '%string%'"
+    plugin_output = paimon_source
+  }
+}
+
+sink {
+  Assert {
+    plugin_input = paimon_source
+    rules {
+      row_rules = [
+        {
+          rule_type = MAX_ROW
+          rule_value = 3
+        }
+        {
+          rule_type = MIN_ROW
+          rule_value = 3
+        }
+      ]
+      field_rules = [
+        {
+          field_name = c_string
+          field_type = string
+          field_value = [
+            {
+              rule_type = NOT_NULL
+            }
+          ]
+        }
+      ]
+    }
+  }
+}

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/resources/paimon_to_assert_with_filter9.conf

@@ -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.
+#
+
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+}
+
+source {
+  Paimon {
+    warehouse = "/tmp/seatunnel_mnt/paimon"
+    database = "full_type"
+    table = "st_test"
+    query = "select * from st_test where c_string like '%string2'"
+    plugin_output = paimon_source
+  }
+}
+
+sink {
+  Assert {
+    plugin_input = paimon_source
+    rules {
+      row_rules = [
+        {
+          rule_type = MAX_ROW
+          rule_value = 1
+        }
+        {
+          rule_type = MIN_ROW
+          rule_value = 1
+        }
+      ]
+      field_rules = [
+        {
+          field_name = c_string
+          field_type = string
+          field_value = [
+            {
+              rule_type = NOT_NULL
+            }
+          ]
+        }
+      ]
+    }
+  }
+}