full database matching rules

Signed-off-by: 张田 <zhang_tian@inspur.com>
apache · Dec 21, 2023 · 7c9ff13 · 7c9ff13
1 parent 6b6d9fc
commit 7c9ff13
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 0 deletions.
diff --git a/flink-cdc-common/src/main/java/com/ververica/cdc/common/event/TableId.java b/flink-cdc-common/src/main/java/com/ververica/cdc/common/event/TableId.java
@@ -84,6 +84,17 @@ public static TableId parse(String tableId) {
         throw new IllegalArgumentException("Invalid tableId: " + tableId);
     }
 
+    public static TableId parse(String namespace, String schemaName, String tableName) {
+        if (namespace != null) {
+            return tableId(namespace, schemaName, tableName);
+        } else if (schemaName != null) {
+            return tableId(schemaName, tableName);
+        } else if (tableName != null) {
+            return tableId(tableName);
+        }
+        throw new IllegalArgumentException("Invalid tableName: " + tableName);
+    }
+
     public String identifier() {
         if (namespace == null || namespace.isEmpty()) {
             if (schemaName == null || schemaName.isEmpty()) {

diff --git a/flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/route/RouteFunction.java b/flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/route/RouteFunction.java
@@ -96,6 +96,15 @@ public Event map(Event event) throws Exception {
             Selectors selectors = route.f0;
             TableId replaceBy = route.f1;
             if (selectors.isMatch(tableId)) {
+                // Add a rule that when configuring * in tablename,
+                // only the namespace name and schemaName name needs to be changed
+                if (replaceBy.getTableName().equalsIgnoreCase("<>")) {
+                    replaceBy =
+                            TableId.parse(
+                                    replaceBy.getNamespace(),
+                                    replaceBy.getSchemaName(),
+                                    tableId.getTableName());
+                }
                 return recreateChangeEvent(changeEvent, replaceBy);
             }
         }