From befc04e14a7f9ff68dc1493e03ebe822387f774d Mon Sep 17 00:00:00 2001 From: Eric Zeng Date: Wed, 15 Apr 2026 15:52:27 +0800 Subject: [PATCH 1/2] feat(mysql): Support batch emitting CreateTableEvents --- .../mysql/factory/MySqlDataSourceFactory.java | 5 + .../mysql/source/MySqlDataSourceOptions.java | 11 + .../reader/MySqlPipelineRecordEmitter.java | 20 +- .../MySqlPipelineRecordEmitterTest.java | 285 ++++++++++++++++++ .../source/config/MySqlSourceConfig.java | 9 +- .../config/MySqlSourceConfigFactory.java | 14 +- 6 files changed, 338 insertions(+), 6 deletions(-) create mode 100644 flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitterTest.java diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java index 1b9cea56182..4878a9c98ea 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java @@ -76,6 +76,7 @@ import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.PASSWORD; import static org.apache.flink.cdc.connectors.mysql.source.MySqlDataSourceOptions.PORT; 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_EMIT_CREATE_TABLE_EVENTS_IN_BATCH_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_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN; @@ -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 emitCreateTableEventsInBatchEnabled = + config.get(SCAN_EMIT_CREATE_TABLE_EVENTS_IN_BATCH_ENABLED); validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1); validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1); @@ -220,6 +223,7 @@ public DataSource createDataSource(Context context) { .treatTinyInt1AsBoolean(treatTinyInt1AsBoolean) .useLegacyJsonFormat(useLegacyJsonFormat) .assignUnboundedChunkFirst(isAssignUnboundedChunkFirst) + .emitCreateTableEventsInBatchEnabled(emitCreateTableEventsInBatchEnabled) .skipSnapshotBackfill(skipSnapshotBackfill); List tableIds = MySqlSchemaUtils.listTables(configFactory.createConfig(0), null); @@ -357,6 +361,7 @@ public Set> optionalOptions() { options.add(TREAT_TINYINT1_AS_BOOLEAN_ENABLED); options.add(PARSE_ONLINE_SCHEMA_CHANGES); options.add(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED); + options.add(SCAN_EMIT_CREATE_TABLE_EVENTS_IN_BATCH_ENABLED); options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP); return options; } diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java index 6aff556e7fa..0ddcf905b52 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java @@ -323,6 +323,17 @@ public class MySqlDataSourceOptions { .withDescription( "Whether to assign the unbounded chunks first during snapshot reading phase. This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk. Defaults to false."); + @Experimental + public static final ConfigOption SCAN_EMIT_CREATE_TABLE_EVENTS_IN_BATCH_ENABLED = + ConfigOptions.key("scan.emit.create-table-events.in-batch.enabled") + .booleanType() + .defaultValue(false) + .withDescription( + "Whether to emit all CreateTableEvents in batch during initialization phase. " + + "When enabled, all CreateTableEvents are collected and emitted together " + + "before processing data records, significantly reducing schema coordination " + + "overhead in scenarios with thousands of tables."); + @Experimental public static final ConfigOption SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP = ConfigOptions.key("scan.incremental.snapshot.backfill.skip") diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java index 41befa1f3d6..367cc2f1759 100644 --- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java @@ -84,6 +84,8 @@ public class MySqlPipelineRecordEmitter extends MySqlRecordEmitter { private boolean isBounded = false; private final boolean isTableIdCaseInsensitive; + private final boolean emitCreateTableEventsInBatch; + private final DebeziumDeserializationSchema debeziumDeserializationSchema; private final Map createTableEventCache; @@ -107,12 +109,20 @@ public MySqlPipelineRecordEmitter( .getCreateTableEventCache(); this.isBounded = StartupOptions.snapshot().equals(sourceConfig.getStartupOptions()); this.isTableIdCaseInsensitive = isTableIdCaseInsensitive; + this.emitCreateTableEventsInBatch = sourceConfig.isEmitCreateTableEventsInBatchEnabled(); + } + + private boolean shouldBatchEmitCreateTableEvents() { + return isBounded || emitCreateTableEventsInBatch; } @Override public void applySplit(MySqlSplit split) { - if ((isBounded) && createTableEventCache.isEmpty() && split instanceof MySqlSnapshotSplit) { - // TableSchemas in MySqlSnapshotSplit only contains one table. + if (shouldBatchEmitCreateTableEvents() + && createTableEventCache.isEmpty() + && split instanceof MySqlSnapshotSplit) { + // In snapshot mode or batch emit mode: batch emit ALL CreateTableEvent at split + // initialization createTableEventCache.putAll(generateCreateTableEvent(sourceConfig)); } else { for (TableChanges.TableChange tableChange : split.getTableSchemas().values()) { @@ -130,11 +140,13 @@ public void applySplit(MySqlSplit split) { protected void processElement( SourceRecord element, SourceOutput output, MySqlSplitState splitState) throws Exception { - if (shouldEmitAllCreateTableEventsInSnapshotMode && isBounded) { - // In snapshot mode, we simply emit all schemas at once. + if (shouldEmitAllCreateTableEventsInSnapshotMode + && shouldBatchEmitCreateTableEvents()) { + // In snapshot mode or batch emit mode, we emit all schemas at once. createTableEventCache.forEach( (tableId, createTableEvent) -> { output.collect(createTableEvent); + alreadySendCreateTableTables.add(tableId); }); shouldEmitAllCreateTableEventsInSnapshotMode = false; } else if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) { diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitterTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitterTest.java new file mode 100644 index 00000000000..7e7d86d9d81 --- /dev/null +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitterTest.java @@ -0,0 +1,285 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.mysql.source.reader; + +import org.apache.flink.cdc.common.event.CreateTableEvent; +import org.apache.flink.cdc.common.event.DataChangeEvent; +import org.apache.flink.cdc.common.event.Event; +import org.apache.flink.cdc.common.event.SchemaChangeEvent; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.schema.Schema; +import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig; +import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory; +import org.apache.flink.cdc.connectors.mysql.source.metrics.MySqlSourceReaderMetrics; +import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset; +import org.apache.flink.cdc.connectors.mysql.source.split.MySqlBinlogSplit; +import org.apache.flink.cdc.connectors.mysql.source.split.MySqlBinlogSplitState; +import org.apache.flink.cdc.connectors.mysql.source.split.MySqlSplitState; +import org.apache.flink.cdc.connectors.mysql.source.split.SourceRecords; +import org.apache.flink.cdc.debezium.event.DebeziumEventDeserializationSchema; +import org.apache.flink.cdc.debezium.event.DebeziumSchemaDataTypeInference; +import org.apache.flink.cdc.debezium.table.DebeziumChangelogMode; +import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput; +import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups; + +import io.debezium.relational.Tables; +import org.apache.kafka.connect.data.SchemaBuilder; +import org.apache.kafka.connect.data.Struct; +import org.apache.kafka.connect.source.SourceRecord; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit tests for {@link MySqlPipelineRecordEmitter}. */ +class MySqlPipelineRecordEmitterTest { + + @Test + void testBatchEmitCreateTableEventsWhenEnabled() throws Exception { + MySqlSourceConfig config = createSourceConfig(true); + DebeziumEventDeserializationSchema schema = createStubSchema(); + + // Pre-populate cache with CreateTableEvents for 3 tables + Map cache = + schema.getCreateTableEventCache(); + TableId table1 = TableId.tableId("test_db", "table1"); + TableId table2 = TableId.tableId("test_db", "table2"); + TableId table3 = TableId.tableId("test_db", "table3"); + cache.put( + new io.debezium.relational.TableId("test_db", null, "table1"), + createTableEvent(table1)); + cache.put( + new io.debezium.relational.TableId("test_db", null, "table2"), + createTableEvent(table2)); + cache.put( + new io.debezium.relational.TableId("test_db", null, "table3"), + createTableEvent(table3)); + + MySqlPipelineRecordEmitter emitter = + new MySqlPipelineRecordEmitter( + schema, + new MySqlSourceReaderMetrics( + UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup()), + config, + false); + + // Feed a dummy record to trigger batch emit + TestingReaderOutput output = new TestingReaderOutput<>(); + emitter.emitRecord( + SourceRecords.fromSingleRecord(createDummyRecord()), + output, + createBinlogSplitState()); + + List emitted = new ArrayList<>(output.getEmittedRecords()); + // Should emit exactly 3 CreateTableEvents (dummy record is unknown type, skipped by + // parent) + assertThat(emitted).hasSize(3); + assertThat(emitted.get(0)).isInstanceOf(CreateTableEvent.class); + assertThat(emitted.get(1)).isInstanceOf(CreateTableEvent.class); + assertThat(emitted.get(2)).isInstanceOf(CreateTableEvent.class); + } + + @Test + void testBatchEmitOnlyOnce() throws Exception { + MySqlSourceConfig config = createSourceConfig(true); + DebeziumEventDeserializationSchema schema = createStubSchema(); + + Map cache = + schema.getCreateTableEventCache(); + cache.put( + new io.debezium.relational.TableId("test_db", null, "table1"), + createTableEvent(TableId.tableId("test_db", "table1"))); + + MySqlPipelineRecordEmitter emitter = + new MySqlPipelineRecordEmitter( + schema, + new MySqlSourceReaderMetrics( + UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup()), + config, + false); + + // First record: should batch emit all CreateTableEvents + TestingReaderOutput output1 = new TestingReaderOutput<>(); + emitter.emitRecord( + SourceRecords.fromSingleRecord(createDummyRecord()), + output1, + createBinlogSplitState()); + assertThat(output1.getEmittedRecords()).hasSize(1); + + // Second record: no more batch emit (shouldEmitAllCreateTableEventsInSnapshotMode=false) + TestingReaderOutput output2 = new TestingReaderOutput<>(); + emitter.emitRecord( + SourceRecords.fromSingleRecord(createDummyRecord()), + output2, + createBinlogSplitState()); + assertThat(output2.getEmittedRecords()).isEmpty(); + } + + @Test + void testNoBatchEmitWhenDisabled() throws Exception { + MySqlSourceConfig config = createSourceConfig(false); + DebeziumEventDeserializationSchema schema = createStubSchema(); + + Map cache = + schema.getCreateTableEventCache(); + cache.put( + new io.debezium.relational.TableId("test_db", null, "table1"), + createTableEvent(TableId.tableId("test_db", "table1"))); + + MySqlPipelineRecordEmitter emitter = + new MySqlPipelineRecordEmitter( + schema, + new MySqlSourceReaderMetrics( + UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup()), + config, + false); + + // With batch disabled and not in snapshot mode, no batch emit should occur + TestingReaderOutput output = new TestingReaderOutput<>(); + emitter.emitRecord( + SourceRecords.fromSingleRecord(createDummyRecord()), + output, + createBinlogSplitState()); + // No CreateTableEvents emitted because batch is disabled and the record is unrecognized + assertThat(output.getEmittedRecords()).isEmpty(); + } + + @Test + void testBatchEmitEmptyCacheDoesNotEmit() throws Exception { + MySqlSourceConfig config = createSourceConfig(true); + DebeziumEventDeserializationSchema schema = createStubSchema(); + // Cache is empty + + MySqlPipelineRecordEmitter emitter = + new MySqlPipelineRecordEmitter( + schema, + new MySqlSourceReaderMetrics( + UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup()), + config, + false); + + TestingReaderOutput output = new TestingReaderOutput<>(); + emitter.emitRecord( + SourceRecords.fromSingleRecord(createDummyRecord()), + output, + createBinlogSplitState()); + // Cache is empty, so nothing is emitted + assertThat(output.getEmittedRecords()).isEmpty(); + } + + @Test + void testConfigDefaultFalse() { + MySqlSourceConfig config = createSourceConfig(false); + assertThat(config.isEmitCreateTableEventsInBatchEnabled()).isFalse(); + } + + @Test + void testConfigEnabledTrue() { + MySqlSourceConfig config = createSourceConfig(true); + assertThat(config.isEmitCreateTableEventsInBatchEnabled()).isTrue(); + } + + // ---- helpers ---- + + private static MySqlSourceConfig createSourceConfig(boolean emitCreateTableEventsInBatch) { + return new MySqlSourceConfigFactory() + .hostname("localhost") + .port(3306) + .username("test") + .password("test") + .databaseList("test_db") + .tableList("test_db\\.test_table") + .emitCreateTableEventsInBatchEnabled(emitCreateTableEventsInBatch) + .createConfig(0); + } + + private static DebeziumEventDeserializationSchema createStubSchema() { + return new DebeziumEventDeserializationSchema( + new DebeziumSchemaDataTypeInference(), DebeziumChangelogMode.ALL) { + @Override + protected boolean isDataChangeRecord(SourceRecord record) { + return false; + } + + @Override + protected boolean isSchemaChangeRecord(SourceRecord record) { + return false; + } + + @Override + public List deserializeDataChangeRecord(SourceRecord record) { + return Collections.emptyList(); + } + + @Override + public List deserializeSchemaChangeRecord(SourceRecord record) { + return Collections.emptyList(); + } + + @Override + protected org.apache.flink.cdc.common.event.TableId getTableId( + SourceRecord record) { + return null; + } + + @Override + protected Map getMetadata(SourceRecord record) { + return Collections.emptyMap(); + } + }; + } + + private static MySqlBinlogSplitState createBinlogSplitState() { + return new MySqlBinlogSplitState( + new MySqlBinlogSplit( + "binlog-split", + BinlogOffset.ofEarliest(), + BinlogOffset.ofNonStopping(), + Collections.emptyList(), + Collections.emptyMap(), + 0)); + } + + /** + * Creates a SourceRecord that is not recognized as any special type (not data change, not + * schema change, not watermark, etc.), so the parent emitter just skips it. This lets us test + * the batch CreateTableEvent emission in isolation. + */ + private static SourceRecord createDummyRecord() { + org.apache.kafka.connect.data.Schema valueSchema = + SchemaBuilder.struct().field("dummy", org.apache.kafka.connect.data.Schema.STRING_SCHEMA).build(); + Struct value = new Struct(valueSchema).put("dummy", "test"); + return new SourceRecord( + Collections.singletonMap("server", "mysql"), + Collections.singletonMap("file", "binlog.000001"), + "test_db.test_table", + null, + null, + valueSchema, + value); + } + + private static CreateTableEvent createTableEvent(TableId tableId) { + return new CreateTableEvent(tableId, Schema.newBuilder().build()); + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java index cf456fcaed0..e1b602e3e0b 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java @@ -72,6 +72,7 @@ public class MySqlSourceConfig implements Serializable { private final boolean parseOnLineSchemaChanges; public static boolean useLegacyJsonFormat = true; private final boolean assignUnboundedChunkFirst; + private final boolean emitCreateTableEventsInBatchEnabled; // -------------------------------------------------------------------------------------------- // Debezium Configurations @@ -112,7 +113,8 @@ public class MySqlSourceConfig implements Serializable { boolean parseOnLineSchemaChanges, boolean treatTinyInt1AsBoolean, boolean useLegacyJsonFormat, - boolean assignUnboundedChunkFirst) { + boolean assignUnboundedChunkFirst, + boolean emitCreateTableEventsInBatchEnabled) { this.hostname = checkNotNull(hostname); this.port = port; this.username = checkNotNull(username); @@ -158,6 +160,7 @@ public class MySqlSourceConfig implements Serializable { this.treatTinyInt1AsBoolean = treatTinyInt1AsBoolean; this.useLegacyJsonFormat = useLegacyJsonFormat; this.assignUnboundedChunkFirst = assignUnboundedChunkFirst; + this.emitCreateTableEventsInBatchEnabled = emitCreateTableEventsInBatchEnabled; } public String getHostname() { @@ -299,4 +302,8 @@ public boolean isSkipSnapshotBackfill() { public boolean isTreatTinyInt1AsBoolean() { return treatTinyInt1AsBoolean; } + + public boolean isEmitCreateTableEventsInBatchEnabled() { + return emitCreateTableEventsInBatchEnabled; + } } diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java index 569b62232db..35446c6eb19 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java @@ -78,6 +78,7 @@ public class MySqlSourceConfigFactory implements Serializable { private boolean treatTinyInt1AsBoolean = true; private boolean useLegacyJsonFormat = true; private boolean assignUnboundedChunkFirst = false; + private boolean emitCreateTableEventsInBatchEnabled = false; public MySqlSourceConfigFactory hostname(String hostname) { this.hostname = hostname; @@ -341,6 +342,16 @@ public MySqlSourceConfigFactory assignUnboundedChunkFirst(boolean assignUnbounde return this; } + /** + * Whether to emit all CreateTableEvents in batch during initialization phase. Defaults to + * false. + */ + public MySqlSourceConfigFactory emitCreateTableEventsInBatchEnabled( + boolean emitCreateTableEventsInBatchEnabled) { + this.emitCreateTableEventsInBatchEnabled = emitCreateTableEventsInBatchEnabled; + return this; + } + /** Creates a new {@link MySqlSourceConfig} for the given subtask {@code subtaskId}. */ public MySqlSourceConfig createConfig(int subtaskId) { // hard code server name, because we don't need to distinguish it, docs: @@ -444,6 +455,7 @@ public MySqlSourceConfig createConfig(int subtaskId, String serverName) { parseOnLineSchemaChanges, treatTinyInt1AsBoolean, useLegacyJsonFormat, - assignUnboundedChunkFirst); + assignUnboundedChunkFirst, + emitCreateTableEventsInBatchEnabled); } } From ae57eef67ef319f446292b1ffd774b50c4131e21 Mon Sep 17 00:00:00 2001 From: Eric Zeng Date: Thu, 9 Jul 2026 08:51:05 +0800 Subject: [PATCH 2/2] docs(mysql): Add scan.emit.create-table-events.in-batch.enabled option --- .../docs/connectors/pipeline-connectors/mysql.md | 11 +++++++++++ .../docs/connectors/pipeline-connectors/mysql.md | 11 +++++++++++ 2 files changed, 22 insertions(+) diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md b/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md index 04ec2844551..842b903809d 100644 --- a/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md +++ b/docs/content.zh/docs/connectors/pipeline-connectors/mysql.md @@ -330,6 +330,17 @@ pipeline: 这是一项实验特性，默认为 false。 + + scan.emit.create-table-events.in-batch.enabled + optional + false + Boolean + + 是否在快照初始化阶段批量发送所有 CreateTableEvent。
+ 开启后，所有表的 Schema 会在处理数据记录前一次性获取并发送，降低大量表场景下的 Schema 协调开销。
+ 默认为 false。 + + scan.incremental.snapshot.backfill.skip optional diff --git a/docs/content/docs/connectors/pipeline-connectors/mysql.md b/docs/content/docs/connectors/pipeline-connectors/mysql.md index e6e61e20720..fb2fbf4fd30 100644 --- a/docs/content/docs/connectors/pipeline-connectors/mysql.md +++ b/docs/content/docs/connectors/pipeline-connectors/mysql.md @@ -350,6 +350,17 @@ pipeline: Experimental option, defaults to false. + + scan.emit.create-table-events.in-batch.enabled + optional + false + Boolean + + Whether to emit all CreateTableEvents in batch during snapshot initialization.
+ When enabled, all table schemas are fetched and emitted together before processing data records, reducing schema coordination overhead for large table counts.
+ Defaults to false. + + scan.incremental.snapshot.backfill.skip optional