[Feature][Connector-V2] Support `` for the SQL To Paimon converter (#10206)

Author: hawk9821

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

@@ -74,7 +74,6 @@
 import java.util.Optional;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
-import java.util.stream.IntStream;
 
 public class SqlToPaimonPredicateConverter {
 
@@ -123,12 +122,11 @@ public static int[] convertSqlSelectToPaimonProjectionIndex(
             }
         }
 
-        String[] columnNamesArray = columnNames.toArray(new String[0]);
         projectionIndex =
-                IntStream.range(0, columnNamesArray.length)
-                        .map(
-                                i -> {
-                                    String fieldName = columnNamesArray[i];
+                columnNames.stream()
+                        .mapToInt(
+                                columnName -> {
+                                    String fieldName = columnName.replace("`", "");
                                     int index = Arrays.asList(fieldNames).indexOf(fieldName);
                                     if (index == -1) {
                                         throw new IllegalArgumentException(
@@ -360,7 +358,7 @@ private static Object convertValueByPaimonDataType(
             RowType rowType, String columnName, Object jsqlParserDataTypeValue) {
         Optional<DataField> theFiled =
                 rowType.getFields().stream()
-                        .filter(field -> field.name().equalsIgnoreCase(columnName))
+                        .filter(field -> field.name().equalsIgnoreCase(columnName.replace("`", "")))
                         .findFirst();
         String strValue = jsqlParserDataTypeValue.toString();
         if (theFiled.isPresent()) {
@@ -424,7 +422,7 @@ private static Object getJSQLParserDataTypeValue(Expression expression) {
     }
 
     private static int getColumnIndex(PredicateBuilder builder, Column column) {
-        int index = builder.indexOf(column.getColumnName());
+        int index = builder.indexOf(column.getColumnName().replace("`", ""));
         if (index == -1) {
             throw new IllegalArgumentException(
                     String.format("The column named [%s] is not exists", column.getColumnName()));

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

@@ -318,4 +318,30 @@ public void testParseDynamicOptions() {
                 dynamicOptions.get("incremental-between-timestamp"));
         assertEquals("my-tag", dynamicOptions.get("scan.tag-name"));
     }
+
+    @Test
+    public void testPiamonQuoteIdentifier() {
+        String query =
+                "SELECT `decimal_col`, `int_col`, `char_col`, `timestamp_col`, `boolean_col`, time_col FROM table WHERE int_col > 3 OR `double_col` < 6.6 ";
+
+        PlainSelect plainSelect = convertToPlainSelect(query);
+        assertNotNull(plainSelect);
+
+        int[] fieldIndex =
+                SqlToPaimonPredicateConverter.convertSqlSelectToPaimonProjectionIndex(
+                        rowType.getFieldNames().toArray(new String[0]), plainSelect);
+        assertNotNull(fieldIndex);
+        assertEquals(6, fieldIndex.length);
+        assertEquals(4, fieldIndex[0]);
+        assertEquals(7, fieldIndex[1]);
+        assertEquals(0, fieldIndex[2]);
+        assertEquals(12, fieldIndex[3]);
+        assertEquals(2, fieldIndex[4]);
+        assertEquals(13, fieldIndex[5]);
+
+        Predicate predicate =
+                SqlToPaimonPredicateConverter.convertSqlWhereToPaimonPredicate(
+                        rowType, plainSelect);
+        assertNotNull(predicate);
+    }
 }