mysql: handle batch CREATE TABLE retries per statement (#6746)

downstreamadapter/sink/mysql/sink_test.go

@@ -17,11 +17,13 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/DATA-DOG/go-sqlmock"
+	dmysql "github.com/go-sql-driver/mysql"
 	"github.com/golang/mock/gomock"
 	"github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
@@ -474,6 +476,73 @@ func TestMysqlSinkMeetsDDLError(t *testing.T) {
 	require.Equal(t, sink.IsNormal(), false)
 }
 
+func TestMysqlSinkBatchCreateTablesAcknowledgement(t *testing.T) {
+	for _, failLastCreate := range []bool{false, true} {
+		name := "all creates accepted"
+		if failLastCreate {
+			name = "last create fails"
+		}
+		t.Run(name, func(t *testing.T) {
+			_, sink, mock := getMysqlSink()
+			defer sink.Close()
+			sink.cfg.EnableDDLTs = false
+			queries := []string{
+				"CREATE TABLE `t1` (`id` INT);",
+				"CREATE TABLE `t2` (`id` INT);",
+			}
+			event := &commonEvent.DDLEvent{
+				Type:       byte(timodel.ActionCreateTables),
+				SchemaName: "upstream",
+				Query:      strings.Join(queries, ""),
+				StartTs:    1,
+				FinishedTs: 2,
+			}
+			for _, tableName := range []string{"t1", "t2"} {
+				event.MultipleTableInfos = append(event.MultipleTableInfos, &common.TableInfo{
+					TableName: common.TableName{
+						Schema: "upstream", Table: tableName,
+						TargetSchema: "test", TargetTable: tableName,
+					},
+				})
+			}
+			flushCount := 0
+			event.AddPostFlushFunc(func() {
+				require.NoError(t, mock.ExpectationsWereMet())
+				flushCount++
+			})
+
+			for i, query := range queries {
+				mock.ExpectQuery("BEGIN; SET @ticdc_ts := TIDB_PARSE_TSO(@@tidb_current_ts); ROLLBACK; SELECT @ticdc_ts; SET @ticdc_ts=NULL;").
+					WillReturnRows(sqlmock.NewRows([]string{"@ticdc_ts"}))
+				mock.ExpectBegin()
+				mock.ExpectExec("USE `test`;").WillReturnResult(sqlmock.NewResult(0, 0))
+				mock.ExpectExec("SET TIMESTAMP = DEFAULT").WillReturnResult(sqlmock.NewResult(0, 0))
+				switch {
+				case i == 0:
+					mock.ExpectExec(query).WillReturnError(&dmysql.MySQLError{Number: 1050, Message: "Table already exists"})
+					mock.ExpectRollback()
+				case failLastCreate:
+					mock.ExpectExec(query).WillReturnError(&dmysql.MySQLError{Number: 1045, Message: "Access denied"})
+					mock.ExpectRollback()
+				default:
+					mock.ExpectExec(query).WillReturnResult(sqlmock.NewResult(0, 0))
+					mock.ExpectCommit()
+				}
+			}
+
+			err := sink.WriteBlockEvent(event)
+			if failLastCreate {
+				require.Error(t, err)
+				require.Zero(t, flushCount)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, 1, flushCount)
+			}
+			require.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}
+
 func TestMysqlSinkFlushLargeBatchEvent(t *testing.T) {
 	sink, mock := MysqlSinkForTestWithMaxTxnRows(4)
 

pkg/sink/mysql/mysql_writer_ddl.go

@@ -214,6 +214,17 @@ func (w *Writer) execDDL(event *commonEvent.DDLEvent) error {
 // If the downstream is TiDB, it will query the DDL and wait until it finishes.
 // For 'add index' ddl, it will return immediately without waiting and will query it during the next DDL execution.
 func (w *Writer) execDDLWithMaxRetries(event *commonEvent.DDLEvent) error {
+	if event.GetDDLType() == timodel.ActionCreateTables {
+		// Retry each CREATE separately so an ignorable error cannot skip the
+		// remaining statements. DDL ts bookkeeping still uses the original event.
+		for _, ddl := range event.GetEvents() {
+			if err := w.execDDLWithMaxRetries(ddl); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+
 	ddlCreateTime := getDDLCreateTime(w.ctx, w.db)
 	return retry.Do(w.ctx, func() error {
 		err := w.statistics.RecordDDLExecution(func() (string, error) {

pkg/sink/mysql/mysql_writer_ddl_test.go

@@ -0,0 +1,200 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// Licensed 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 mysql
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/go-sql-driver/mysql"
+	"github.com/pingcap/ticdc/pkg/common"
+	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
+	"github.com/pingcap/tidb/pkg/meta/model"
+	"github.com/stretchr/testify/require"
+)
+
+func newBatchCreateTablesEvent() (*commonEvent.DDLEvent, []string) {
+	event := &commonEvent.DDLEvent{
+		Type:       byte(model.ActionCreateTables),
+		SchemaName: "test",
+		StartTs:    1,
+		FinishedTs: 2,
+		BlockedTables: &commonEvent.InfluencedTables{
+			InfluenceType: commonEvent.InfluenceTypeNormal,
+			TableIDs:      []int64{0},
+		},
+	}
+	queries := make([]string, 0, 3)
+	for i := 1; i <= 3; i++ {
+		tableName := fmt.Sprintf("t%d", i)
+		queries = append(queries, fmt.Sprintf("CREATE TABLE `%s` (`id` INT);", tableName))
+		event.MultipleTableInfos = append(event.MultipleTableInfos, &common.TableInfo{
+			TableName: common.TableName{Schema: "test", Table: tableName, TableID: int64(i)},
+		})
+		event.NeedAddedTables = append(event.NeedAddedTables, commonEvent.Table{SchemaID: 1, TableID: int64(i)})
+	}
+	event.Query = strings.Join(queries, "")
+	return event, queries
+}
+
+func TestFlushDDLEventCreateTables(t *testing.T) {
+	tableExists := &mysql.MySQLError{Number: 1050, Message: "Table already exists"}
+	accessDenied := &mysql.MySQLError{Number: 1045, Message: "Access denied"}
+	for _, tc := range []struct {
+		name          string
+		queryErrors   [][]error
+		expectedError error
+		replayBatch   bool
+	}{
+		{name: "successful batch", queryErrors: [][]error{{nil}, {nil}, {nil}}},
+		{name: "first table exists", queryErrors: [][]error{{tableExists}, {nil}, {nil}}},
+		{name: "middle table exists", queryErrors: [][]error{{nil}, {tableExists}, {nil}}},
+		{name: "retry after partial execution", queryErrors: [][]error{{nil}, {mysql.ErrInvalidConn, tableExists}, {nil}}},
+		{name: "later create fails", queryErrors: [][]error{{tableExists}, {accessDenied}}, expectedError: accessDenied},
+		{name: "replay whole batch after partial success", queryErrors: [][]error{{nil}, {accessDenied}}, expectedError: accessDenied, replayBatch: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			writer, db, mock := newTestMysqlWriter(t)
+			defer db.Close()
+			writer.ddlTsTableInit = true
+			event, queries := newBatchCreateTablesEvent()
+			original := *event
+
+			ddlTsQuery := "INSERT INTO tidb_cdc.ddl_ts_v1 (ticdc_cluster_id, changefeed, ddl_ts, table_id, finished, is_syncpoint) VALUES " +
+				"('default', 'test/test', '2', 0, %d, 0), ('default', 'test/test', '2', 1, %d, 0), " +
+				"('default', 'test/test', '2', 2, %d, 0), ('default', 'test/test', '2', 3, %d, 0) " +
+				"ON DUPLICATE KEY UPDATE finished=VALUES(finished), ddl_ts=VALUES(ddl_ts), is_syncpoint=VALUES(is_syncpoint);"
+			mock.ExpectBegin()
+			mock.ExpectExec(fmt.Sprintf(ddlTsQuery, 0, 0, 0, 0)).WillReturnResult(sqlmock.NewResult(1, 4))
+			mock.ExpectCommit()
+
+			for i, queryErrors := range tc.queryErrors {
+				mock.ExpectQuery("BEGIN; SET @ticdc_ts := TIDB_PARSE_TSO(@@tidb_current_ts); ROLLBACK; SELECT @ticdc_ts; SET @ticdc_ts=NULL;").
+					WillReturnRows(sqlmock.NewRows([]string{"@ticdc_ts"}))
+				for _, queryError := range queryErrors {
+					mock.ExpectBegin()
+					mock.ExpectExec("USE `test`;").WillReturnResult(sqlmock.NewResult(0, 0))
+					mock.ExpectExec("SET TIMESTAMP = DEFAULT").WillReturnResult(sqlmock.NewResult(0, 0))
+					queryExpectation := mock.ExpectExec(queries[i])
+					if queryError != nil {
+						queryExpectation.WillReturnError(queryError)
+						mock.ExpectRollback()
+					} else {
+						queryExpectation.WillReturnResult(sqlmock.NewResult(0, 0))
+						mock.ExpectCommit()
+					}
+				}
+			}
+			if tc.expectedError == nil {
+				// The original event's progress is completed only after every CREATE.
+				mock.ExpectBegin()
+				mock.ExpectExec(fmt.Sprintf(ddlTsQuery, 1, 1, 1, 1)).WillReturnResult(sqlmock.NewResult(1, 4))
+				mock.ExpectCommit()
+			}
+
+			err := writer.FlushDDLEvent(event)
+			if tc.expectedError != nil {
+				require.ErrorIs(t, err, tc.expectedError)
+			} else {
+				require.NoError(t, err)
+			}
+			require.Equal(t, original, *event)
+			require.NoError(t, mock.ExpectationsWereMet())
+			if !tc.replayBatch {
+				return
+			}
+
+			// Replaying the original batch must accept the table created by the
+			// first attempt and complete the remaining CREATEs before finishing.
+			mock.ExpectBegin()
+			mock.ExpectExec(fmt.Sprintf(ddlTsQuery, 0, 0, 0, 0)).WillReturnResult(sqlmock.NewResult(1, 4))
+			mock.ExpectCommit()
+			for i, query := range queries {
+				mock.ExpectQuery("BEGIN; SET @ticdc_ts := TIDB_PARSE_TSO(@@tidb_current_ts); ROLLBACK; SELECT @ticdc_ts; SET @ticdc_ts=NULL;").
+					WillReturnRows(sqlmock.NewRows([]string{"@ticdc_ts"}))
+				mock.ExpectBegin()
+				mock.ExpectExec("USE `test`;").WillReturnResult(sqlmock.NewResult(0, 0))
+				mock.ExpectExec("SET TIMESTAMP = DEFAULT").WillReturnResult(sqlmock.NewResult(0, 0))
+				if i == 0 {
+					mock.ExpectExec(query).WillReturnError(tableExists)
+					mock.ExpectRollback()
+				} else {
+					mock.ExpectExec(query).WillReturnResult(sqlmock.NewResult(0, 0))
+					mock.ExpectCommit()
+				}
+			}
+			mock.ExpectBegin()
+			mock.ExpectExec(fmt.Sprintf(ddlTsQuery, 1, 1, 1, 1)).WillReturnResult(sqlmock.NewResult(1, 4))
+			mock.ExpectCommit()
+
+			require.NoError(t, writer.FlushDDLEvent(event))
+			require.Equal(t, original, *event)
+			require.NoError(t, mock.ExpectationsWereMet())
+		})
+	}
+}
+
+func TestFlushDDLEventCreateTablesWaitsForCurrentStatement(t *testing.T) {
+	writer, db, mock := newTestMysqlWriterForTiDB(t)
+	defer db.Close()
+	writer.cfg.EnableDDLTs = false
+	writer.ddlPollInterval = time.Millisecond
+	event, queries := newBatchCreateTablesEvent()
+	createTime := "2026-10-27 16:47:22.856000"
+
+	for i, query := range queries {
+		mock.ExpectQuery("BEGIN; SET @ticdc_ts := TIDB_PARSE_TSO(@@tidb_current_ts); ROLLBACK; SELECT @ticdc_ts; SET @ticdc_ts=NULL;").
+			WillReturnRows(sqlmock.NewRows([]string{"@ticdc_ts"}).AddRow(createTime))
+		mock.ExpectBegin()
+		mock.ExpectExec("USE `test`;").WillReturnResult(sqlmock.NewResult(0, 0))
+		mock.ExpectExec("SET TIMESTAMP = DEFAULT").WillReturnResult(sqlmock.NewResult(0, 0))
+		if i == 1 {
+			mock.ExpectExec(query).WillReturnError(mysql.ErrInvalidConn)
+			mock.ExpectRollback()
+			mock.ExpectQuery(fmt.Sprintf(checkRunningSQL, createTime, query)).
+				WillReturnRows(sqlmock.NewRows([]string{"QUERY", "STATE", "JOB_ID", "JOB_TYPE", "SCHEMA_STATE"}).
+					AddRow(query, "synced", "1", "create table", "public"))
+		} else {
+			mock.ExpectExec(query).WillReturnResult(sqlmock.NewResult(0, 0))
+			mock.ExpectCommit()
+		}
+	}
+
+	require.NoError(t, writer.FlushDDLEvent(event))
+	require.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestFlushDDLEventPreservesRenameTables(t *testing.T) {
+	writer, db, mock := newTestMysqlWriter(t)
+	defer db.Close()
+	writer.cfg.EnableDDLTs = false
+	event := &commonEvent.DDLEvent{
+		Type:       byte(model.ActionRenameTables),
+		SchemaName: "test",
+		Query:      "RENAME TABLE `t1` TO `t2`, `t3` TO `t4`;",
+	}
+	mock.ExpectQuery("BEGIN; SET @ticdc_ts := TIDB_PARSE_TSO(@@tidb_current_ts); ROLLBACK; SELECT @ticdc_ts; SET @ticdc_ts=NULL;").
+		WillReturnRows(sqlmock.NewRows([]string{"@ticdc_ts"}))
+	mock.ExpectBegin()
+	mock.ExpectExec("USE `test`;").WillReturnResult(sqlmock.NewResult(0, 0))
+	expectDDLExec(mock, event, writer.cfg.Timezone)
+	mock.ExpectCommit()
+
+	require.NoError(t, writer.FlushDDLEvent(event))
+	require.NoError(t, mock.ExpectationsWereMet())
+}

tests/integration_tests/batch_add_table/run.sh

@@ -27,7 +27,15 @@ function run_with_fast_create_table() {
 		;;
 	*) SINK_URI="mysql://normal:123456@127.0.0.1:3306/" ;;
 	esac
-	cdc_cli_changefeed create --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml"
+	local start_ts
+	local retry_table
+	if [ "$SINK_TYPE" == "mysql" ]; then
+		# Generate the batch before starting replication so its first table can
+		# already exist downstream when the original batch is replayed.
+		start_ts=$(run_cdc_cli_tso_query "$UP_PD_HOST_1" "$UP_PD_PORT_1")
+	else
+		cdc_cli_changefeed create --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml"
+	fi
 	case $SINK_TYPE in
 	kafka) run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=open-protocol&partition-num=4&version=${KAFKA_VERSION}&max-message-bytes=10485760" "$CUR/conf/changefeed.toml" ;;
 	storage) run_storage_consumer $WORK_DIR $SINK_URI "$CUR/conf/changefeed.toml" "" ;;
@@ -43,10 +51,43 @@ function run_with_fast_create_table() {
 		check_table_exists test.t_$i ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	done
 
+	if [ "$SINK_TYPE" == "mysql" ]; then
+		# Require a real ActionCreateTables job, excluding the filtered table.
+		local batch_query
+		batch_query=$(mysql -h "$UP_TIDB_HOST" -P "$UP_TIDB_PORT" -u root -N -B --raw -e \
+			"SELECT QUERY FROM information_schema.ddl_jobs
+			 WHERE DB_NAME = 'test' AND JOB_TYPE = 'create tables'
+			 AND QUERY NOT REGEXP 't_1([^0-9]|$)'
+			 ORDER BY JOB_ID LIMIT 1;")
+		if [ -z "$batch_query" ]; then
+			echo "expected a batch CREATE TABLE job without the filtered table"
+			exit 1
+		fi
+		printf '%s\n' "$batch_query" >"$WORK_DIR/batch_create_retry.log"
+		local first_query=${batch_query%%;*}
+		local remaining_query=${batch_query#*;}
+		if [[ "$remaining_query" =~ (t_[0-9]+) ]]; then
+			retry_table=${BASH_REMATCH[1]}
+		else
+			echo "expected another CREATE TABLE after the first statement"
+			exit 1
+		fi
+		run_sql "USE test; $first_query;" "$DOWN_TIDB_HOST" "$DOWN_TIDB_PORT"
+		cdc_cli_changefeed create --start-ts="$start_ts" --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml"
+	fi
+
 	for ((i = 1; i <= 100; i++)); do
 		mysql -h ${UP_TIDB_HOST} -P ${UP_TIDB_PORT} -u root -D "test" -e "insert into t_$i values(1,2)"
 	done
 
+	if [ "$SINK_TYPE" == "mysql" ]; then
+		# Error 1050 for the first CREATE must not skip this later CREATE or its DML.
+		check_table_exists "test.$retry_table" "$DOWN_TIDB_HOST" "$DOWN_TIDB_PORT" 30
+		ensure 10 "run_sql 'SELECT COUNT(*) AS replicated_rows FROM test.$retry_table WHERE a = 1 AND b = 2;' \
+			$DOWN_TIDB_HOST $DOWN_TIDB_PORT && check_contains 'replicated_rows: 1'"
+		check_logs_contains "$WORK_DIR" "Execute DDL failed, but error can be ignored"
+	fi
+
 	check_table_exists test.t_100 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 
 	run_sql_file $CUR/data/prepare.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}