支持按照SR表的字段顺序写入数据

Author: zhuxt2015

dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/postgres/PostgresCDCBuilder.java

@@ -118,4 +118,29 @@ public String generateUrl(String schema) {
     protected String getMetadataType() {
         return METADATA_TYPE;
     }
+
+    @Override
+    public Map<String, String> parseMetaDataConfig() {
+        String url = String.format(
+                "jdbc:postgres://%s:%d/%s",
+                config.getHostname(), config.getPort(), composeJdbcProperties(config.getJdbc()));
+        return parseMetaDataSingleConfig(url);
+    }
+
+    private String composeJdbcProperties(Map<String, String> jdbcProperties) {
+        if (jdbcProperties == null || jdbcProperties.isEmpty()) {
+            return "";
+        }
+
+        StringBuilder sb = new StringBuilder();
+        sb.append('?');
+        jdbcProperties.forEach((k, v) -> {
+            sb.append(k);
+            sb.append("=");
+            sb.append(v);
+            sb.append("&");
+        });
+        sb.deleteCharAt(sb.length() - 1);
+        return sb.toString();
+    }
 }

dinky-cdc/dinky-cdc-core/src/main/java/org/dinky/cdc/sql/SQLSinkBuilder.java

@@ -100,12 +100,16 @@ private List<Operation> addSinkInsert(
             FlinkTableObjectIdentifier targetTable,
             String sinkSchemaName,
             FlinkTableObjectIdentifier sinkTable) {
-        String pkList = StringUtils.join(getPKList(table), ".");
-        String flinkDDL = FlinkStatementUtil.getFlinkDDL(table, targetTable, config, sinkSchemaName, sinkTable, pkList);
+        Table sinkTableObject = table.getSinkTable();
+        if (sinkTableObject == null) {
+            sinkTableObject = table;
+        }
+        String pkList = StringUtils.join(getPKList(sinkTableObject), ".");
+        String flinkDDL = FlinkStatementUtil.getFlinkDDL(sinkTableObject, targetTable, config, sinkSchemaName, sinkTable, pkList);
         logger.info(flinkDDL);
         customTableEnvironment.executeSql(flinkDDL);
         logger.info("Create {} FlinkSQL DDL successful...", targetTable);
-        return createInsertOperations(table, sourceTable, targetTable);
+        return createInsertOperations(sinkTableObject, sourceTable, targetTable);
     }
 
     @Override

dinky-common/src/main/java/org/dinky/data/model/Table.java

@@ -67,6 +67,9 @@ public class Table implements Serializable, Comparable<Table>, Cloneable {
 
     private List<Column> columns;
 
+    /** The sink table for the source table */
+    private Table sinkTable;
+
     /** 驱动类型, @see org.dinky.metadata.enums.DriverType */
     private String driverType;
 

dinky-core/src/main/java/org/dinky/trans/ddl/CreateCDCSourceOperation.java

@@ -129,7 +129,9 @@ public TableResult execute(Executor executor) {
                     String tableName = schemaTableName.split("\\.")[1];
                     table.setColumns(driver.listColumnsSortByPK(realSchemaName, tableName));
                     schemaList.add(schema);
-
+                    Driver sinkRealDriver = getDriver(config, schemaName);
+                    final List<Table> sinkTables= getSinkTables(config, schemaName);
+                    setSinkTable(table, sinkTables, sinkBuilder, sinkRealDriver);
                     if (null != sinkDriver) {
                         final String createTableOptions = config.getSink().get(FlinkCDCConfig.AUTO_CREATE_OPTIONS);
                         Table sinkTable = (Table) table.clone();
@@ -151,6 +153,8 @@ public TableResult execute(Executor executor) {
                     Driver driver = Driver.build(confMap.get("name"), confMap.get("type"), JsonUtils.toMap(confMap));
 
                     final List<Table> tables = driver.listTables(schemaName);
+                    Driver sinkRealDriver = getDriver(config, schemaName);
+                    final List<Table> sinkTables= getSinkTables(config, schemaName);
                     for (Table table : tables) {
                         if (!Asserts.isEquals(table.getType(), "VIEW")) {
                             if (Asserts.isNotNullCollection(tableRegList)) {
@@ -160,13 +164,15 @@ public TableResult execute(Executor executor) {
                                         table.setColumns(driver.listColumnsSortByPK(schemaName, table.getName()));
                                         schema.getTables().add(table);
                                         schemaTableNameList.add(table.getSchemaTableName());
+                                        setSinkTable(table, sinkTables, sinkBuilder, sinkRealDriver);
                                         break;
                                     }
                                 }
                             } else {
                                 table.setColumns(driver.listColumnsSortByPK(schemaName, table.getName()));
                                 schemaTableNameList.add(table.getSchemaTableName());
                                 schema.getTables().add(table);
+                                setSinkTable(table, sinkTables, sinkBuilder, sinkRealDriver);
                             }
                         }
                     }
@@ -223,19 +229,70 @@ public TableResult execute(Executor executor) {
         return tableResultBuilder.build();
     }
 
+    private static void setSinkTable(Table table, List<Table> sinkTables, SinkBuilder sinkBuilder, Driver sinkRealDriver) {
+        String sinkTableName = sinkBuilder.getSinkTableName(table);
+        for (Table sinkTable : sinkTables) {
+            String sinkTableSchema = sinkTable.getSchema();
+            String sinkTableSchemaTableName = sinkTable.getSchemaTableName();
+            String currentSinkTableName = sinkTableSchemaTableName;
+            if (Asserts.isContainsString(sinkTableSchemaTableName, ".")) {
+                currentSinkTableName = sinkTableSchemaTableName.split("\\.")[1];
+            }
+            if (sinkTableName.equals(currentSinkTableName)) {
+                if (null != sinkRealDriver) {
+                    sinkTable.setColumns(sinkRealDriver.listColumnsSortByPK(sinkTableSchema, currentSinkTableName));
+                }
+                table.setSinkTable(sinkTable);
+                break;
+            }
+        }
+    }
+
+    private List<Table> getSinkTables(FlinkCDCConfig config, String schemaName) throws Exception {
+        List<Table> sinkTables;
+        Driver sinkDriver = getDriver(config, schemaName);
+        if (null == sinkDriver) {
+            return new ArrayList<>();
+        }
+        Map<String, String> sink = config.getSink();
+        String schema = schemaName;
+        String sinkDb = sink.get(FlinkCDCConfig.SINK_DB);
+        if (Asserts.isNotNullString(sinkDb)) {
+            schema = SqlUtil.replaceAllParam(sinkDb, "schemaName", schemaName);
+        }
+        sinkTables = sinkDriver.listTables(schema);
+        return sinkTables;
+    }
+
     private Driver checkAndCreateSinkSchema(FlinkCDCConfig config, String schemaName) throws Exception {
         Map<String, String> sink = config.getSink();
         String autoCreate = sink.get(FlinkCDCConfig.AUTO_CREATE);
         if (!Asserts.isEqualsIgnoreCase(autoCreate, "true") || Asserts.isNullString(schemaName)) {
             return null;
         }
-        String url = sink.get("url");
+        return getDriver(config, schemaName);
+    }
+
+    private Driver getDriver(FlinkCDCConfig config, String schemaName) throws Exception {
+        Map<String, String> sink = config.getSink();
+        String connector = sink.get("connector");
+        String url;
+        if (Asserts.isEquals(connector, "starrocks")) {
+            url = sink.get("jdbc-url");
+        } else if (Asserts.isEqualsIgnoreCase(connector, "doris")) {
+            url = "jdbc:mysql://" + sink.get("fenodes");
+        } else if (Asserts.isEqualsIgnoreCase(connector, "jdbc")) {
+            url = sink.get("url");
+        } else {
+            return null;
+        }
         String schema = schemaName;
         String sinkDb = sink.get(FlinkCDCConfig.SINK_DB);
         if (Asserts.isNotNullString(sinkDb)) {
             schema = SqlUtil.replaceAllParam(sinkDb, "schemaName", schemaName);
         }
-        Driver driver = Driver.build(sink.get("connector"), url, sink.get("username"), sink.get("password"));
+
+        Driver driver = Driver.build(connector, url, sink.get("username"), sink.get("password"));
         if (null != driver && !driver.existSchema(schema)) {
             driver.createSchema(schema);
         }