[Bug] [CDCSOURCE] fix cdcsource split (DataLinkDC#3030)

hugh018 · Jan 19, 2024 · ed7581d · ed7581d
1 parent ade4120
commit ed7581d
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 3 deletions.
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
@@ -35,7 +35,7 @@ private FlinkStatementUtil() {}
 
     public static String getCDCInsertSql(Table table, String targetName, String sourceName) {
         StringBuilder sb = new StringBuilder("INSERT INTO ");
-        sb.append(targetName);
+        sb.append("`").append(targetName).append("`");
         sb.append(" SELECT\n");
         for (int i = 0; i < table.getColumns().size(); i++) {
             sb.append("    ");

diff --git a/dinky-core/src/main/java/org/dinky/trans/ddl/CreateCDCSourceOperation.java b/dinky-core/src/main/java/org/dinky/trans/ddl/CreateCDCSourceOperation.java
@@ -123,9 +123,9 @@ public TableResult execute(Executor executor) {
                     // 分库分表所有表结构都是一样的，取出列表中第一个表名即可
                     String schemaTableName = table.getSchemaTableNameList().get(0);
                     // 真实的表名
+                    String realSchemaName = schemaTableName.split("\\.")[0];
                     String tableName = schemaTableName.split("\\.")[1];
-                    table.setColumns(driver.listColumnsSortByPK(schemaName, tableName));
-                    table.setColumns(driver.listColumnsSortByPK(schemaName, table.getName()));
+                    table.setColumns(driver.listColumnsSortByPK(realSchemaName, tableName));
                     schemaList.add(schema);
 
                     if (null != sinkDriver) {