ddl: fix incorrect query in async ddl check (#3331)

close #3329

Author: wk989898

pkg/sink/mysql/config.go

@@ -32,6 +32,7 @@ import (
 	cerror "github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/security"
 	"github.com/pingcap/ticdc/pkg/util"
+	"github.com/pingcap/tidb/br/pkg/version"
 	"github.com/pingcap/tidb/pkg/sessionctx/vardef"
 	"go.uber.org/zap"
 )
@@ -138,6 +139,9 @@ type Config struct {
 	DryRunBlockInterval time.Duration
 	// SlowQuery is the threshold time above which the query will be logged.
 	SlowQuery time.Duration
+
+	// ServerInfo is the version info of the downstream
+	ServerInfo version.ServerInfo
 }
 
 // New returns the default mysql backend config.
@@ -294,7 +298,8 @@ func NewMysqlConfigAndDB(
 		return nil, nil, err
 	}
 
-	cfg.HasVectorType = ShouldFormatVectorType(db, cfg)
+	cfg.ServerInfo = getTiDBVersion(db)
+	cfg.HasVectorType = shouldFormatVectorType(cfg)
 
 	// By default, cache-prep-stmts=true, an LRU cache is used for prepared statements,
 	// two connections are required to process a transaction.

pkg/sink/mysql/helper.go

@@ -34,19 +34,29 @@ import (
 	timodel "github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/parser/charset"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
-	tmysql "github.com/pingcap/tidb/pkg/parser/mysql"
 	"go.uber.org/zap"
 )
 
 const checkRunningAddIndexSQL = `
-SELECT JOB_ID, JOB_TYPE, SCHEMA_STATE, STATE
+SELECT *
 FROM information_schema.ddl_jobs
-WHERE TABLE_ID = "%d"
+WHERE DB_NAME = "%s"
+    AND TABLE_NAME = "%s"
+    AND JOB_TYPE LIKE "add index%%"
+    AND (STATE = "running" OR STATE = "queueing")
+LIMIT 1;
+`
+
+// Ref: https://github.com/pingcap/tidb/issues/55725
+const checkRunningAddIndexSQLForOldVersion = `
+ADMIN SHOW DDL JOBS 1
+WHERE DB_NAME = "%s" 
+    AND TABLE_NAME = "%s"
     AND JOB_TYPE LIKE "add index%%"
     AND (STATE = "running" OR STATE = "queueing");
 `
 
-const checkRunningSQL = `SELECT JOB_ID, JOB_TYPE, SCHEMA_STATE, SCHEMA_ID, TABLE_ID, STATE, QUERY FROM information_schema.ddl_jobs 
+const checkRunningSQL = `SELECT * FROM information_schema.ddl_jobs 
 	WHERE CREATE_TIME >= "%s" AND QUERY = "%s";`
 
 // CheckIfBDRModeIsSupported checks if the downstream supports set tidb_cdc_write_source variable
@@ -396,28 +406,42 @@ func needWaitAsyncExecDone(t timodel.ActionType) bool {
 	}
 }
 
-// ShouldFormatVectorType return true if vector type should be converted to longtext.
-func ShouldFormatVectorType(db *sql.DB, cfg *Config) bool {
+func getTiDBVersion(db *sql.DB) version.ServerInfo {
+	versionInfo, err := export.SelectVersion(db)
+	if err != nil {
+		log.Warn("fail to get version", zap.Error(err))
+		return version.ParseServerInfo("")
+	}
+	return version.ParseServerInfo(versionInfo)
+}
+
+// shouldFormatVectorType return true if vector type should be converted to longtext.
+func shouldFormatVectorType(cfg *Config) bool {
 	if !cfg.HasVectorType {
 		log.Warn("please set `has-vector-type` to be true if a column is vector type when the downstream is not TiDB or TiDB version less than specify version",
 			zap.Any("hasVectorType", cfg.HasVectorType), zap.Any("supportVectorVersion", defaultSupportVectorVersion))
 		return false
 	}
-	versionInfo, err := export.SelectVersion(db)
-	if err != nil {
-		log.Warn("fail to get version", zap.Error(err), zap.Bool("isTiDB", cfg.IsTiDB))
-		return false
-	}
-	serverInfo := version.ParseServerInfo(versionInfo)
 	ver := semver.New(defaultSupportVectorVersion)
-	if !cfg.IsTiDB || serverInfo.ServerVersion.LessThan(*ver) {
+	if !cfg.IsTiDB || cfg.ServerInfo.ServerVersion.LessThan(*ver) {
 		log.Error("downstream unsupport vector type. it will be converted to longtext",
-			zap.String("version", serverInfo.ServerVersion.String()), zap.String("supportVectorVersion", defaultSupportVectorVersion), zap.Bool("isTiDB", cfg.IsTiDB))
+			zap.String("version", cfg.ServerInfo.ServerVersion.String()), zap.String("supportVectorVersion", defaultSupportVectorVersion), zap.Bool("isTiDB", cfg.IsTiDB))
 		return true
 	}
 	return false
 }
 
+// getCheckRunningAddIndexSQL return different sql according to tidb version
+func getCheckRunningAddIndexSQL(cfg *Config) string {
+	ver := semver.New(defaultRunningAddIndexNewSQLVersion)
+	if cfg.ServerInfo.ServerVersion.LessThan(*ver) {
+		log.Info("it will check running AddIndex SQL with old version",
+			zap.String("version", cfg.ServerInfo.ServerVersion.String()))
+		return checkRunningAddIndexSQLForOldVersion
+	}
+	return checkRunningAddIndexSQL
+}
+
 func isRetryableDMLError(err error) bool {
 	if !cerror.IsRetryableError(err) {
 		return false
@@ -538,15 +562,15 @@ func AdjustSQLModeCompatible(sqlModes string) (string, error) {
 	disable := strings.Join(needDisable, ",")
 	enable := strings.Join(needEnable, ",")
 
-	mode, err := tmysql.GetSQLMode(sqlModes)
+	mode, err := mysql.GetSQLMode(sqlModes)
 	if err != nil {
 		return sqlModes, err
 	}
-	disableMode, err2 := tmysql.GetSQLMode(disable)
+	disableMode, err2 := mysql.GetSQLMode(disable)
 	if err2 != nil {
 		return sqlModes, err2
 	}
-	enableMode, err3 := tmysql.GetSQLMode(enable)
+	enableMode, err3 := mysql.GetSQLMode(enable)
 	if err3 != nil {
 		return sqlModes, err3
 	}
@@ -558,9 +582,9 @@ func AdjustSQLModeCompatible(sqlModes string) (string, error) {
 }
 
 // GetSQLModeStrBySQLMode get string represent of sql_mode by sql_mode.
-func GetSQLModeStrBySQLMode(sqlMode tmysql.SQLMode) string {
+func GetSQLModeStrBySQLMode(sqlMode mysql.SQLMode) string {
 	var sqlModeStr []string
-	for str, SQLMode := range tmysql.Str2SQLMode {
+	for str, SQLMode := range mysql.Str2SQLMode {
 		if sqlMode&SQLMode != 0 {
 			sqlModeStr = append(sqlModeStr, str)
 		}

pkg/sink/mysql/mysql_writer.go

@@ -38,6 +38,8 @@ const (
 
 	defaultSupportVectorVersion = "8.4.0"
 
+	defaultRunningAddIndexNewSQLVersion = "8.5.0"
+
 	defaultErrorCausedSafeModeDuration = 5 * time.Second
 )
 
@@ -108,10 +110,10 @@ func (w *Writer) SetTableSchemaStore(tableSchemaStore *commonEvent.TableSchemaSt
 }
 
 func (w *Writer) FlushDDLEvent(event *commonEvent.DDLEvent) error {
-	// if w.cfg.IsTiDB {
-	// 	// first we check whether there is some async ddl executed now.
-	// 	w.waitAsyncDDLDone(event)
-	// }
+	if w.cfg.IsTiDB {
+		// first we check whether there is some async ddl executed now.
+		w.waitAsyncDDLDone(event)
+	}
 	if w.cfg.IsTiDB || !event.TiDBOnly {
 		// we write ddl ts before ddl first, and update the ddl ts item after ddl executed,
 		// to ensure the atomic with ddl writing when server is restarted.

pkg/sink/mysql/mysql_writer_ddl.go

@@ -25,6 +25,7 @@ import (
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/retry"
+	"github.com/pingcap/tidb/dumpling/export"
 	timodel "github.com/pingcap/tidb/pkg/meta/model"
 	"go.uber.org/zap"
 )
@@ -189,23 +190,16 @@ func (w *Writer) waitAsyncDDLDone(event *commonEvent.DDLEvent) {
 		return
 	}
 
-	var relatedTableIDs []int64
 	switch event.GetBlockedTables().InfluenceType {
-	case commonEvent.InfluenceTypeNormal:
-		relatedTableIDs = event.GetBlockedTables().TableIDs
 	// db-class, all-class ddl with not affect by async ddl, just return
 	case commonEvent.InfluenceTypeDB, commonEvent.InfluenceTypeAll:
 		return
 	}
 
-	for _, tableID := range relatedTableIDs {
-		// tableID 0 means table trigger, which can't do async ddl
-		if tableID == 0 {
-			continue
-		}
+	for _, blockedTable := range event.GetBlockedTableNames() {
 		// query the downstream,
 		// if the ddl is still running, we should wait for it.
-		err := w.checkAndWaitAsyncDDLDoneDownstream(tableID)
+		err := w.checkAndWaitAsyncDDLDoneDownstream(blockedTable.SchemaName, blockedTable.TableName)
 		if err != nil {
 			log.Error("check previous asynchronous ddl failed",
 				zap.String("keyspace", w.ChangefeedID.Keyspace()),
@@ -216,51 +210,45 @@ func (w *Writer) waitAsyncDDLDone(event *commonEvent.DDLEvent) {
 }
 
 // true means the async ddl is still running, false means the async ddl is done.
-func (w *Writer) doQueryAsyncDDL(tableID int64, query string) (bool, error) {
+func (w *Writer) doQueryAsyncDDL(query string) (bool, error) {
 	start := time.Now()
 	rows, err := w.db.QueryContext(w.ctx, query)
 	log.Debug("query duration", zap.Any("duration", time.Since(start)), zap.Any("query", query))
 	if err != nil {
 		return false, errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("failed to query ddl jobs table; Query is %s", query)))
 	}
-
-	defer rows.Close()
-	var jobID int64
-	var jobType string
-	var schemaState string
-	var state string
-
-	noRows := true
-	for rows.Next() {
-		noRows = false
-		err := rows.Scan(&jobID, &jobType, &schemaState, &state)
-		if err != nil {
-			return false, errors.WrapError(errors.ErrMySQLTxnError, errors.WithMessage(err, fmt.Sprintf("failed to query ddl jobs table; Query is %s", query)))
-		}
-
-		log.Info("async ddl is still running",
+	rets, err := export.GetSpecifiedColumnValuesAndClose(rows, "JOB_ID", "JOB_TYPE", "SCHEMA_STATE", "STATE", "QUERY")
+	if err != nil {
+		log.Error("check previous asynchronous ddl failed",
 			zap.String("changefeed", w.ChangefeedID.String()),
-			zap.Duration("checkDuration", time.Since(start)),
-			zap.Any("tableID", tableID),
-			zap.Any("jobID", jobID),
-			zap.String("jobType", jobType),
-			zap.String("schemaState", schemaState),
-			zap.String("state", state))
-		break
+			zap.Error(err))
+		return false, errors.Trace(err)
 	}
 
-	if noRows {
+	if len(rets) == 0 {
 		return false, nil
 	}
+	ret := rets[0]
+	jobID, jobType, schemaState, state, runningDDL := ret[0], ret[1], ret[2], ret[3], ret[4]
+	log.Info("async ddl is still running",
+		zap.String("changefeed", w.ChangefeedID.String()),
+		zap.Duration("checkDuration", time.Since(start)),
+		zap.String("runningDDL", runningDDL),
+		zap.String("query", query),
+		zap.Any("jobID", jobID),
+		zap.String("jobType", jobType),
+		zap.String("schemaState", schemaState),
+		zap.String("state", state))
 
 	return true, nil
 }
 
 // query the ddl jobs to find the state of the async ddl
 // if the ddl is still running, we should wait for it.
-func (w *Writer) checkAndWaitAsyncDDLDoneDownstream(tableID int64) error {
-	query := fmt.Sprintf(checkRunningAddIndexSQL, tableID)
-	running, err := w.doQueryAsyncDDL(tableID, query)
+func (w *Writer) checkAndWaitAsyncDDLDoneDownstream(schemaName, tableName string) error {
+	checkSQL := getCheckRunningAddIndexSQL(w.cfg)
+	query := fmt.Sprintf(checkSQL, schemaName, tableName)
+	running, err := w.doQueryAsyncDDL(query)
 	if err != nil {
 		return err
 	}
@@ -276,7 +264,7 @@ func (w *Writer) checkAndWaitAsyncDDLDoneDownstream(tableID int64) error {
 		case <-w.ctx.Done():
 			return nil
 		case <-ticker.C:
-			running, err = w.doQueryAsyncDDL(tableID, query)
+			running, err = w.doQueryAsyncDDL(query)
 			if err != nil {
 				return err
 			}

pkg/sink/mysql/mysql_writer_test.go

@@ -21,16 +21,20 @@ import (
 	"time"
 
 	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/go-sql-driver/mysql"
 	lru "github.com/hashicorp/golang-lru"
 	"github.com/pingcap/errors"
+	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/heartbeatpb"
 	"github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/config/kerneltype"
 	cerror "github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/metrics"
+	"github.com/pingcap/tidb/br/pkg/version"
 	ticonfig "github.com/pingcap/tidb/pkg/config"
 	"github.com/pingcap/tidb/pkg/disttask/framework/handle"
+	timodel "github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/sessionctx/vardef"
 	"github.com/stretchr/testify/require"
 )
@@ -65,6 +69,8 @@ func newTestMysqlWriterForTiDB(t *testing.T) (*Writer, *sql.DB, sqlmock.Sqlmock)
 	cfg.SyncPointRetention = 100 * time.Second
 	cfg.IsTiDB = true
 	cfg.EnableDDLTs = defaultEnableDDLTs
+	cfg.ServerInfo = version.ParseServerInfo(defaultRunningAddIndexNewSQLVersion)
+
 	changefeedID := common.NewChangefeedID4Test("test", "test")
 	statistics := metrics.NewStatistics(changefeedID, "mysqlSink")
 	writer := NewWriter(ctx, 0, db, cfg, changefeedID, statistics)
@@ -439,7 +445,6 @@ func TestMysqlWriter_RemoveDDLTsTable(t *testing.T) {
 	require.NoError(t, err)
 }
 
-/*
 // Test the async ddl can be write successfully
 func TestMysqlWriter_AsyncDDL(t *testing.T) {
 	writer, db, mock := newTestMysqlWriterForTiDB(t)
@@ -502,7 +507,7 @@ func TestMysqlWriter_AsyncDDL(t *testing.T) {
 	err = mock.ExpectationsWereMet()
 	require.NoError(t, err)
 
-	mock.ExpectQuery(fmt.Sprintf(checkRunningAddIndexSQL, 1)).WillReturnError(sqlmock.ErrCancelled)
+	mock.ExpectQuery(fmt.Sprintf(checkRunningAddIndexSQL, "test", "t")).WillReturnError(sqlmock.ErrCancelled)
 	mock.ExpectBegin()
 	mock.ExpectExec("INSERT INTO tidb_cdc.ddl_ts_v1 (ticdc_cluster_id, changefeed, ddl_ts, table_id, finished, is_syncpoint) VALUES ('default', 'test/test', '1', 1, 0, 0) ON DUPLICATE KEY UPDATE finished=VALUES(finished), ddl_ts=VALUES(ddl_ts), is_syncpoint=VALUES(is_syncpoint);").WillReturnResult(sqlmock.NewResult(1, 1))
 	mock.ExpectCommit()
@@ -538,7 +543,7 @@ func TestMysqlWriter_AsyncDDL(t *testing.T) {
 	mock.ExpectCommit()
 
 	// for add column ddl for table t
-	mock.ExpectQuery(fmt.Sprintf(checkRunningAddIndexSQL, 1)).WillReturnError(sqlmock.ErrCancelled)
+	mock.ExpectQuery(fmt.Sprintf(checkRunningAddIndexSQL, "test", "t")).WillReturnError(sqlmock.ErrCancelled)
 	mock.ExpectBegin()
 	mock.ExpectExec("INSERT INTO tidb_cdc.ddl_ts_v1 (ticdc_cluster_id, changefeed, ddl_ts, table_id, finished, is_syncpoint) VALUES ('default', 'test/test', '3', 1, 0, 0) ON DUPLICATE KEY UPDATE finished=VALUES(finished), ddl_ts=VALUES(ddl_ts), is_syncpoint=VALUES(is_syncpoint);").WillReturnResult(sqlmock.NewResult(1, 1))
 	mock.ExpectCommit()
@@ -567,6 +572,9 @@ func TestMysqlWriter_AsyncDDL(t *testing.T) {
 				InfluenceType: commonEvent.InfluenceTypeNormal,
 				TableIDs:      []int64{1},
 			},
+			BlockedTableNames: []commonEvent.SchemaTableName{
+				{SchemaName: addIndexjob.SchemaName, TableName: addIndexjob.TableName},
+			},
 		}
 
 		err = writer.FlushDDLEvent(addIndexddlEvent)
@@ -598,6 +606,9 @@ func TestMysqlWriter_AsyncDDL(t *testing.T) {
 				InfluenceType: commonEvent.InfluenceTypeNormal,
 				TableIDs:      []int64{0},
 			},
+			BlockedTableNames: []commonEvent.SchemaTableName{
+				{SchemaName: job2.SchemaName, TableName: job2.TableName},
+			},
 			NeedAddedTables: []commonEvent.Table{{TableID: 2, SchemaID: 1}},
 		}
 
@@ -619,6 +630,9 @@ func TestMysqlWriter_AsyncDDL(t *testing.T) {
 				InfluenceType: commonEvent.InfluenceTypeNormal,
 				TableIDs:      []int64{1},
 			},
+			BlockedTableNames: []commonEvent.SchemaTableName{
+				{SchemaName: job.SchemaName, TableName: job.TableName},
+			},
 		}
 
 		err = writer.FlushDDLEvent(ddlEvent)
@@ -628,7 +642,6 @@ func TestMysqlWriter_AsyncDDL(t *testing.T) {
 	err = mock.ExpectationsWereMet()
 	require.NoError(t, err)
 }
-*/
 
 func TestCheckIsDuplicateEntryError(t *testing.T) {
 	writer, db, _ := newTestMysqlWriter(t)

tests/integration_tests/ddl_wait/run.sh

@@ -33,6 +33,10 @@ function run() {
 
 	run_sql "alter table test.t modify column col decimal(30,10);"
 	run_sql "alter table test.t add index (col);"
+	run_sql "alter table test.t add index (col);"
+	sleep 3
+	cleanup_process $CDC_BINARY
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
 	# make sure all tables are equal in upstream and downstream
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml 180