[Fix-3876] [cdc] Remove quotation marks when building FlinkSQL (DataL…

…inkDC#3877)
aiwenmo · Oct 18, 2024 · 51ee0bb · 51ee0bb
1 parent 202a13e
commit 51ee0bb
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/sql/catalog/SQLCatalogSinkBuilder.java b/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/sql/catalog/SQLCatalogSinkBuilder.java
@@ -66,7 +66,7 @@ public void addTableSink(
         String catalogName = config.getSink().get("catalog.name");
         String sinkSchemaName = getSinkSchemaName(table);
         String tableName = getSinkTableName(table);
-        String sinkTableName = catalogName + "." + sinkSchemaName + "." + tableName;
+        String sinkTableName = catalogName + ".`" + sinkSchemaName + "`.`" + tableName + "`";
         // Because the name of the view on Flink is not allowed to have -, it needs to be replaced with - here_
         String viewName = replaceViewNameMiddleLineToUnderLine("VIEW_" + table.getSchemaTableNameWithUnderline());
 

diff --git a/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/utils/FlinkStatementUtil.java b/dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/utils/FlinkStatementUtil.java
@@ -36,7 +36,7 @@ private FlinkStatementUtil() {}
 
     public static String getCDCInsertSql(Table table, String targetName, String sourceName, FlinkCDCConfig config) {
         StringBuilder sb = new StringBuilder("INSERT INTO ");
-        sb.append("`").append(targetName).append("`");
+        sb.append(targetName);
         sb.append(" SELECT\n");
         for (int i = 0; i < table.getColumns().size(); i++) {
             sb.append("    ");