[Flink-38911][cdc connector mysql] mysql-cdc-connector datastream support scan.binlog.newly-added-table.enabled

docs/content.zh/docs/connectors/flink-sources/mysql-cdc.md

@@ -398,6 +398,21 @@ Flink SQL> SELECT * FROM orders;
       <td>Boolean</td>
       <td>是否在快照结束后关闭空闲的 Reader。 此特性需要 flink 版本大于等于 1.14 并且 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' 需要设置为 true。</td>
     </tr>
+    <tr>
+      <td>scan.binlog.newly-added-table.enabled</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">false</td>
+      <td>Boolean</td>
+      <td>不能与 <code>scan.newly-added-table.enabled</code> 同时启用。<br>
+          <b><code>table-name</code>参数 匹配模式示例：</b>
+          <ul>
+            <li><code>db\.*</code> - 捕获数据库 'db' 下的所有表</li>
+            <li><code>db\.user_\.*</code> - 捕获数据库 'db' 下类似 'user_orders'、'user_profiles' 的表</li>
+            <li><code>db\.order_[0-9]+</code> - 捕获数据库 'db' 下类似 'order_1'、'order_2' 的表</li>
+            <li><code>db1\.*,db2\.user_\.*</code> - 捕获 'db1' 下所有表和 'db2' 下 'user_*' 开头的表</li>
+          </ul>
+      </td>
+    </tr>
     <tr>
       <td>scan.parse.online.schema.changes.enabled</td>
       <td>optional</td>

docs/content/docs/connectors/flink-sources/mysql-cdc.md

@@ -400,6 +400,21 @@ Only valid for cdc 1.x version. During a snapshot operation, the connector will
           so it does not need to be explicitly configured 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true'
       </td>
     </tr>
+    <tr>
+      <td>scan.binlog.newly-added-table.enabled</td>
+      <td>optional</td>
+      <td style="word-wrap: break-word;">false</td>
+      <td>Boolean</td>
+      <td>Cannot be enabled together with <code>scan.newly-added-table.enabled</code>. <br>
+          <b><code>table-name</code> config Examples:</b>
+          <ul>
+            <li><code>db\.*</code> - captures all tables in database 'db'</li>
+            <li><code>db\.user_\.*</code> - captures tables like 'user_orders', 'user_profiles' in database 'db'</li>
+            <li><code>db\.order_[0-9]+</code> - captures tables like 'order_1', 'order_2' in database 'db'</li>
+            <li><code>db1\.*,db2\.user_\.*</code> - captures all tables in 'db1' and 'user_*' tables in 'db2'</li>
+          </ul>
+      </td>
+    </tr>
     <tr>
       <td>debezium.binary.handling.mode</td>
       <td>optional</td>

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java

@@ -335,14 +335,25 @@ private boolean shouldEmit(SourceRecord sourceRecord) {
     }
 
     private boolean hasEnterPureBinlogPhase(TableId tableId, BinlogOffset position) {
-        // the existed tables those have finished snapshot reading
+        // Case 1: the existed tables those have finished snapshot reading
         if (maxSplitHighWatermarkMap.containsKey(tableId)
                 && position.isAfter(maxSplitHighWatermarkMap.get(tableId))) {
             pureBinlogPhaseTables.add(tableId);
             return true;
         }
 
-        // Use still need to capture new sharding table if user disable scan new added table,
+        // Case 2: binlog-only mode for newly added tables
+        // Capture new tables that match the filter without snapshot
+        if (statefulTaskContext.getSourceConfig().isScanBinlogNewlyAddedTableEnabled()) {
+            if (!maxSplitHighWatermarkMap.containsKey(tableId)
+                    && capturedTableFilter.test(tableId)) {
+                LOG.info("Auto-capturing newly added table in binlog-only mode: {}", tableId);
+                pureBinlogPhaseTables.add(tableId);
+                return true;
+            }
+        }
+
+        // Case 3: Use still need to capture new sharding table if user disable scan new added table,
         // The history records for all new added tables(including sharding table and normal table)
         // will be capture after restore from a savepoint if user enable scan new added table
         if (!statefulTaskContext.getSourceConfig().isScanNewlyAddedTableEnabled()) {

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.cdc.connectors.mysql.source;
 
+import org.apache.flink.cdc.common.annotation.Experimental;
 import org.apache.flink.cdc.common.annotation.PublicEvolving;
 import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
 import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
@@ -211,6 +212,18 @@ public MySqlSourceBuilder<T> scanNewlyAddedTableEnabled(boolean scanNewlyAddedTa
         return this;
     }
 
+    /**
+     * Whether to capture newly added tables in binlog reading phase without snapshot. This option
+     * can only be used with stream-only startup modes. Cannot be enabled together with {@link
+     * #scanNewlyAddedTableEnabled(boolean)}.
+     */
+    @Experimental
+    public MySqlSourceBuilder<T> scanBinlogNewlyAddedTableEnabled(
+            boolean scanBinlogNewlyAddedTableEnabled) {
+        this.configFactory.scanBinlogNewlyAddedTableEnabled(scanBinlogNewlyAddedTableEnabled);
+        return this;
+    }
+
     /** Specifies the startup options. */
     public MySqlSourceBuilder<T> startupOptions(StartupOptions startupOptions) {
         this.configFactory.startupOptions(startupOptions);

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java

@@ -65,6 +65,7 @@ public class MySqlSourceConfig implements Serializable {
     private final boolean includeHeartbeatEvents;
     private final boolean includeTransactionMetadataEvents;
     private final boolean scanNewlyAddedTableEnabled;
+    private final boolean scanBinlogNewlyAddedTableEnabled;
     private final boolean closeIdleReaders;
     private final Properties jdbcProperties;
     private final Map<ObjectPath, String> chunkKeyColumns;
@@ -104,6 +105,7 @@ public class MySqlSourceConfig implements Serializable {
             boolean includeHeartbeatEvents,
             boolean includeTransactionMetadataEvents,
             boolean scanNewlyAddedTableEnabled,
+            boolean scanBinlogNewlyAddedTableEnabled,
             boolean closeIdleReaders,
             Properties dbzProperties,
             Properties jdbcProperties,
@@ -135,6 +137,7 @@ public class MySqlSourceConfig implements Serializable {
         this.includeHeartbeatEvents = includeHeartbeatEvents;
         this.includeTransactionMetadataEvents = includeTransactionMetadataEvents;
         this.scanNewlyAddedTableEnabled = scanNewlyAddedTableEnabled;
+        this.scanBinlogNewlyAddedTableEnabled = scanBinlogNewlyAddedTableEnabled;
         this.closeIdleReaders = closeIdleReaders;
         this.dbzProperties = checkNotNull(dbzProperties);
         this.dbzConfiguration = Configuration.from(dbzProperties);
@@ -245,6 +248,10 @@ public boolean isScanNewlyAddedTableEnabled() {
         return scanNewlyAddedTableEnabled;
     }
 
+    public boolean isScanBinlogNewlyAddedTableEnabled() {
+        return scanBinlogNewlyAddedTableEnabled;
+    }
+
     public boolean isCloseIdleReaders() {
         return closeIdleReaders;
     }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java

@@ -18,6 +18,7 @@
 package org.apache.flink.cdc.connectors.mysql.source.config;
 
 import org.apache.flink.cdc.common.annotation.Internal;
+import org.apache.flink.cdc.common.route.TableIdRouter;
 import org.apache.flink.cdc.connectors.mysql.debezium.EmbeddedFlinkDatabaseHistory;
 import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;
 import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
@@ -68,6 +69,7 @@ public class MySqlSourceConfigFactory implements Serializable {
     private boolean includeHeartbeatEvents = false;
     private boolean includeTransactionMetadataEvents = false;
     private boolean scanNewlyAddedTableEnabled = false;
+    private boolean scanBinlogNewlyAddedTableEnabled = false;
     private boolean closeIdleReaders = false;
     private Properties jdbcProperties;
     private Duration heartbeatInterval = MySqlSourceOptions.HEARTBEAT_INTERVAL.defaultValue();
@@ -258,6 +260,17 @@ public MySqlSourceConfigFactory scanNewlyAddedTableEnabled(boolean scanNewlyAdde
         return this;
     }
 
+    /**
+     * Whether to capture newly added tables in binlog reading phase without snapshot. This option
+     * can only be used with stream-only startup modes. Cannot be enabled together with {@link
+     * #scanNewlyAddedTableEnabled(boolean)}.
+     */
+    public MySqlSourceConfigFactory scanBinlogNewlyAddedTableEnabled(
+            boolean scanBinlogNewlyAddedTableEnabled) {
+        this.scanBinlogNewlyAddedTableEnabled = scanBinlogNewlyAddedTableEnabled;
+        return this;
+    }
+
     /** Custom properties that will overwrite the default JDBC connection URL. */
     public MySqlSourceConfigFactory jdbcProperties(Properties jdbcProperties) {
         this.jdbcProperties = jdbcProperties;
@@ -397,8 +410,21 @@ public MySqlSourceConfig createConfig(int subtaskId, String serverName) {
         if (databaseList != null) {
             props.setProperty("database.include.list", String.join(",", databaseList));
         }
+        // Validate: Two modes are mutually exclusive
+        if (scanBinlogNewlyAddedTableEnabled && scanNewlyAddedTableEnabled) {
+            throw new IllegalArgumentException(
+                    "Cannot enable both 'scan.binlog.newly-added-table.enabled' and "
+                            + "'scan.newly-added-table.enabled' as they may cause duplicate data");
+        }
+
         if (tableList != null) {
-            props.setProperty("table.include.list", String.join(",", tableList));
+            if (scanBinlogNewlyAddedTableEnabled) {
+                props.setProperty(
+                        "table.include.list",
+                        TableIdRouter.convertTableListToRegExpPattern(String.join(",", tableList)));
+            } else {
+                props.setProperty("table.include.list", String.join(",", tableList));
+            }
         }
         if (serverTimeZone != null) {
             props.setProperty("database.serverTimezone", serverTimeZone);
@@ -436,6 +462,7 @@ public MySqlSourceConfig createConfig(int subtaskId, String serverName) {
                 includeHeartbeatEvents,
                 includeTransactionMetadataEvents,
                 scanNewlyAddedTableEnabled,
+                scanBinlogNewlyAddedTableEnabled,
                 closeIdleReaders,
                 props,
                 jdbcProperties,

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceOptions.java

@@ -243,6 +243,22 @@ public class MySqlSourceOptions {
                     .withDescription(
                             "Whether capture the scan the newly added tables or not, by default is false. This option is only useful when we start the job from a savepoint/checkpoint.");
 
+    @Experimental
+    public static final ConfigOption<Boolean> SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED =
+            ConfigOptions.key("scan.binlog.newly-added-table.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "In binlog reading stage, whether to capture newly added tables "
+                                    + "that match the table patterns. When enabled, new tables will be "
+                                    + "captured without snapshot, only binlog events will be emitted. "
+                                    + "Cannot be enabled together with 'scan.newly-added-table.enabled'. "
+                                    + "table-name pattern examples: "
+                                    + "'db.\\.*' (all tables in database 'db'), "
+                                    + "'db.user_\\.*' (tables like 'user_orders', 'user_profiles'), "
+                                    + "'db.order_[0-9]+' (tables like 'order_1', 'order_2'), "
+                                    + "'db1.\\.*,db2.user_\\.*' (all tables in 'db1' and 'user_*' tables in 'db2').");
+
     @Experimental
     public static final ConfigOption<String> SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN =
             ConfigOptions.key("scan.incremental.snapshot.chunk.key-column")

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSource.java

@@ -93,6 +93,7 @@ public class MySqlTableSource implements ScanTableSource, SupportsReadingMetadat
     private final double distributionFactorLower;
     private final StartupOptions startupOptions;
     private final boolean scanNewlyAddedTableEnabled;
+    private final boolean scanBinlogNewlyAddedTableEnabled;
     private final boolean closeIdleReaders;
     private final Properties jdbcProperties;
     private final Duration heartbeatInterval;
@@ -136,6 +137,7 @@ public MySqlTableSource(
             double distributionFactorLower,
             StartupOptions startupOptions,
             boolean scanNewlyAddedTableEnabled,
+            boolean scanBinlogNewlyAddedTableEnabled,
             boolean closeIdleReaders,
             Properties jdbcProperties,
             Duration heartbeatInterval,
@@ -166,6 +168,7 @@ public MySqlTableSource(
         this.distributionFactorLower = distributionFactorLower;
         this.startupOptions = startupOptions;
         this.scanNewlyAddedTableEnabled = scanNewlyAddedTableEnabled;
+        this.scanBinlogNewlyAddedTableEnabled = scanBinlogNewlyAddedTableEnabled;
         this.closeIdleReaders = closeIdleReaders;
         this.jdbcProperties = jdbcProperties;
         this.parseOnlineSchemaChanges = parseOnlineSchemaChanges;
@@ -233,6 +236,7 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
                             .startupOptions(startupOptions)
                             .deserializer(deserializer)
                             .scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled)
+                            .scanBinlogNewlyAddedTableEnabled(scanBinlogNewlyAddedTableEnabled)
                             .closeIdleReaders(closeIdleReaders)
                             .jdbcProperties(jdbcProperties)
                             .heartbeatInterval(heartbeatInterval)
@@ -322,6 +326,7 @@ public DynamicTableSource copy() {
                         distributionFactorLower,
                         startupOptions,
                         scanNewlyAddedTableEnabled,
+                        scanBinlogNewlyAddedTableEnabled,
                         closeIdleReaders,
                         jdbcProperties,
                         heartbeatInterval,
@@ -353,6 +358,7 @@ public boolean equals(Object o) {
                 && distributionFactorUpper == that.distributionFactorUpper
                 && distributionFactorLower == that.distributionFactorLower
                 && scanNewlyAddedTableEnabled == that.scanNewlyAddedTableEnabled
+                && scanBinlogNewlyAddedTableEnabled == that.scanBinlogNewlyAddedTableEnabled
                 && closeIdleReaders == that.closeIdleReaders
                 && Objects.equals(physicalSchema, that.physicalSchema)
                 && Objects.equals(hostname, that.hostname)
@@ -405,6 +411,7 @@ public int hashCode() {
                 producedDataType,
                 metadataKeys,
                 scanNewlyAddedTableEnabled,
+                scanBinlogNewlyAddedTableEnabled,
                 closeIdleReaders,
                 jdbcProperties,
                 heartbeatInterval,

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactory.java

@@ -90,6 +90,8 @@ public DynamicTableSource createDynamicTableSource(Context context) {
                 config.get(MySqlSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
         boolean scanNewlyAddedTableEnabled =
                 config.get(MySqlSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED);
+        boolean scanBinlogNewlyAddedTableEnabled =
+                config.get(MySqlSourceOptions.SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED);
         Duration heartbeatInterval = config.get(MySqlSourceOptions.HEARTBEAT_INTERVAL);
         String chunkKeyColumn =
                 config.getOptional(MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN)
@@ -148,6 +150,7 @@ public DynamicTableSource createDynamicTableSource(Context context) {
                 distributionFactorLower,
                 startupOptions,
                 scanNewlyAddedTableEnabled,
+                scanBinlogNewlyAddedTableEnabled,
                 closeIdleReaders,
                 JdbcUrlUtils.getJdbcProperties(context.getCatalogTable().getOptions()),
                 heartbeatInterval,
@@ -198,6 +201,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         options.add(MySqlSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
         options.add(MySqlSourceOptions.CONNECT_MAX_RETRIES);
         options.add(MySqlSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED);
+        options.add(MySqlSourceOptions.SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED);
         options.add(MySqlSourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
         options.add(MySqlSourceOptions.HEARTBEAT_INTERVAL);
         options.add(MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN);