From e4ff7b82d1a609094f6a6c02194222075eaab13d Mon Sep 17 00:00:00 2001 From: lelovelo <41282750+lelovelo@users.noreply.github.com> Date: Fri, 7 Aug 2026 16:26:40 +0800 Subject: [PATCH] [mysql] Preserve schema for no-op CREATE TABLE LIKE --- .../CustomAlterTableParserListener.java | 12 +++ .../CustomCreateTableParserListener.java | 50 ++++++++++ .../CustomMySqlAntlrDdlParserListener.java | 3 +- .../parser/CustomMySqlAntlrDdlParserTest.java | 95 +++++++++++++++++++ 4 files changed, 158 insertions(+), 2 deletions(-) create mode 100644 flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomCreateTableParserListener.java 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/parser/CustomMySqlAntlrDdlParserTest.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/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 db2c1c8840e..4448ed0a25e 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 @@ -89,6 +89,18 @@ public CustomAlterTableParserListener( public void exitCopyCreateTable(MySqlParser.CopyCreateTableContext ctx) { TableId tableId = parser.parseQualifiedTableId(ctx.tableName(0).fullId()); TableId originalTableId = parser.parseQualifiedTableId(ctx.tableName(1).fullId()); + + // MySQL logs CREATE TABLE IF NOT EXISTS even when the target already exists. In that case + // the statement is a no-op and must not replace the schema restored from history. + if (ctx.ifNotExists() != null && parser.databaseTables().forTable(tableId) != null) { + LOG.debug( + "Ignoring no-op CREATE TABLE IF NOT EXISTS {} LIKE {} because the target table already exists", + tableId, + originalTableId); + super.exitCopyCreateTable(ctx); + return; + } + Table original = parser.databaseTables().forTable(originalTableId); if (original != null) { parser.databaseTables() 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/CustomCreateTableParserListener.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/CustomCreateTableParserListener.java new file mode 100644 index 00000000000..702d7ca261e --- /dev/null +++ b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomCreateTableParserListener.java @@ -0,0 +1,50 @@ +/* + * 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.parser; + +import io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser; +import io.debezium.connector.mysql.antlr.listener.CreateTableParserListener; +import io.debezium.ddl.parser.mysql.generated.MySqlParser; +import org.antlr.v4.runtime.tree.ParseTreeListener; + +import java.util.List; + +/** + * Handles regular CREATE TABLE statements while leaving CREATE TABLE ... LIKE processing to {@link + * CustomAlterTableParserListener}. + * + *

The custom listener owns both the Debezium table cache update and the Flink CDC schema event. + * Keeping that operation in one listener lets it preserve MySQL's no-op semantics for {@code IF NOT + * EXISTS} without the Debezium listener overwriting the target schema first. + */ +final class CustomCreateTableParserListener extends CreateTableParserListener { + + CustomCreateTableParserListener(MySqlAntlrDdlParser parser, List listeners) { + super(parser, listeners); + } + + /** + * Copy-table statements are handled atomically by {@link CustomAlterTableParserListener}. + * Calling the parent implementation here would update the shared schema cache before the custom + * listener can determine whether MySQL treated the statement as a no-op. + */ + @Override + public void exitCopyCreateTable(MySqlParser.CopyCreateTableContext ctx) { + // Intentionally empty. + } +} 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/CustomMySqlAntlrDdlParserListener.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/CustomMySqlAntlrDdlParserListener.java index 68d5b3e0c14..b2d25a52344 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/CustomMySqlAntlrDdlParserListener.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/CustomMySqlAntlrDdlParserListener.java @@ -25,7 +25,6 @@ import io.debezium.connector.mysql.antlr.listener.AlterTableParserListener; import io.debezium.connector.mysql.antlr.listener.AlterViewParserListener; import io.debezium.connector.mysql.antlr.listener.CreateAndAlterDatabaseParserListener; -import io.debezium.connector.mysql.antlr.listener.CreateTableParserListener; import io.debezium.connector.mysql.antlr.listener.CreateUniqueIndexParserListener; import io.debezium.connector.mysql.antlr.listener.CreateViewParserListener; import io.debezium.connector.mysql.antlr.listener.DropDatabaseParserListener; @@ -82,7 +81,7 @@ public CustomMySqlAntlrDdlParserListener( // initialize listeners listeners.add(new CreateAndAlterDatabaseParserListener(parser)); listeners.add(new DropDatabaseParserListener(parser)); - listeners.add(new CreateTableParserListener(parser, listeners)); + listeners.add(new CustomCreateTableParserListener(parser, listeners)); listeners.add( new CustomAlterTableParserListener( parser, listeners, parsedEvents, tinyInt1isBit, isTableIdCaseInsensitive)); 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/parser/CustomMySqlAntlrDdlParserTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/parser/CustomMySqlAntlrDdlParserTest.java new file mode 100644 index 00000000000..c0e595acab3 --- /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/parser/CustomMySqlAntlrDdlParserTest.java @@ -0,0 +1,95 @@ +/* + * 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.parser; + +import org.apache.flink.cdc.common.event.CreateTableEvent; +import org.apache.flink.cdc.common.event.SchemaChangeEvent; + +import io.debezium.relational.Table; +import io.debezium.relational.TableId; +import io.debezium.relational.Tables; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests MySQL copy-table DDL handling in {@link CustomMySqlAntlrDdlParser}. */ +class CustomMySqlAntlrDdlParserTest { + + private static final TableId TARGET_TABLE = new TableId("inventory", null, "target_table"); + private static final TableId TEMPLATE_TABLE = new TableId("inventory", null, "template_table"); + + @Test + void shouldPreserveExistingSchemaForNoOpCopyCreateTable() { + Tables tables = new Tables(); + CustomMySqlAntlrDdlParser parser = createParser(); + parser.parse( + "CREATE TABLE inventory.target_table (" + + "id BIGINT NOT NULL, name VARCHAR(32), status INT, PRIMARY KEY (id));" + + "CREATE TABLE inventory.template_table (" + + "id BIGINT NOT NULL, revision BIGINT, name VARCHAR(64), status INT, " + + "PRIMARY KEY (id));", + tables); + parser.getAndClearParsedEvents(); + + parser.parse( + "CREATE TABLE IF NOT EXISTS inventory.target_table " + + "LIKE inventory.template_table;", + tables); + + Table target = tables.forTable(TARGET_TABLE); + assertThat(target).isNotNull(); + assertThat(target.retrieveColumnNames()).containsExactly("id", "name", "status"); + assertThat(target.columnWithName("name").length()).isEqualTo(32); + assertThat(target.primaryKeyColumnNames()).containsExactly("id"); + assertThat(parser.getAndClearParsedEvents()).isEmpty(); + } + + @Test + void shouldCopySchemaWhenTargetDoesNotExist() { + Tables tables = new Tables(); + CustomMySqlAntlrDdlParser parser = createParser(); + parser.parse( + "CREATE TABLE inventory.template_table (" + + "id BIGINT NOT NULL, revision BIGINT, name VARCHAR(64), status INT, " + + "PRIMARY KEY (id));", + tables); + parser.getAndClearParsedEvents(); + + parser.parse( + "CREATE TABLE IF NOT EXISTS inventory.target_table " + + "LIKE inventory.template_table;", + tables); + + Table target = tables.forTable(TARGET_TABLE); + Table template = tables.forTable(TEMPLATE_TABLE); + assertThat(target).isNotNull(); + assertThat(target.retrieveColumnNames()) + .containsExactlyElementsOf(template.retrieveColumnNames()); + assertThat(target.primaryKeyColumnNames()) + .containsExactlyElementsOf(template.primaryKeyColumnNames()); + List events = parser.getAndClearParsedEvents(); + assertThat(events).hasSize(1); + assertThat(events.get(0)).isInstanceOf(CreateTableEvent.class); + } + + private CustomMySqlAntlrDdlParser createParser() { + return new CustomMySqlAntlrDdlParser(false, false, false); + } +}