Skip to content

[FLINK-40102][mysql]Support batch emitting CreateTableEvents to reduce schema coordination overhead #4466

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
eric666666 wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[FLINK-40102][mysql]Support batch emitting CreateTableEvents to reduce schema coordination overhead #4466

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
befc04e
feat(mysql): Support batch emitting CreateTableEvents
eric666666 Apr 15, 2026
ae57eef
docs(mysql): Add scan.emit.create-table-events.in-batch.enabled option
eric666666 Jul 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions docs/content.zh/docs/connectors/pipeline-connectors/mysql.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -330,6 +330,17 @@ pipeline:
这是一项实验特性，默认为 false。
</td>
</tr>
<tr>
<td>scan.emit.create-table-events.in-batch.enabled</td>
<td>optional</td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
<td>
是否在快照初始化阶段批量发送所有 CreateTableEvent。<br>
开启后，所有表的 Schema 会在处理数据记录前一次性获取并发送，降低大量表场景下的 Schema 协调开销。<br>
默认为 false。
</td>
</tr>
<tr>
<td>scan.incremental.snapshot.backfill.skip</td>
<td>optional</td>
Expand Down
11 changes: 11 additions & 0 deletions docs/content/docs/connectors/pipeline-connectors/mysql.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -350,6 +350,17 @@ pipeline:
Experimental option, defaults to false.
</td>
</tr>
<tr>
<td>scan.emit.create-table-events.in-batch.enabled</td>
<td>optional</td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
<td>
Whether to emit all CreateTableEvents in batch during snapshot initialization.<br>
When enabled, all table schemas are fetched and emitted together before processing data records, reducing schema coordination overhead for large table counts.<br>
Defaults to false.
</td>
</tr>
<tr>
<td>scan.incremental.snapshot.backfill.skip</td>
<td>optional</td>
Expand Down
5 changes: 5 additions & 0 deletions ...l/src/main/java/org/apache/flink/cdc/connectors/mysql/factory/MySqlDataSourceFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -220,6 +223,7 @@ public DataSource createDataSource(Context context) {
.treatTinyInt1AsBoolean(treatTinyInt1AsBoolean)
.useLegacyJsonFormat(useLegacyJsonFormat)
.assignUnboundedChunkFirst(isAssignUnboundedChunkFirst)
.emitCreateTableEventsInBatchEnabled(emitCreateTableEventsInBatchEnabled)
.skipSnapshotBackfill(skipSnapshotBackfill);

List<TableId> tableIds = MySqlSchemaUtils.listTables(configFactory.createConfig(0), null);
Expand Down Expand Up @@ -357,6 +361,7 @@ public Set<ConfigOption<?>> 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;
}
Expand Down
11 changes: 11 additions & 0 deletions ...ql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlDataSourceOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Boolean> 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<Boolean> SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP =
ConfigOptions.key("scan.incremental.snapshot.backfill.skip")
Expand Down
20 changes: 16 additions & 4 deletions .../java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,8 @@ public class MySqlPipelineRecordEmitter extends MySqlRecordEmitter<Event> {
private boolean isBounded = false;
private final boolean isTableIdCaseInsensitive;

private final boolean emitCreateTableEventsInBatch;

private final DebeziumDeserializationSchema<Event> debeziumDeserializationSchema;

private final Map<TableId, CreateTableEvent> createTableEventCache;
Expand All @@ -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()) {
Expand All @@ -130,11 +140,13 @@ public void applySplit(MySqlSplit split) {
protected void processElement(
SourceRecord element, SourceOutput<Event> 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.
Comment on lines +143 to +145

@yuxiqian yuxiqian Jul 9, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIUC CreateTableEvents are still emitted one-by-one with this option enabled, and lots of schema coordination might block job from starting up.

@eric666666 eric666666 Jul 9, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @yuxiqian. The goal of this PR is specifically the INITIAL-mode case: without it, each snapshot split's low-watermark triggers a per-table CreateTableEvent interleaved with the data stream. Each one makes the downstream SchemaOperator emit a FlushEvent, which flushes whatever data is currently buffered — so with thousands of tables the data stream is interrupted thousands of times, and on Paimon/Iceberg that means thousands of tiny files + throughput loss.

With the option on, all CreateTableEvents are emitted together up front (first record, before any DataChangeEvent). The FlushEvent count is unchanged, but those flushes now hit an empty buffer (no data has flowed yet), so no real data files are produced. Per-table JDBC (SHOW CREATE TABLE) is also collapsed from N connections to one batched fetch.

The downstream per-event coordination you pointed at (one blocking RPC per CreateTableEvent) is a separate concern; happy to address it in a follow-up.

createTableEventCache.forEach(
(tableId, createTableEvent) -> {
output.collect(createTableEvent);
alreadySendCreateTableTables.add(tableId);
});
shouldEmitAllCreateTableEventsInSnapshotMode = false;
} else if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState()) {
Expand Down
Loading