From a43dcb952b09c61ced420c7c62db62c48fff2d26 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Fri, 30 Oct 2026 04:22:18 +0000 Subject: [PATCH 1/3] init Signed-off-by: wk989898 --- downstreamadapter/sink/mysql/sink_test.go | 69 ++++++++ pkg/sink/mysql/mysql_writer_ddl.go | 11 ++ pkg/sink/mysql/mysql_writer_ddl_test.go | 200 ++++++++++++++++++++++ 3 files changed, 280 insertions(+) create mode 100644 pkg/sink/mysql/mysql_writer_ddl_test.go diff --git a/downstreamadapter/sink/mysql/sink_test.go b/downstreamadapter/sink/mysql/sink_test.go index 796242ebaf..db34ecbc08 100644 --- a/downstreamadapter/sink/mysql/sink_test.go +++ b/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) diff --git a/pkg/sink/mysql/mysql_writer_ddl.go b/pkg/sink/mysql/mysql_writer_ddl.go index 2bdb608770..d5c3721b19 100644 --- a/pkg/sink/mysql/mysql_writer_ddl.go +++ b/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) { diff --git a/pkg/sink/mysql/mysql_writer_ddl_test.go b/pkg/sink/mysql/mysql_writer_ddl_test.go new file mode 100644 index 0000000000..850dde0a0c --- /dev/null +++ b/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(checkRunningSQL).WithArgs(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()) +} From 863ea4b250a81e2caf864eef126e263b2da1c2bd Mon Sep 17 00:00:00 2001 From: wk989898 Date: Fri, 30 Oct 2026 06:27:05 +0000 Subject: [PATCH 2/3] add test Signed-off-by: wk989898 --- .../integration_tests/batch_add_table/run.sh | 43 ++++++++++++++++++- 1 file changed, 42 insertions(+), 1 deletion(-) diff --git a/tests/integration_tests/batch_add_table/run.sh b/tests/integration_tests/batch_add_table/run.sh index e50b367c97..e3742fd750 100755 --- a/tests/integration_tests/batch_add_table/run.sh +++ b/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} From c3108e4f9557ccbca85a1bdabdacd07d0499b7db Mon Sep 17 00:00:00 2001 From: wk989898 Date: Thu, 5 Nov 2026 08:23:58 +0000 Subject: [PATCH 3/3] mysql: adapt batch create DDL polling test to release branch --- pkg/sink/mysql/mysql_writer_ddl_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/sink/mysql/mysql_writer_ddl_test.go b/pkg/sink/mysql/mysql_writer_ddl_test.go index 850dde0a0c..3b2df5f727 100644 --- a/pkg/sink/mysql/mysql_writer_ddl_test.go +++ b/pkg/sink/mysql/mysql_writer_ddl_test.go @@ -166,7 +166,7 @@ func TestFlushDDLEventCreateTablesWaitsForCurrentStatement(t *testing.T) { if i == 1 { mock.ExpectExec(query).WillReturnError(mysql.ErrInvalidConn) mock.ExpectRollback() - mock.ExpectQuery(checkRunningSQL).WithArgs(createTime, query). + 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 {