[Fix][Connector-v2][Clickhouse] Handle special characters in column comments

.github/workflows/backend.yml

@@ -449,7 +449,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 120
+    timeout-minutes: 180
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}
@@ -510,7 +510,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 120
+    timeout-minutes: 180
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}
@@ -540,7 +540,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 120
+    timeout-minutes: 180
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}

seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/util/CatalogUtil.java

@@ -32,6 +32,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.function.Function;
+import java.util.regex.Matcher;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -70,7 +71,7 @@ public String getCreateTableSql(
         template =
                 template.replaceAll(
                         SaveModePlaceHolder.ROWTYPE_PRIMARY_KEY.getReplacePlaceHolder(),
-                        primaryKey);
+                        Matcher.quoteReplacement(primaryKey));
         SqlTemplate.canHandledByTemplateWithPlaceholder(
                 template,
                 SaveModePlaceHolder.ROWTYPE_UNIQUE_KEY.getPlaceHolder(),
@@ -80,7 +81,8 @@ public String getCreateTableSql(
 
         template =
                 template.replaceAll(
-                        SaveModePlaceHolder.ROWTYPE_UNIQUE_KEY.getReplacePlaceHolder(), uniqueKey);
+                        SaveModePlaceHolder.ROWTYPE_UNIQUE_KEY.getReplacePlaceHolder(),
+                        Matcher.quoteReplacement(uniqueKey));
         Map<String, CreateTableParser.ColumnInfo> columnInTemplate =
                 CreateTableParser.getColumnList(template);
         template = mergeColumnInTemplate(columnInTemplate, tableSchema, template);
@@ -95,20 +97,27 @@ public String getCreateTableSql(
             // TODO: Remove this compatibility config
             template =
                     template.replaceAll(
-                            SaveModePlaceHolder.TABLE_NAME.getReplacePlaceHolder(), table);
+                            SaveModePlaceHolder.TABLE_NAME.getReplacePlaceHolder(),
+                            Matcher.quoteReplacement(table));
             log.warn(
                     "The variable placeholder `${table_name}` has been marked as deprecated and will be removed soon, please use `${table}`");
         }
 
-        return template.replaceAll(SaveModePlaceHolder.DATABASE.getReplacePlaceHolder(), database)
-                .replaceAll(SaveModePlaceHolder.TABLE.getReplacePlaceHolder(), table)
+        return template.replaceAll(
+                        SaveModePlaceHolder.DATABASE.getReplacePlaceHolder(),
+                        Matcher.quoteReplacement(database))
                 .replaceAll(
-                        SaveModePlaceHolder.ROWTYPE_FIELDS.getReplacePlaceHolder(), rowTypeFields)
+                        SaveModePlaceHolder.TABLE.getReplacePlaceHolder(),
+                        Matcher.quoteReplacement(table))
+                .replaceAll(
+                        SaveModePlaceHolder.ROWTYPE_FIELDS.getReplacePlaceHolder(),
+                        Matcher.quoteReplacement(rowTypeFields))
                 .replaceAll(
                         SaveModePlaceHolder.COMMENT.getReplacePlaceHolder(),
-                        Objects.isNull(comment)
-                                ? ""
-                                : comment.replace("'", "''").replace("\\", "\\\\"));
+                        Matcher.quoteReplacement(
+                                Objects.isNull(comment)
+                                        ? ""
+                                        : comment.replace("'", "''").replace("\\", "\\\\")));
     }
 
     private String mergeColumnInTemplate(

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseIT.java

@@ -131,6 +131,46 @@ public void testClickhouseWithCreateSchemaWhenComment(TestContainer container)
         Assertions.assertEquals(0, execResult.getExitCode());
     }
 
+    @TestTemplate
+    public void testClickhouseAutoCreateTableWithSpecialCharactersInComments(
+            TestContainer testContainer) throws Exception {
+        String testTableName = "test_special_chars_comments_table";
+
+        String createSourceTableSql =
+                String.format(
+                        "CREATE TABLE IF NOT EXISTS %s.%s ("
+                                + "id UInt64, "
+                                + "col_with_dollar_comment String COMMENT 'Comment with $1 and $2 special chars', "
+                                + "col_with_backslash_comment String COMMENT 'Comment with \\\\ backslash', "
+                                + "col_with_mixed_chars String COMMENT '~`!@#$%%^&*()_+-*/-=[]{}', "
+                                + "col_with_chinese_chars String COMMENT '这是特殊符号测试英文键盘：~`!@#$%%^&*()_+-*/-=[]{}'"
+                                + ") ENGINE = MergeTree() ORDER BY id",
+                        DATABASE, testTableName);
+
+        String sinkTableName = testTableName + "_sink";
+
+        try (Statement statement = connection.createStatement()) {
+            statement.execute(createSourceTableSql);
+
+            String insertSql =
+                    String.format(
+                            "INSERT INTO %s.%s VALUES "
+                                    + "(1, 'value1', 'value2', 'value3', 'value4')",
+                            DATABASE, testTableName);
+            statement.execute(insertSql);
+        }
+
+        Container.ExecResult execResult =
+                testContainer.executeJob("/clickhouse_auto_create_with_special_comments.conf");
+
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
+
+        Assertions.assertEquals(1, countData(sinkTableName));
+
+        dropTable(DATABASE + "." + testTableName);
+        dropTable(DATABASE + "." + sinkTableName);
+    }
+
     @TestTemplate
     public void clickhouseWithCreateSchemaWhenNotExist(TestContainer container) throws Exception {
         String tableName = "default.sink_table_for_schema";

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/resources/clickhouse_auto_create_with_special_comments.conf

@@ -0,0 +1,58 @@
+#
+# 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.
+#
+######
+###### This config file tests auto create table with special characters in column comments
+###### Testing regex special characters like $ and \ are properly handled by Matcher.quoteReplacement
+######
+
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+  checkpoint.interval = 10
+}
+
+source {
+  Clickhouse {
+    host = "clickhouse:8123"
+    table_path = "default.test_special_chars_comments_table"
+    sql = "select * from default.test_special_chars_comments_table"
+    username = "default"
+    password = ""
+    plugin_output = "source_table"
+  }
+}
+
+sink {
+  Clickhouse {
+    host = "clickhouse:8123"
+    database = "default"
+    table = "test_special_chars_comments_table_sink"
+    username = "default"
+    password = ""
+    "schema_save_mode" = "CREATE_SCHEMA_WHEN_NOT_EXIST"
+    "data_save_mode" = "APPEND_DATA"
+    "save_mode_create_template" = """
+     CREATE TABLE IF NOT EXISTS `${database}`.`${table}` (
+    ${rowtype_fields}
+    ) ENGINE = MergeTree()
+    ORDER BY (id)
+    COMMENT '${comment}';
+    """
+    support_upsert = true
+    allow_experimental_lightweight_delete = true
+  }
+}