[Feature-506][client] The CDCSOURCE table parameter supports line fee…

[Feature-506][client] The CDCSOURCE table parameter supports line fee…

dlink-client/dlink-client-1.11/src/main/java/com/dlink/cdc/AbstractCDCBuilder.java

@@ -42,6 +42,7 @@ public abstract class AbstractCDCBuilder {
         }
         List<String> tableList = getTableList();
         for (String tableName : tableList) {
+            tableName = tableName.trim();
             if (Asserts.isNotNullString(tableName) && tableName.contains(".")) {
                 String[] names = tableName.split("\\\\.");
                 if (!schemaList.contains(names[0])) {

dlink-client/dlink-client-1.11/src/main/java/com/dlink/cdc/mysql/MysqlCDCBuilder.java

@@ -87,6 +87,7 @@ public class MysqlCDCBuilder extends AbstractCDCBuilder implements CDCBuilder {
         }
         List<String> tableList = getTableList();
         for (String tableName : tableList) {
+            tableName = tableName.trim();
             if (Asserts.isNotNullString(tableName) && tableName.contains(".")) {
                 String[] names = tableName.split("\\\\.");
                 if (!schemaList.contains(names[0])) {

dlink-client/dlink-client-1.12/src/main/java/com/dlink/cdc/AbstractCDCBuilder.java

@@ -42,6 +42,7 @@ public abstract class AbstractCDCBuilder {
         }
         List<String> tableList = getTableList();
         for (String tableName : tableList) {
+            tableName = tableName.trim();
             if (Asserts.isNotNullString(tableName) && tableName.contains(".")) {
                 String[] names = tableName.split("\\\\.");
                 if (!schemaList.contains(names[0])) {

dlink-client/dlink-client-1.12/src/main/java/com/dlink/cdc/mysql/MysqlCDCBuilder.java

@@ -100,6 +100,7 @@ public class MysqlCDCBuilder extends AbstractCDCBuilder implements CDCBuilder {
         }
         List<String> tableList = getTableList();
         for (String tableName : tableList) {
+            tableName = tableName.trim();
             if (Asserts.isNotNullString(tableName) && tableName.contains(".")) {
                 String[] names = tableName.split("\\\\.");
                 if (!schemaList.contains(names[0])) {

dlink-client/dlink-client-1.13/src/main/java/com/dlink/cdc/AbstractCDCBuilder.java

@@ -42,6 +42,7 @@ public abstract class AbstractCDCBuilder {
         }
         List<String> tableList = getTableList();
         for (String tableName : tableList) {
+            tableName = tableName.trim();
             if (Asserts.isNotNullString(tableName) && tableName.contains(".")) {
                 String[] names = tableName.split("\\\\.");
                 if (!schemaList.contains(names[0])) {

dlink-client/dlink-client-1.13/src/main/java/com/dlink/cdc/mysql/MysqlCDCBuilder.java

@@ -145,6 +145,7 @@ public class MysqlCDCBuilder extends AbstractCDCBuilder implements CDCBuilder {
         }
         List<String> tableList = getTableList();
         for (String tableName : tableList) {
+            tableName = tableName.trim();
             if (Asserts.isNotNullString(tableName) && tableName.contains(".")) {
                 String[] names = tableName.split("\\\\.");
                 if (!schemaList.contains(names[0])) {

dlink-client/dlink-client-1.14/src/main/java/com/dlink/cdc/AbstractCDCBuilder.java

@@ -42,6 +42,7 @@ public abstract class AbstractCDCBuilder {
         }
         List<String> tableList = getTableList();
         for (String tableName : tableList) {
+            tableName = tableName.trim();
             if (Asserts.isNotNullString(tableName) && tableName.contains(".")) {
                 String[] names = tableName.split("\\\\.");
                 if (!schemaList.contains(names[0])) {

dlink-client/dlink-client-1.14/src/main/java/com/dlink/cdc/mysql/MysqlCDCBuilder.java

@@ -145,6 +145,7 @@ public class MysqlCDCBuilder extends AbstractCDCBuilder implements CDCBuilder {
         }
         List<String> tableList = getTableList();
         for (String tableName : tableList) {
+            tableName = tableName.trim();
             if (Asserts.isNotNullString(tableName) && tableName.contains(".")) {
                 String[] names = tableName.split("\\\\.");
                 if (!schemaList.contains(names[0])) {

dlink-client/dlink-client-1.15/src/main/java/com/dlink/cdc/AbstractCDCBuilder.java

@@ -42,6 +42,7 @@ public abstract class AbstractCDCBuilder {
         }
         List<String> tableList = getTableList();
         for (String tableName : tableList) {
+            tableName = tableName.trim();
             if (Asserts.isNotNullString(tableName) && tableName.contains(".")) {
                 String[] names = tableName.split("\\\\.");
                 if (!schemaList.contains(names[0])) {

dlink-client/dlink-client-1.15/src/main/java/com/dlink/cdc/mysql/MysqlCDCBuilder.java

@@ -145,6 +145,7 @@ public class MysqlCDCBuilder extends AbstractCDCBuilder implements CDCBuilder {
         }
         List<String> tableList = getTableList();
         for (String tableName : tableList) {
+            tableName = tableName.trim();
             if (Asserts.isNotNullString(tableName) && tableName.contains(".")) {
                 String[] names = tableName.split("\\\\.");
                 if (!schemaList.contains(names[0])) {

dlink-core/src/test/java/com/dlink/core/SqlParserTest.java

@@ -66,17 +66,30 @@ public class SqlParserTest {
     @Test
     public void createCDCSourceTest() {
         String sql = "EXECUTE CDCSOURCE demo WITH (\n" +
-            " 'type'='mysql-cdc',\n" +
-            " 'hostname'='127.0.0.1',\n" +
-            " 'port'='3306',\n" +
-            " 'password'='dlink',\n" +
-            " 'hostname'='dlink',\n" +
-            " 'checkpoint'='3000',\n" +
-            " 'parallelism'='1',\n" +
-            " 'database'='dlink,test',\n" +
-            " 'table'='',\n" +
-            " 'topic'='dlinkcdc',\n" +
-            " 'brokers'='127.0.0.1:9092'\n" +
+            "  'connector' = 'mysql-cdc',\n" +
+            "  'hostname' = '10.1.51.25',\n" +
+            "  'port' = '3306',\n" +
+            "  'username' = 'dfly',\n" +
+            "  'password' = 'Dareway@2020',\n" +
+            "  'checkpoint' = '3000',\n" +
+            "  'scan.startup.mode' = 'initial',\n" +
+            "  'parallelism' = '1',\n" +
+            "  -- 'database-name'='test',\n" +
+            "  'table-name' = 'test\\.student,\n" +
+            " test\\.score',\n" +
+            "  -- 'sink.connector'='datastream-doris',\n" +
+            "  'sink.connector' = 'doris',\n" +
+            "  'sink.fenodes' = '10.1.51.26:8030',\n" +
+            "  'sink.username' = 'root',\n" +
+            "  'sink.password' = 'dw123456',\n" +
+            "  'sink.sink.batch.size' = '1',\n" +
+            "  'sink.sink.max-retries' = '1',\n" +
+            "  'sink.sink.batch.interval' = '60000',\n" +
+            "  'sink.sink.db' = 'test',\n" +
+            "  'sink.table.prefix' = 'ODS_',\n" +
+            "  'sink.table.upper' = 'true',\n" +
+            "  'sink.table.identifier' = '${schemaName}.${tableName}',\n" +
+            "  'sink.sink.enable-delete' = 'true'\n" +
             ");";
         Map<String, List<String>> lists = SingleSqlParserFactory.generateParser(sql);
         System.out.println(lists.toString());

dlink-executor/src/main/java/com/dlink/trans/ddl/CreateCDCSourceOperation.java

@@ -74,15 +74,15 @@ public class CreateCDCSourceOperation extends AbstractOperation implements Opera
                     if (!Asserts.isEquals(table.getType(), "VIEW")) {
                         if (Asserts.isNotNullCollection(tableRegList)) {
                             for (String tableReg : tableRegList) {
-                                if (table.getSchemaTableName().matches(tableReg) && !schema.getTables().contains(Table.build(table.getName()))) {
-                                    table.setColumns(driver.listColumns(schemaName, table.getName()));
+                                if (table.getSchemaTableName().matches(tableReg.trim()) && !schema.getTables().contains(Table.build(table.getName()))) {
+                                    table.setColumns(driver.listColumnsSortByPK(schemaName, table.getName()));
                                     schema.getTables().add(table);
                                     schemaTableNameList.add(table.getSchemaTableName());
                                     break;
                                 }
                             }
                         } else {
-                            table.setColumns(driver.listColumns(schemaName, table.getName()));
+                            table.setColumns(driver.listColumnsSortByPK(schemaName, table.getName()));
                             schemaTableNameList.add(table.getSchemaTableName());
                             schema.getTables().add(table);
                         }

dlink-metadata/dlink-metadata-base/src/main/java/com/dlink/metadata/driver/AbstractJdbcDriver.java

@@ -235,6 +235,23 @@ public abstract class AbstractJdbcDriver extends AbstractDriver {
         return columns;
     }
 
+    @Override
+    public List<Column> listColumnsSortByPK(String schemaName, String tableName) {
+        List<Column> columnList = listColumns(schemaName, tableName);
+        List<Column> columnListSortByPK = new ArrayList<>();
+        for(Column column: columnList){
+            if(column.isKeyFlag()){
+                columnListSortByPK.add(column);
+            }
+        }
+        for(Column column: columnList){
+            if(!column.isKeyFlag()){
+                columnListSortByPK.add(column);
+            }
+        }
+        return columnListSortByPK;
+    }
+
     @Override
     public boolean createTable(Table table) throws Exception {
         String sql = getCreateTableSql(table).replaceAll("\r\n", " ");

dlink-metadata/dlink-metadata-base/src/main/java/com/dlink/metadata/driver/Driver.java

@@ -71,6 +71,8 @@ public interface Driver {
 
     List<Column> listColumns(String schemaName, String tableName);
 
+    List<Column> listColumnsSortByPK(String schemaName, String tableName);
+
     List<Schema> getSchemasAndTables();
 
     List<Table> getTablesAndColumns(String schemaName);