You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2025/09/18 05:31:43 UTC
(flink-cdc) branch master updated: [FLINK-36178][pipeline-connector/mysql] Parse `CREATE TABLE ... LIKE …` syntax. (#3590)
This is an automated email from the ASF dual-hosted git repository.
kunni pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 6b6eb8022 [FLINK-36178][pipeline-connector/mysql] Parse `CREATE TABLE ... LIKE …` syntax. (#3590)
6b6eb8022 is described below
commit 6b6eb80229b25f50db19995289eace2d786d4818
Author: Kunni <de...@gmail.com>
AuthorDate: Thu Sep 18 13:31:39 2025 +0800
[FLINK-36178][pipeline-connector/mysql] Parse `CREATE TABLE ... LIKE …` syntax. (#3590)
---
.../parser/CustomAlterTableParserListener.java | 30 ++++++
.../source/reader/MySqlPipelineRecordEmitter.java | 17 +++-
.../mysql/source/MySqlPipelineITCase.java | 112 +++++++++++++++++++++
3 files changed, 158 insertions(+), 1 deletion(-)
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/parser/CustomAlterTableParserListener.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomAlterTableParserListener.java
index e1b759932..82f5be7d9 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomAlterTableParserListener.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomAlterTableParserListener.java
@@ -34,6 +34,7 @@ import io.debezium.ddl.parser.mysql.generated.MySqlParser;
import io.debezium.ddl.parser.mysql.generated.MySqlParserBaseListener;
import io.debezium.relational.Column;
import io.debezium.relational.ColumnEditor;
+import io.debezium.relational.Table;
import io.debezium.relational.TableEditor;
import io.debezium.relational.TableId;
import org.antlr.v4.runtime.tree.ParseTreeListener;
@@ -82,6 +83,35 @@ public class CustomAlterTableParserListener extends MySqlParserBaseListener {
this.isTableIdCaseInsensitive = isTableIdCaseInsensitive;
}
+ @Override
+ public void exitCopyCreateTable(MySqlParser.CopyCreateTableContext ctx) {
+ TableId tableId = parser.parseQualifiedTableId(ctx.tableName(0).fullId());
+ TableId originalTableId = parser.parseQualifiedTableId(ctx.tableName(1).fullId());
+ Table original = parser.databaseTables().forTable(originalTableId);
+ if (original != null) {
+ parser.databaseTables()
+ .overwriteTable(
+ tableId,
+ original.columns(),
+ original.primaryKeyColumnNames(),
+ original.defaultCharsetName());
+ parser.signalCreateTable(tableId, ctx);
+ Schema.Builder builder = Schema.newBuilder();
+ original.columns().forEach(column -> builder.column(toCdcColumn(column)));
+ if (!original.primaryKeyColumnNames().isEmpty()) {
+ builder.primaryKey(original.primaryKeyColumnNames());
+ }
+ changes.add(new CreateTableEvent(toCdcTableId(tableId), builder.build()));
+ } else {
+ LOG.warn(
+ "Could not find schema of {} while parsing sql of `CREATE TABLE {} LIKE {}`",
+ originalTableId,
+ tableId,
+ originalTableId);
+ }
+ super.exitCopyCreateTable(ctx);
+ }
+
@Override
public void enterColumnCreateTable(MySqlParser.ColumnCreateTableContext ctx) {
TableId tableId = parser.parseQualifiedTableId(ctx.tableName().fullId());
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 3fa9cbb78..c946b9e29 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
@@ -141,7 +141,8 @@ public class MySqlPipelineRecordEmitter extends MySqlRecordEmitter<Event> {
}
}
} else {
- if (isDataChangeRecord(element) || isSchemaChangeEvent(element)) {
+ boolean isDataChangeRecord = isDataChangeRecord(element);
+ if (isDataChangeRecord || isSchemaChangeEvent(element)) {
TableId tableId = getTableId(element);
if (!alreadySendCreateTableTables.contains(tableId)) {
CreateTableEvent createTableEvent = createTableEventCache.get(tableId);
@@ -151,6 +152,20 @@ public class MySqlPipelineRecordEmitter extends MySqlRecordEmitter<Event> {
}
alreadySendCreateTableTables.add(tableId);
}
+ // In rare case, we may miss some CreateTableEvents before DataChangeEvents.
+ // Don't send CreateTableEvent for SchemaChangeEvents as it's the latest schema.
+ if (isDataChangeRecord && !createTableEventCache.containsKey(tableId)) {
+ try (JdbcConnection jdbc = openJdbcConnection(sourceConfig)) {
+ Schema schema = getSchema(jdbc, tableId);
+ CreateTableEvent createTableEvent =
+ new CreateTableEvent(
+ org.apache.flink.cdc.common.event.TableId.tableId(
+ tableId.catalog(), tableId.table()),
+ schema);
+ output.collect(createTableEvent);
+ createTableEventCache.put(tableId, createTableEvent);
+ }
+ }
}
}
super.processElement(element, output, splitState);
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/MySqlPipelineITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
index 7504e80fd..2c45c14f4 100644
--- a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
+++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
@@ -20,6 +20,7 @@ package org.apache.flink.cdc.connectors.mysql.source;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.data.DecimalData;
import org.apache.flink.cdc.common.data.binary.BinaryStringData;
import org.apache.flink.cdc.common.event.AddColumnEvent;
import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
@@ -64,6 +65,7 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.testcontainers.lifecycle.Startables;
+import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -1382,6 +1384,116 @@ class MySqlPipelineITCase extends MySqlSourceTestBase {
.physicalColumn("notes", DataTypes.STRING())
.primaryKey("id", "name")
.build()));
+
+ // Test create table DDL with like syntax
+ statement.execute(
+ String.format(
+ "CREATE TABLE `%s`.`newlyAddedTable4` LIKE `%s`.`newlyAddedTable3`",
+ inventoryDatabase.getDatabaseName(),
+ inventoryDatabase.getDatabaseName()));
+ expected.add(
+ new CreateTableEvent(
+ TableId.tableId(
+ inventoryDatabase.getDatabaseName(), "newlyAddedTable4"),
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.DECIMAL(20, 0).notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(17).notNull())
+ .physicalColumn("notes", DataTypes.STRING())
+ .primaryKey("id", "name")
+ .build()));
+
+ // Test create table DDL with as syntax, Primary key information will not be retained.
+ statement.execute(
+ String.format(
+ "CREATE TABLE `%s`.`newlyAddedTable5` AS SELECT * FROM `%s`.`newlyAddedTable3`",
+ inventoryDatabase.getDatabaseName(),
+ inventoryDatabase.getDatabaseName()));
+ expected.add(
+ new CreateTableEvent(
+ TableId.tableId(
+ inventoryDatabase.getDatabaseName(), "newlyAddedTable5"),
+ Schema.newBuilder()
+ .physicalColumn(
+ "id", DataTypes.DECIMAL(20, 0).notNull(), null, "0")
+ .physicalColumn("name", DataTypes.VARCHAR(17).notNull())
+ .physicalColumn("notes", DataTypes.STRING())
+ .build()));
+
+ // Database and table that does not match the filter of regular expression.
+ statement.execute(
+ String.format(
+ "CREATE DATABASE `%s_copy`", inventoryDatabase.getDatabaseName()));
+ statement.execute(
+ String.format(
+ "CREATE TABLE `%s_copy`.`newlyAddedTable`("
+ + "id SERIAL,"
+ + "name VARCHAR(17),"
+ + "notes TEXT,"
+ + "PRIMARY KEY (id));",
+ inventoryDatabase.getDatabaseName()));
+
+ // This should be ignored as another_database is not included in the captured regular
+ // expression.
+ statement.execute(
+ String.format(
+ "CREATE TABLE `%s`.`newlyAddedTable6` LIKE `%s_copy`.`newlyAddedTable`",
+ inventoryDatabase.getDatabaseName(),
+ inventoryDatabase.getDatabaseName()));
+
+ // This should not be ignored as MySQL will build and emit a new sql like:
+ // CREATE TABLE `newlyAddedTable7` (
+ // `id` bigint unsigned NOT NULL DEFAULT '0',
+ // `name` varchar(17) DEFAULT NULL,
+ // `notes` text
+ // ) START TRANSACTION.
+ statement.execute(
+ String.format(
+ "CREATE TABLE `%s`.`newlyAddedTable7` AS SELECT * FROM `%s_copy`.`newlyAddedTable`",
+ inventoryDatabase.getDatabaseName(),
+ inventoryDatabase.getDatabaseName()));
+ // Primary key information will not be retained.
+ expected.add(
+ new CreateTableEvent(
+ TableId.tableId(
+ inventoryDatabase.getDatabaseName(), "newlyAddedTable7"),
+ Schema.newBuilder()
+ .physicalColumn(
+ "id", DataTypes.DECIMAL(20, 0).notNull(), null, "0")
+ .physicalColumn("name", DataTypes.VARCHAR(17))
+ .physicalColumn("notes", DataTypes.STRING())
+ .build()));
+
+ // The CreateTableEvent is not correctly emitted.
+ statement.execute(
+ String.format(
+ "INSERT `%s`.`newlyAddedTable6` VALUES(1, 'Mark', 'eu')",
+ inventoryDatabase.getDatabaseName()));
+ expected.add(
+ new CreateTableEvent(
+ TableId.tableId(
+ inventoryDatabase.getDatabaseName(), "newlyAddedTable6"),
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.DECIMAL(20, 0).notNull(), null)
+ .physicalColumn("name", DataTypes.VARCHAR(17))
+ .physicalColumn("notes", DataTypes.STRING())
+ .primaryKey("id")
+ .build()));
+ expected.add(
+ DataChangeEvent.insertEvent(
+ TableId.tableId(
+ inventoryDatabase.getDatabaseName(), "newlyAddedTable6"),
+ new BinaryRecordDataGenerator(
+ new DataType[] {
+ DataTypes.DECIMAL(20, 0),
+ DataTypes.VARCHAR(17),
+ DataTypes.STRING()
+ })
+ .generate(
+ new Object[] {
+ DecimalData.fromBigDecimal(BigDecimal.ONE, 20, 0),
+ new BinaryStringData("Mark"),
+ new BinaryStringData("eu")
+ })));
}
List<Event> actual = fetchResults(events, expected.size());
assertEqualsInAnyOrder(