[pipeline][mysql] skip binlog deserialization for non-captured table

Author: guoxuanlin

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java

@@ -78,6 +78,7 @@
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_BINLOG_NEWLY_ADDED_TABLE_ENABLED;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP;
+import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_ONLY_DESERIALIZE_CAPTURED_TABLES_CHANGELOG_ENABLED;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE;
 import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED;
@@ -167,6 +168,8 @@ public DataSource createDataSource(Context context) {
         boolean useLegacyJsonFormat = config.get(USE_LEGACY_JSON_FORMAT);
         boolean isAssignUnboundedChunkFirst =
                 config.get(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED);
+        boolean onlyDeserializeCapturedTablesChangelog =
+                config.get(SCAN_ONLY_DESERIALIZE_CAPTURED_TABLES_CHANGELOG_ENABLED);
 
         validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
         validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1);
@@ -220,7 +223,9 @@ public DataSource createDataSource(Context context) {
                         .treatTinyInt1AsBoolean(treatTinyInt1AsBoolean)
                         .useLegacyJsonFormat(useLegacyJsonFormat)
                         .assignUnboundedChunkFirst(isAssignUnboundedChunkFirst)
-                        .skipSnapshotBackfill(skipSnapshotBackfill);
+                        .skipSnapshotBackfill(skipSnapshotBackfill)
+                        .onlyDeserializeCapturedTablesChangelog(
+                                onlyDeserializeCapturedTablesChangelog);
 
         List<TableId> tableIds = MySqlSchemaUtils.listTables(configFactory.createConfig(0), null);
 
@@ -358,6 +363,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         options.add(PARSE_ONLINE_SCHEMA_CHANGES);
         options.add(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED);
         options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
+        options.add(SCAN_ONLY_DESERIALIZE_CAPTURED_TABLES_CHANGELOG_ENABLED);
         return options;
     }
 

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

@@ -330,4 +330,16 @@ public class MySqlDataSourceOptions {
                     .defaultValue(false)
                     .withDescription(
                             "Whether to skip backfill in snapshot reading phase. If backfill is skipped, changes on captured tables during snapshot phase will be consumed later in change log reading phase instead of being merged into the snapshot.WARNING: Skipping backfill might lead to data inconsistency because some change log events happened within the snapshot phase might be replayed (only at-least-once semantic is promised). For example updating an already updated value in snapshot, or deleting an already deleted entry in snapshot. These replayed change log events should be handled specially.");
+
+    @Experimental
+    public static final ConfigOption<Boolean>
+            SCAN_ONLY_DESERIALIZE_CAPTURED_TABLES_CHANGELOG_ENABLED =
+                    ConfigOptions.key(
+                                    "scan.only.deserialize.captured.tables.changelog.enabled")
+                            .booleanType()
+                            .defaultValue(false)
+                            .withDescription(
+                                    "Whether to only deserialize changelog events for captured tables during incremental phase. "
+                                            + "When set to true, only changelog events for the target tables will be deserialized, "
+                                            + "which can speed up binlog reading. Defaults to false.");
 }

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

@@ -0,0 +1,119 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.debezium.connector.mysql;
+
+import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData;
+import com.github.shyiko.mysql.binlog.event.EventData;
+import com.github.shyiko.mysql.binlog.event.TableMapEventData;
+import com.github.shyiko.mysql.binlog.event.UpdateRowsEventData;
+import com.github.shyiko.mysql.binlog.event.WriteRowsEventData;
+import com.github.shyiko.mysql.binlog.event.deserialization.EventDataDeserializer;
+import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream;
+import io.debezium.relational.TableId;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Predicate;
+
+/**
+ * A filtering wrapper around row event data deserializers (WRITE_ROWS, UPDATE_ROWS, DELETE_ROWS).
+ *
+ * <p>When enabled, this deserializer reads the table ID from the first 6 bytes of the event data,
+ * looks up the table name via the TABLE_MAP event cache, and checks whether the table is in the
+ * captured table list. If the table is NOT captured, it returns an empty EventData (with tableId set
+ * but rows empty) to skip the expensive row-by-row deserialization. The remaining bytes are
+ * automatically skipped by {@code EventDeserializer.deserializeEventData()} via {@code
+ * skipToTheEndOfTheBlock()}.
+ *
+ * <p>If the table IS captured, it resets the stream position and delegates to the original
+ * deserializer for full deserialization.
+ */
+public class FilteringRowsEventDataDeserializer<T extends EventData>
+        implements EventDataDeserializer<T> {
+
+    private static final Logger LOGGER =
+            LoggerFactory.getLogger(FilteringRowsEventDataDeserializer.class);
+
+    /** The type of row event this deserializer handles. */
+    public enum RowEventType {
+        WRITE,
+        UPDATE,
+        DELETE
+    }
+
+    private final EventDataDeserializer<T> delegate;
+    private final Map<Long, TableMapEventData> tableMapEventByTableId;
+    private final Predicate<TableId> tableFilter;
+    private final RowEventType rowEventType;
+
+    public FilteringRowsEventDataDeserializer(
+            EventDataDeserializer<T> delegate,
+            Map<Long, TableMapEventData> tableMapEventByTableId,
+            Predicate<TableId> tableFilter,
+            RowEventType rowEventType) {
+        this.delegate = delegate;
+        this.tableMapEventByTableId = tableMapEventByTableId;
+        this.tableFilter = tableFilter;
+        this.rowEventType = rowEventType;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public T deserialize(ByteArrayInputStream inputStream) throws IOException {
+        // Mark the current position so we can reset if the table is captured
+        inputStream.mark(Integer.MAX_VALUE);
+
+        // Read the 6-byte table number (same format as all row events)
+        long tableNumber = inputStream.readLong(6);
+
+        // Look up the table name from the TABLE_MAP event cache
+        TableMapEventData tableMapEvent = tableMapEventByTableId.get(tableNumber);
+        if (tableMapEvent != null) {
+            String database = tableMapEvent.getDatabase();
+            String table = tableMapEvent.getTable();
+            TableId tableId = new TableId(database, null, table);
+
+            if (!tableFilter.test(tableId)) {
+                // Table is NOT captured, skip deserialization by returning empty EventData
+                LOGGER.trace(
+                        "Skipping deserialization for non-captured table: {}.{}", database, table);
+                return createEmptyEventData(tableNumber);
+            }
+        }
+
+        // Table IS captured (or unknown), reset and delegate to original deserializer
+        inputStream.reset();
+        return delegate.deserialize(inputStream);
+    }
+
+    @SuppressWarnings("unchecked")
+    private T createEmptyEventData(long tableNumber) {
+        switch (rowEventType) {
+            case WRITE:
+                WriteRowsEventData writeData = new WriteRowsEventData();
+                writeData.setTableId(tableNumber);
+                writeData.setRows(Collections.emptyList());
+                return (T) writeData;
+            case UPDATE:
+                UpdateRowsEventData updateData = new UpdateRowsEventData();
+                updateData.setTableId(tableNumber);
+                updateData.setRows(Collections.emptyList());
+                return (T) updateData;
+            case DELETE:
+                DeleteRowsEventData deleteData = new DeleteRowsEventData();
+                deleteData.setTableId(tableNumber);
+                deleteData.setRows(Collections.emptyList());
+                return (T) deleteData;
+            default:
+                throw new IllegalStateException("Unknown row event type: " + rowEventType);
+        }
+    }
+}

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

@@ -131,6 +131,8 @@ public class MySqlStreamingChangeEventSource
     private final MySqlConnection connection;
     private final EventDispatcher<MySqlPartition, TableId> eventDispatcher;
     private final ErrorHandler errorHandler;
+    private final boolean onlyDeserializeCapturedTablesChangelog;
+    private final Predicate<TableId> capturedTableFilter;
 
     @SingleThreadAccess("binlog client thread")
     private Instant eventTimestamp;
@@ -204,6 +206,28 @@ public MySqlStreamingChangeEventSource(
             Clock clock,
             MySqlTaskContext taskContext,
             MySqlStreamingChangeEventSourceMetrics metrics) {
+        this(
+                connectorConfig,
+                connection,
+                dispatcher,
+                errorHandler,
+                clock,
+                taskContext,
+                metrics,
+                false,
+                null);
+    }
+
+    public MySqlStreamingChangeEventSource(
+            MySqlConnectorConfig connectorConfig,
+            MySqlConnection connection,
+            EventDispatcher<MySqlPartition, TableId> dispatcher,
+            ErrorHandler errorHandler,
+            Clock clock,
+            MySqlTaskContext taskContext,
+            MySqlStreamingChangeEventSourceMetrics metrics,
+            boolean onlyDeserializeCapturedTablesChangelog,
+            Predicate<TableId> capturedTableFilter) {
 
         this.taskContext = taskContext;
         this.connectorConfig = connectorConfig;
@@ -212,6 +236,8 @@ public MySqlStreamingChangeEventSource(
         this.eventDispatcher = dispatcher;
         this.errorHandler = errorHandler;
         this.metrics = metrics;
+        this.onlyDeserializeCapturedTablesChangelog = onlyDeserializeCapturedTablesChangelog;
+        this.capturedTableFilter = capturedTableFilter;
 
         eventDeserializationFailureHandlingMode =
                 connectorConfig.getEventProcessingFailureHandlingMode();
@@ -319,27 +345,87 @@ public Event nextEvent(ByteArrayInputStream inputStream) throws IOException {
         // Add our custom deserializers ...
         eventDeserializer.setEventDataDeserializer(EventType.STOP, new StopEventDataDeserializer());
         eventDeserializer.setEventDataDeserializer(EventType.GTID, new GtidEventDataDeserializer());
-        eventDeserializer.setEventDataDeserializer(
-                EventType.WRITE_ROWS,
-                new RowDeserializers.WriteRowsDeserializer(tableMapEventByTableId));
-        eventDeserializer.setEventDataDeserializer(
-                EventType.UPDATE_ROWS,
-                new RowDeserializers.UpdateRowsDeserializer(tableMapEventByTableId));
-        eventDeserializer.setEventDataDeserializer(
-                EventType.DELETE_ROWS,
-                new RowDeserializers.DeleteRowsDeserializer(tableMapEventByTableId));
-        eventDeserializer.setEventDataDeserializer(
-                EventType.EXT_WRITE_ROWS,
-                new RowDeserializers.WriteRowsDeserializer(tableMapEventByTableId)
-                        .setMayContainExtraInformation(true));
-        eventDeserializer.setEventDataDeserializer(
-                EventType.EXT_UPDATE_ROWS,
-                new RowDeserializers.UpdateRowsDeserializer(tableMapEventByTableId)
-                        .setMayContainExtraInformation(true));
-        eventDeserializer.setEventDataDeserializer(
-                EventType.EXT_DELETE_ROWS,
-                new RowDeserializers.DeleteRowsDeserializer(tableMapEventByTableId)
-                        .setMayContainExtraInformation(true));
+
+        // Create row deserializers, optionally wrapped with filtering for non-captured tables
+        RowDeserializers.WriteRowsDeserializer writeRowsDeserializer =
+                new RowDeserializers.WriteRowsDeserializer(tableMapEventByTableId);
+        RowDeserializers.UpdateRowsDeserializer updateRowsDeserializer =
+                new RowDeserializers.UpdateRowsDeserializer(tableMapEventByTableId);
+        RowDeserializers.DeleteRowsDeserializer deleteRowsDeserializer =
+                new RowDeserializers.DeleteRowsDeserializer(tableMapEventByTableId);
+        RowDeserializers.WriteRowsDeserializer extWriteRowsDeserializer =
+                (RowDeserializers.WriteRowsDeserializer)
+                        new RowDeserializers.WriteRowsDeserializer(tableMapEventByTableId)
+                                .setMayContainExtraInformation(true);
+        RowDeserializers.UpdateRowsDeserializer extUpdateRowsDeserializer =
+                (RowDeserializers.UpdateRowsDeserializer)
+                        new RowDeserializers.UpdateRowsDeserializer(tableMapEventByTableId)
+                                .setMayContainExtraInformation(true);
+        RowDeserializers.DeleteRowsDeserializer extDeleteRowsDeserializer =
+                (RowDeserializers.DeleteRowsDeserializer)
+                        new RowDeserializers.DeleteRowsDeserializer(tableMapEventByTableId)
+                                .setMayContainExtraInformation(true);
+
+        if (onlyDeserializeCapturedTablesChangelog && capturedTableFilter != null) {
+            LOGGER.debug(
+                    "Only deserializing changelog events for captured tables is enabled. "
+                            + "Non-captured table row events will skip deserialization.");
+            eventDeserializer.setEventDataDeserializer(
+                    EventType.WRITE_ROWS,
+                    new FilteringRowsEventDataDeserializer<>(
+                            writeRowsDeserializer,
+                            tableMapEventByTableId,
+                            capturedTableFilter,
+                            FilteringRowsEventDataDeserializer.RowEventType.WRITE));
+            eventDeserializer.setEventDataDeserializer(
+                    EventType.UPDATE_ROWS,
+                    new FilteringRowsEventDataDeserializer<>(
+                            updateRowsDeserializer,
+                            tableMapEventByTableId,
+                            capturedTableFilter,
+                            FilteringRowsEventDataDeserializer.RowEventType.UPDATE));
+            eventDeserializer.setEventDataDeserializer(
+                    EventType.DELETE_ROWS,
+                    new FilteringRowsEventDataDeserializer<>(
+                            deleteRowsDeserializer,
+                            tableMapEventByTableId,
+                            capturedTableFilter,
+                            FilteringRowsEventDataDeserializer.RowEventType.DELETE));
+            eventDeserializer.setEventDataDeserializer(
+                    EventType.EXT_WRITE_ROWS,
+                    new FilteringRowsEventDataDeserializer<>(
+                            extWriteRowsDeserializer,
+                            tableMapEventByTableId,
+                            capturedTableFilter,
+                            FilteringRowsEventDataDeserializer.RowEventType.WRITE));
+            eventDeserializer.setEventDataDeserializer(
+                    EventType.EXT_UPDATE_ROWS,
+                    new FilteringRowsEventDataDeserializer<>(
+                            extUpdateRowsDeserializer,
+                            tableMapEventByTableId,
+                            capturedTableFilter,
+                            FilteringRowsEventDataDeserializer.RowEventType.UPDATE));
+            eventDeserializer.setEventDataDeserializer(
+                    EventType.EXT_DELETE_ROWS,
+                    new FilteringRowsEventDataDeserializer<>(
+                            extDeleteRowsDeserializer,
+                            tableMapEventByTableId,
+                            capturedTableFilter,
+                            FilteringRowsEventDataDeserializer.RowEventType.DELETE));
+        } else {
+            eventDeserializer.setEventDataDeserializer(
+                    EventType.WRITE_ROWS, writeRowsDeserializer);
+            eventDeserializer.setEventDataDeserializer(
+                    EventType.UPDATE_ROWS, updateRowsDeserializer);
+            eventDeserializer.setEventDataDeserializer(
+                    EventType.DELETE_ROWS, deleteRowsDeserializer);
+            eventDeserializer.setEventDataDeserializer(
+                    EventType.EXT_WRITE_ROWS, extWriteRowsDeserializer);
+            eventDeserializer.setEventDataDeserializer(
+                    EventType.EXT_UPDATE_ROWS, extUpdateRowsDeserializer);
+            eventDeserializer.setEventDataDeserializer(
+                    EventType.EXT_DELETE_ROWS, extDeleteRowsDeserializer);
+        }
         client.setEventDeserializer(eventDeserializer);
     }
 

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

@@ -137,7 +137,11 @@ public void submitSplit(MySqlSplit mySqlSplit) {
                         (MySqlStreamingChangeEventSourceMetrics)
                                 statefulTaskContext.getStreamingChangeEventSourceMetrics(),
                         currentBinlogSplit,
-                        createEventFilter());
+                        createEventFilter(),
+                        statefulTaskContext
+                                .getSourceConfig()
+                                .isOnlyDeserializeCapturedTablesChangelog(),
+                        statefulTaskContext.getSourceConfig().getTableFilter());
 
         executorService.submit(
                 () -> {

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

@@ -69,7 +69,44 @@ public MySqlBinlogSplitReadTask(
             MySqlStreamingChangeEventSourceMetrics metrics,
             MySqlBinlogSplit binlogSplit,
             Predicate<Event> eventFilter) {
-        super(connectorConfig, connection, dispatcher, errorHandler, clock, taskContext, metrics);
+        this(
+                connectorConfig,
+                connection,
+                dispatcher,
+                signalEventDispatcher,
+                errorHandler,
+                clock,
+                taskContext,
+                metrics,
+                binlogSplit,
+                eventFilter,
+                false,
+                null);
+    }
+
+    public MySqlBinlogSplitReadTask(
+            MySqlConnectorConfig connectorConfig,
+            MySqlConnection connection,
+            EventDispatcherImpl<TableId> dispatcher,
+            SignalEventDispatcher signalEventDispatcher,
+            ErrorHandler errorHandler,
+            Clock clock,
+            MySqlTaskContext taskContext,
+            MySqlStreamingChangeEventSourceMetrics metrics,
+            MySqlBinlogSplit binlogSplit,
+            Predicate<Event> eventFilter,
+            boolean onlyDeserializeCapturedTablesChangelog,
+            Predicate<TableId> capturedTableFilter) {
+        super(
+                connectorConfig,
+                connection,
+                dispatcher,
+                errorHandler,
+                clock,
+                taskContext,
+                metrics,
+                onlyDeserializeCapturedTablesChangelog,
+                capturedTableFilter);
         this.binlogSplit = binlogSplit;
         this.eventDispatcher = dispatcher;
         this.errorHandler = errorHandler;