ticdc: Support Vector data type | tidb=master pd=master tikv=master

cdc/entry/codec.go

@@ -210,6 +210,9 @@ func unflatten(datum types.Datum, ft *types.FieldType, loc *time.Location) (type
 		byteSize := (ft.GetFlen() + 7) >> 3
 		datum.SetUint64(0)
 		datum.SetMysqlBit(types.NewBinaryLiteralFromUint(val, byteSize))
+	case mysql.TypeTiDBVectorFloat32:
+		datum.SetVectorFloat32(types.ZeroVectorFloat32)
+		return datum, nil
 	}
 	return datum, nil
 }

cdc/entry/mounter.go

@@ -888,6 +888,9 @@ func formatColVal(datum types.Datum, col *timodel.ColumnInfo) (
 		}
 		const sizeOfV = unsafe.Sizeof(v)
 		return v, int(sizeOfV), warn, nil
+	case mysql.TypeTiDBVectorFloat32:
+		b := datum.GetVectorFloat32()
+		return b, b.Len(), "", nil
 	default:
 		// NOTICE: GetValue() may return some types that go sql not support, which will cause sink DML fail
 		// Make specified convert upper if you need

cdc/sink/ddlsink/mysql/async_ddl.go

@@ -15,25 +15,29 @@ package mysql
 
 import (
 	"context"
-	"database/sql"
 	"fmt"
 	"time"
 
 	"github.com/pingcap/log"
+	"github.com/pingcap/tidb/dumpling/export"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tiflow/cdc/model"
-	"github.com/pingcap/tiflow/pkg/errors"
 	"go.uber.org/zap"
 )
 
+const timeout = 5 * time.Second
+
+// TODO: Use the flollowing SQL to check the ddl job status after tidb optimize
+// the information_schema.ddl_jobs table. Ref: https://github.com/pingcap/tidb/issues/55725
+//
+// SELECT JOB_ID, JOB_TYPE, SCHEMA_STATE, SCHEMA_ID, TABLE_ID, STATE, QUERY
+// FROM information_schema.ddl_jobs
 var checkRunningAddIndexSQL = `
-SELECT JOB_ID, JOB_TYPE, SCHEMA_STATE, SCHEMA_ID, TABLE_ID, STATE, QUERY
-FROM information_schema.ddl_jobs
+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")
-LIMIT 1;
+    AND (STATE = "running" OR STATE = "queueing");
 `
 
 func (m *DDLSink) shouldAsyncExecDDL(ddl *model.DDLEvent) bool {
@@ -92,9 +96,23 @@ func (m *DDLSink) asyncExecDDL(ctx context.Context, ddl *model.DDLEvent) error {
 	}
 }
 
+func (m *DDLSink) needWaitAsyncExecDone(t timodel.ActionType) bool {
+	if !m.cfg.IsTiDB {
+		return false
+	}
+	switch t {
+	case timodel.ActionCreateTable, timodel.ActionCreateTables:
+		return false
+	case timodel.ActionCreateSchema:
+		return false
+	default:
+		return true
+	}
+}
+
 // Should always wait for async ddl done before executing the next ddl.
 func (m *DDLSink) waitAsynExecDone(ctx context.Context, ddl *model.DDLEvent) {
-	if !m.cfg.IsTiDB {
+	if !m.needWaitAsyncExecDone(ddl.Type) {
 		return
 	}
 
@@ -105,16 +123,17 @@ func (m *DDLSink) waitAsynExecDone(ctx context.Context, ddl *model.DDLEvent) {
 	if ddl.PreTableInfo != nil {
 		tables[ddl.PreTableInfo.TableName] = struct{}{}
 	}
-	if len(tables) == 0 || m.checkAsyncExecDDLDone(ctx, tables) {
-		return
-	}
 
 	log.Debug("wait async exec ddl done",
 		zap.String("namespace", m.id.Namespace),
 		zap.String("changefeed", m.id.ID),
 		zap.Any("tables", tables),
 		zap.Uint64("commitTs", ddl.CommitTs),
 		zap.String("ddl", ddl.Query))
+	if len(tables) == 0 || m.checkAsyncExecDDLDone(ctx, tables) {
+		return
+	}
+
 	ticker := time.NewTicker(5 * time.Second)
 	defer ticker.Stop()
 	for {
@@ -131,6 +150,8 @@ func (m *DDLSink) waitAsynExecDone(ctx context.Context, ddl *model.DDLEvent) {
 }
 
 func (m *DDLSink) checkAsyncExecDDLDone(ctx context.Context, tables map[model.TableName]struct{}) bool {
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
 	for table := range tables {
 		done := m.doCheck(ctx, table)
 		if !done {
@@ -141,6 +162,7 @@ func (m *DDLSink) checkAsyncExecDDLDone(ctx context.Context, tables map[model.Ta
 }
 
 func (m *DDLSink) doCheck(ctx context.Context, table model.TableName) (done bool) {
+	start := time.Now()
 	if v, ok := m.lastExecutedNormalDDLCache.Get(table); ok {
 		ddlType := v.(timodel.ActionType)
 		if ddlType == timodel.ActionAddIndex {
@@ -152,35 +174,41 @@ func (m *DDLSink) doCheck(ctx context.Context, table model.TableName) (done bool
 		return true
 	}
 
-	ret := m.db.QueryRowContext(ctx, fmt.Sprintf(checkRunningAddIndexSQL, table.Schema, table.Table))
-	if ret.Err() != nil {
+	rows, err := m.db.QueryContext(ctx, fmt.Sprintf(checkRunningAddIndexSQL, table.Schema, table.Table))
+	defer func() {
+		if rows != nil {
+			_ = rows.Err()
+		}
+	}()
+	if err != nil {
 		log.Error("check async exec ddl failed",
 			zap.String("namespace", m.id.Namespace),
 			zap.String("changefeed", m.id.ID),
-			zap.Error(ret.Err()))
+			zap.Error(err))
 		return true
 	}
-	var jobID, jobType, schemaState, schemaID, tableID, state, query string
-	if err := ret.Scan(&jobID, &jobType, &schemaState, &schemaID, &tableID, &state, &query); err != nil {
-		if !errors.Is(err, sql.ErrNoRows) {
-			log.Error("check async exec ddl failed",
-				zap.String("namespace", m.id.Namespace),
-				zap.String("changefeed", m.id.ID),
-				zap.Error(err))
-		}
+	rets, err := export.GetSpecifiedColumnValuesAndClose(rows, "JOB_ID", "JOB_TYPE", "SCHEMA_STATE", "STATE")
+	if err != nil {
+		log.Error("check async exec ddl failed",
+			zap.String("namespace", m.id.Namespace),
+			zap.String("changefeed", m.id.ID),
+			zap.Error(err))
 		return true
 	}
 
+	if len(rets) == 0 {
+		return true
+	}
+	ret := rets[0]
+	jobID, jobType, schemaState, state := ret[0], ret[1], ret[2], ret[3]
 	log.Info("async ddl is still running",
 		zap.String("namespace", m.id.Namespace),
 		zap.String("changefeed", m.id.ID),
+		zap.Duration("checkDuration", time.Since(start)),
 		zap.String("table", table.String()),
 		zap.String("jobID", jobID),
 		zap.String("jobType", jobType),
 		zap.String("schemaState", schemaState),
-		zap.String("schemaID", schemaID),
-		zap.String("tableID", tableID),
-		zap.String("state", state),
-		zap.String("query", query))
+		zap.String("state", state))
 	return false
 }

cdc/sink/ddlsink/mysql/async_ddl_test.go

@@ -45,8 +45,14 @@ func TestWaitAsynExecDone(t *testing.T) {
 
 		// Case 1: there is a running add index job
 		mock.ExpectQuery(fmt.Sprintf(checkRunningAddIndexSQL, "test", "sbtest0")).WillReturnRows(
-			sqlmock.NewRows([]string{"JOB_ID", "JOB_TYPE", "SCHEMA_STATE", "SCHEMA_ID", "TABLE_ID", "STATE", "QUERY"}).
-				AddRow("1", "add index", "running", "1", "1", "running", "Create index idx1 on test.sbtest0(a)"),
+			sqlmock.NewRows([]string{
+				"JOB_ID", "DB_NAME", "TABLE_NAME", "JOB_TYPE", "SCHEMA_STATE", "SCHEMA_ID", "TABLE_ID",
+				"ROW_COUNT", "CREATE_TIME", "START_TIME", "END_TIME", "STATE",
+			}).AddRow(
+				1, "test", "sbtest0", "add index", "write reorganization", 1, 1, 0, time.Now(), nil, time.Now(), "running",
+			).AddRow(
+				2, "test", "sbtest0", "add index", "write reorganization", 1, 1, 0, time.Now(), time.Now(), time.Now(), "queueing",
+			),
 		)
 		// Case 2: there is no running add index job
 		// Case 3: no permission to query ddl_jobs, TiDB will return empty result
@@ -157,3 +163,21 @@ func TestAsyncExecAddIndex(t *testing.T) {
 	require.True(t, time.Since(start) >= 10*time.Second)
 	sink.Close()
 }
+
+func TestNeedWaitAsyncExecDone(t *testing.T) {
+	sink := &DDLSink{
+		cfg: &pmysql.Config{
+			IsTiDB: false,
+		},
+	}
+	require.False(t, sink.needWaitAsyncExecDone(timodel.ActionTruncateTable))
+
+	sink.cfg.IsTiDB = true
+	require.True(t, sink.needWaitAsyncExecDone(timodel.ActionTruncateTable))
+	require.True(t, sink.needWaitAsyncExecDone(timodel.ActionDropTable))
+	require.True(t, sink.needWaitAsyncExecDone(timodel.ActionDropIndex))
+
+	require.False(t, sink.needWaitAsyncExecDone(timodel.ActionCreateTable))
+	require.False(t, sink.needWaitAsyncExecDone(timodel.ActionCreateTables))
+	require.False(t, sink.needWaitAsyncExecDone(timodel.ActionCreateSchema))
+}

cdc/sink/ddlsink/mysql/format_ddl.go

@@ -0,0 +1,62 @@
+// Copyright 2024 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mysql
+
+import (
+	"bytes"
+
+	"github.com/pingcap/log"
+	"github.com/pingcap/tidb/pkg/parser"
+	"github.com/pingcap/tidb/pkg/parser/ast"
+	"github.com/pingcap/tidb/pkg/parser/format"
+	"github.com/pingcap/tidb/pkg/parser/mysql"
+	"go.uber.org/zap"
+)
+
+type visiter struct{}
+
+func (f *visiter) Enter(n ast.Node) (node ast.Node, skipChildren bool) {
+	switch v := n.(type) {
+	case *ast.ColumnDef:
+		switch v.Tp.GetType() {
+		case mysql.TypeTiDBVectorFloat32:
+			v.Tp.SetType(mysql.TypeLongBlob)
+			v.Tp.SetCharset("")
+			v.Tp.SetCollate("")
+			v.Tp.SetFlen(-1)
+			v.Options = []*ast.ColumnOption{} // clear COMMENT
+		}
+	}
+	return n, false
+}
+
+func (f *visiter) Leave(n ast.Node) (node ast.Node, ok bool) {
+	return n, true
+}
+
+func formatQuery(sql string) string {
+	p := parser.New()
+	stmt, err := p.ParseOneStmt(sql, "", "")
+	if err != nil {
+		log.Error("format query parse one stmt failed", zap.Error(err))
+	}
+	stmt.Accept(&visiter{})
+
+	buf := new(bytes.Buffer)
+	restoreCtx := format.NewRestoreCtx(format.DefaultRestoreFlags, buf)
+	if err = stmt.Restore(restoreCtx); err != nil {
+		log.Error("format query restore failed", zap.Error(err))
+	}
+	return buf.String()
+}

cdc/sink/ddlsink/mysql/format_ddl_test.go

@@ -0,0 +1,43 @@
+// Copyright 2024 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mysql
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/pingcap/log"
+	"github.com/pingcap/tidb/pkg/parser"
+	"github.com/pingcap/tidb/pkg/parser/format"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+func TestFormatQuery(t *testing.T) {
+	sql := "CREATE TABLE `test` (`id` INT PRIMARY KEY,`data` VECTOR(5));"
+	expectSql := "CREATE TABLE `test` (`id` INT PRIMARY KEY,`data` LONGTEXT)"
+	p := parser.New()
+	stmt, err := p.ParseOneStmt(sql, "", "")
+	if err != nil {
+		log.Error("format query parse one stmt failed", zap.Error(err))
+	}
+	stmt.Accept(&visiter{})
+
+	buf := new(bytes.Buffer)
+	restoreCtx := format.NewRestoreCtx(format.DefaultRestoreFlags, buf)
+	if err = stmt.Restore(restoreCtx); err != nil {
+		log.Error("format query restore failed", zap.Error(err))
+	}
+	require.Equal(t, buf.String(), expectSql)
+}