update

JNSimba · Nov 30, 2023 · 081b968 · 081b968
1 parent d7e36a8
commit 081b968
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 2 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/CdcTools.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/CdcTools.java
@@ -129,6 +129,7 @@ private static void syncDatabase(MultipleParameterTool params, DatabaseSync data
                 .setTableConfig(tableMap)
                 .setCreateTableOnly(createTableOnly)
                 .setNewSchemaChange(useNewSchemaChange)
+                .setSingleSink(singleSink)
                 .create();
         databaseSync.build();
         if(StringUtils.isNullOrWhitespaceOnly(jobName)){

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/tools/cdc/DatabaseSync.java
@@ -273,10 +273,10 @@ protected String getSyncTableList(List<String> syncTables){
         }else{
             // includingTablePattern and ^excludingPattern
             String includingPattern = String.format("(%s)\\.(%s)", getTableListPrefix(), includingTables);
-            if (excludingTables.isEmpty()) {
+            if (StringUtils.isNullOrWhitespaceOnly(excludingTables)) {
                 return includingPattern;
             }else{
-                String excludingPattern = String.format("?!(^%s$)", getTableListPrefix() + "\\." + excludingTables);
+                String excludingPattern = String.format("?!(%s\\.(%s))$", getTableListPrefix(), excludingTables);
                 return String.format("(%s)(%s)", includingPattern, excludingPattern);
             }
         }

diff --git a/flink-doris-connector/src/test/java/org/apache/doris/flink/tools/cdc/DatabaseSyncTest.java b/flink-doris-connector/src/test/java/org/apache/doris/flink/tools/cdc/DatabaseSyncTest.java
@@ -17,7 +17,10 @@
 package org.apache.doris.flink.tools.cdc;
 
 import org.apache.doris.flink.tools.cdc.mysql.MysqlDatabaseSync;
+import org.apache.flink.configuration.Configuration;
+import org.junit.Assert;
 import org.junit.Test;
+
 import java.util.Arrays;
 
 /**
@@ -37,4 +40,17 @@ public void multiToOneRulesParserTest() throws Exception{
             databaseSync.multiToOneRulesParser(arr[0], arr[1]);
         });
     }
+
+    @Test
+    public void getSyncTableListTest() throws Exception{
+        DatabaseSync databaseSync = new MysqlDatabaseSync();
+        databaseSync.setSingleSink(false);
+        databaseSync.setIncludingTables("tbl_1|tbl_2");
+        Configuration config = new Configuration();
+        config.setString("database-name", "db");
+        config.setString("table-name", "tbl.*");
+        databaseSync.setConfig(config);
+        String syncTableList = databaseSync.getSyncTableList(Arrays.asList("tbl_1", "tbl_2"));
+        Assert.assertEquals("db\\.tbl_1|db\\.tbl_2", syncTableList);
+    }
 }