fix

Author: czy006

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/io/debezium/relational/ddl/DdlChanges.java

@@ -10,11 +10,14 @@
 import java.util.List;
 import java.util.Set;
 import java.util.function.Predicate;
+import java.util.regex.Pattern;
 
 import io.debezium.annotation.NotThreadSafe;
 import io.debezium.relational.RelationalTableFilters;
 import io.debezium.relational.TableId;
 import io.debezium.relational.Tables.TableFilter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Copied from Debezium project.
@@ -25,9 +28,11 @@
 @NotThreadSafe
 public class DdlChanges implements DdlParserListener {
 
+  private final static Logger LOGGER = LoggerFactory.getLogger(DdlChanges.class);
   private final String terminator;
   private final List<Event> events = new ArrayList<>();
   private final Set<String> databaseNames = new HashSet<>();
+  private static final Pattern GHOST_TABLE_PATTERN = Pattern.compile("^_(.*)_(del|gho|ghc)$");
 
   /**
    * Create a new changes object with ';' as the terminator token.
@@ -253,8 +258,31 @@ public boolean anyMatch(RelationalTableFilters filters) {
             || databaseFilter.test(((SetVariableEvent) event).databaseName().get())));
   }
 
-  public boolean anyMatch(Predicate<Event> predicate) {
-    return events.stream().anyMatch(predicate);
+  public boolean anyMatch(RelationalTableFilters filters,boolean parserOnlineDDL) {
+    Predicate<String> databaseFilter = filters.databaseFilter();
+    TableFilter tableFilter = filters.dataCollectionFilter();
+    return events.stream().anyMatch(event -> {
+      if (event instanceof DatabaseEvent) {
+        DatabaseEvent dbEvent = (DatabaseEvent) event;
+        return databaseFilter.test(dbEvent.databaseName());
+      }
+      if (event instanceof TableEvent) {
+        TableEvent tableEvent = (TableEvent) event;
+        TableId tableId = tableEvent.tableId();
+        boolean isIncludedByFilter = tableFilter.isIncluded(tableId);
+        boolean isGhostTable = parserOnlineDDL
+                && tableId != null
+                && GHOST_TABLE_PATTERN.matcher(tableId.table()).matches();
+        LOGGER.info("isIncludedByFilter:{},isGhostTable:{},result:{}", isIncludedByFilter, isGhostTable,isIncludedByFilter || isGhostTable);
+        return isIncludedByFilter || isGhostTable;
+      }
+      if (event instanceof SetVariableEvent) {
+        SetVariableEvent varEvent = (SetVariableEvent) event;
+        return !varEvent.databaseName().isPresent()
+                || (varEvent.databaseName().isPresent()
+                && databaseFilter.test(varEvent.databaseName().get()));
+      }
+      return false;
+    });
   }
-
 }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlDatabaseSchema.java

@@ -112,9 +112,9 @@ public MySqlDatabaseSchema(MySqlConnectorConfig connectorConfig, MySqlValueConve
             valueConverter,
             getTableFilter());
     this.ddlChanges = this.ddlParser.getDdlChanges();
+    this.parseOnLineSchemaChanges = parseOnLineSchemaChanges;
     this.connectorConfig = connectorConfig;
     filters = connectorConfig.getTableFilters();
-    this.parseOnLineSchemaChanges = parseOnLineSchemaChanges;
   }
 
   /**
@@ -158,11 +158,13 @@ private boolean isGhostTable(TableId tableId) {
    * table to schema change history.
    */
   private boolean isTableIncluded(String MethodName,TableId tableId) {
-    // 先应用原有过滤规则
     boolean isIncludedByOriginalFilter = filters.dataCollectionFilter().isIncluded(tableId);
+    LOGGER.info("MethodName: {},Including isIncludedByOriginalFilter table that would otherwise be filtered: {}," +
+                    "isIncludedByOriginalFilter:{}",
+            MethodName, tableId,isIncludedByOriginalFilter);
     if (parseOnLineSchemaChanges) {
       if (!isIncludedByOriginalFilter && isGhostTable(tableId)) {
-        LOGGER.info("MethodName{},Including gh-ost table that would otherwise be filtered: {}",MethodName, tableId);
+        LOGGER.info("MethodName: {},Including gh-ost table that would otherwise be filtered: {}",MethodName, tableId);
         return true;
       }
     }
@@ -226,7 +228,10 @@ public void applySchemaChange(SchemaChangeEvent schemaChange) {
     // - or DDLs for monitored objects
     if (!databaseHistory.storeOnlyCapturedTables() || isGlobalSetVariableStatement(schemaChange.getDdl(), schemaChange.getDatabase())
             || schemaChange.getTables().stream().map(Table::id).anyMatch(x->isTableIncluded("applySchemaChange",x))) {
-      LOGGER.info("Recorded DDL statements for database '{}': {}", schemaChange.getDatabase(), schemaChange.getDdl());
+      boolean applySchemaChange = schemaChange.getTables().stream().map(Table::id).anyMatch(x -> isTableIncluded("applySchemaChange", x));
+      LOGGER.info("Recorded DDL statements for database '{}': {},{},{},{}", schemaChange.getDatabase(),
+              schemaChange.getDdl()
+              ,applySchemaChange,!databaseHistory.storeOnlyCapturedTables(),isGlobalSetVariableStatement(schemaChange.getDdl(), schemaChange.getDatabase()));
       record(schemaChange, schemaChange.getTableChanges());
     }
   }
@@ -239,7 +244,7 @@ public List<SchemaChangeEvent> parseSnapshotDdl(MySqlPartition partition, String
 
   public List<SchemaChangeEvent> parseStreamingDdl(MySqlPartition partition, String ddlStatements, String databaseName,
                                                    MySqlOffsetContext offset, Instant sourceTime) {
-    LOGGER.debug("Processing streaming DDL '{}' for database '{}'", ddlStatements, databaseName);
+    LOGGER.info("Processing streaming DDL '{}' for database '{}'", ddlStatements, databaseName);
     return parseDdl(partition, ddlStatements, databaseName, offset, sourceTime, false);
   }
 
@@ -265,10 +270,7 @@ private List<SchemaChangeEvent> parseDdl(MySqlPartition partition, String ddlSta
       }
     }
     // No need to send schema events or store DDL if no table has changed
-    if (!databaseHistory.storeOnlyCapturedTables() || isGlobalSetVariableStatement(ddlStatements, databaseName) || ddlChanges.anyMatch(event -> {
-      TableId tableId = this.getTableId(event);
-      return tableId != null && isTableIncluded("parseDdl",tableId);
-    })) {
+    if (!databaseHistory.storeOnlyCapturedTables() || isGlobalSetVariableStatement(ddlStatements, databaseName) || ddlChanges.anyMatch(filters,parseOnLineSchemaChanges)){
 
       // We are supposed to _also_ record the schema changes as SourceRecords, but these need to be filtered
       // by database. Unfortunately, the databaseName on the event might not be the same database as that
@@ -407,13 +409,12 @@ public boolean storeOnlyCapturedTables() {
    *         connector's configuration
    */
   public boolean assignTableNumber(long tableNumber, TableId id) {
-
+    if (!isTableIncluded("assignTableNumber",id)) {
+      excludeTableIdsByTableNumber.put(tableNumber, id);
+      return false;
+    }
     final TableSchema tableSchema = schemaFor(id);
     if (tableSchema == null) {
-      if (isTableIncluded("assignTableNumber",id)) {
-        tableIdsByTableNumber.put(tableNumber, id);
-        return true;
-      }
       excludeTableIdsByTableNumber.put(tableNumber, id);
       return false;
     }