[Feature][CDC][Mysql] Support read database list (#4255)

Author: hailin0

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.cdc.debezium.row;
 
 import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.table.catalog.TablePath;
 import org.apache.seatunnel.api.table.type.MultipleRowType;
 import org.apache.seatunnel.api.table.type.RowKind;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
@@ -113,12 +114,14 @@ public void deserialize(SourceRecord record, Collector<SeaTunnelRow> collector)
         Schema valueSchema = record.valueSchema();
 
         Struct sourceStruct = messageStruct.getStruct(Envelope.FieldName.SOURCE);
+        String databaseName = sourceStruct.getString(AbstractSourceInfo.DATABASE_NAME_KEY);
         String tableName = sourceStruct.getString(AbstractSourceInfo.TABLE_NAME_KEY);
+        String tableId = TablePath.of(databaseName, tableName).toString();
         SeaTunnelRowDebeziumDeserializationConverters converters;
         if (!multipleTableRowConverters.isEmpty()) {
-            converters = multipleTableRowConverters.get(tableName);
+            converters = multipleTableRowConverters.get(tableId);
             if (converters == null) {
-                log.debug("Ignore newly added table {}", tableName);
+                log.debug("Ignore newly added table {}", tableId);
                 return;
             }
         } else {
@@ -128,26 +131,26 @@ public void deserialize(SourceRecord record, Collector<SeaTunnelRow> collector)
         if (operation == Envelope.Operation.CREATE || operation == Envelope.Operation.READ) {
             SeaTunnelRow insert = extractAfterRow(converters, record, messageStruct, valueSchema);
             insert.setRowKind(RowKind.INSERT);
-            insert.setTableId(tableName);
+            insert.setTableId(tableId);
             validator.validate(insert, RowKind.INSERT);
             collector.collect(insert);
         } else if (operation == Envelope.Operation.DELETE) {
             SeaTunnelRow delete = extractBeforeRow(converters, record, messageStruct, valueSchema);
             validator.validate(delete, RowKind.DELETE);
             delete.setRowKind(RowKind.DELETE);
-            delete.setTableId(tableName);
+            delete.setTableId(tableId);
             collector.collect(delete);
         } else {
             SeaTunnelRow before = extractBeforeRow(converters, record, messageStruct, valueSchema);
             validator.validate(before, RowKind.UPDATE_BEFORE);
             before.setRowKind(RowKind.UPDATE_BEFORE);
-            before.setTableId(tableName);
+            before.setTableId(tableId);
             collector.collect(before);
 
             SeaTunnelRow after = extractAfterRow(converters, record, messageStruct, valueSchema);
             validator.validate(after, RowKind.UPDATE_AFTER);
             after.setRowKind(RowKind.UPDATE_AFTER);
-            after.setTableId(tableName);
+            after.setTableId(tableId);
             collector.collect(after);
         }
     }

seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java

@@ -82,8 +82,9 @@ TableSource<T, SplitT, StateT> createSource(TableFactoryContext context) {
             } else {
                 Map<String, SeaTunnelRowType> rowTypeMap = new HashMap<>();
                 for (CatalogTable catalogTable : context.getCatalogTables()) {
-                    String tableId = catalogTable.getTableId().getTableName();
-                    rowTypeMap.put(tableId, catalogTable.getTableSchema().toPhysicalRowDataType());
+                    rowTypeMap.put(
+                            catalogTable.getTableId().toTablePath().toString(),
+                            catalogTable.getTableSchema().toPhysicalRowDataType());
                 }
                 dataType = new MultipleRowType(rowTypeMap);
             }

seatunnel-engine/seatunnel-engine-core/src/main/java/org/apache/seatunnel/engine/core/parse/MultipleTableJobConfigParser.java

@@ -298,7 +298,7 @@ private int getParallelism(ReadonlyConfig config) {
                             Collections.singletonList(leftAction),
                             sink,
                             factoryUrls,
-                            new SinkConfig(catalogTable.getTableId().getTableName()));
+                            new SinkConfig(catalogTable.getTableId().toTablePath().toString()));
             handleSaveMode(sink);
             sinkAction.setParallelism(leftAction.getParallelism());
             sinkActions.add(sinkAction);