[mysql] Preserve schema for no-op CREATE TABLE IF NOT EXISTS ... LIKE

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()

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}.
+ *
+ * <p>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<ParseTreeListener> 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.
+    }
+}

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));

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<SchemaChangeEvent> events = parser.getAndClearParsedEvents();
+        assertThat(events).hasSize(1);
+        assertThat(events.get(0)).isInstanceOf(CreateTableEvent.class);
+    }
+
+    private CustomMySqlAntlrDdlParser createParser() {
+        return new CustomMySqlAntlrDdlParser(false, false, false);
+    }
+}