support TiCDC

Author: bb7133

r/example.result

@@ -27,36 +27,14 @@ Error 1064 (42000): You have an error in your SQL syntax; check the manual that
 explain analyze format='brief' select * from t;
 id	estRows	actRows	task	access object	execution info	operator info	memory	disk
 TableReader	10000.00	5	root	NULL	time:<num>, loops:<num>, RU:<num>, cop_task: {num:<num>, max:<num>, proc_keys:<num>, tot_proc:<num>, tot_wait:<num>, copr_cache_hit_ratio:<num>, build_task_duration:<num>, max_distsql_concurrency:<num>, rpc_info:{Cop:{num_rpc:<num>, total_time:<num>}}	data:TableFullScan	<num> Bytes	N/A
-└─TableFullScan	10000.00	5	cop[tikv]	table:t	tikv_task:{time:<num>, loops:<num>, scan_detail: {total_process_keys:<num>, total_process_keys_size:<num>, total_keys:<num>, get_snapshot_time:<num>, rocksdb: {key_skipped_count:<num>, block: {}}}, time_detail: {total_process_time:<num>, total_wait_time:<num>, tikv_wall_time:<num>}	keep order:false, stats:pseudo	N/A	N/A
+└─TableFullScan	10000.00	5	cop[tikv]	table:t	tikv_task:{time:<num>, loops:<num>, scan_detail: {total_process_keys:<num>, total_process_keys_size:<num>, total_keys:<num>, get_snapshot_time:<num>, rocksdb: {delete_skipped_count:<num>, key_skipped_count:<num>, block: {}}}, time_detail: {total_process_time:<num>, total_wait_time:<num>, tikv_wall_time:<num>}	keep order:false, stats:pseudo	N/A	N/A
 explain analyze select * from t;
 id	estRows	actRows	task	access object	execution info	operator info	memory	disk
 TableReader_5	10000.00	5	root	NULL	time:<num>, loops:<num>, RU:<num>, cop_task: {num:<num>, max:<num>, proc_keys:<num>, tot_proc:<num>, tot_wait:<num>, copr_cache_hit_ratio:<num>, build_task_duration:<num>, max_distsql_concurrency:<num>, rpc_info:{Cop:{num_rpc:<num>, total_time:<num>}}	data:TableFullScan_4	<num> Bytes	N/A
-└─TableFullScan_4	10000.00	5	cop[tikv]	table:t	tikv_task:{time:<num>, loops:<num>, scan_detail: {total_process_keys:<num>, total_process_keys_size:<num>, total_keys:<num>, get_snapshot_time:<num>, rocksdb: {key_skipped_count:<num>, block: {}}}, time_detail: {total_process_time:<num>, total_wait_time:<num>, tikv_wall_time:<num>}	keep order:false, stats:pseudo	N/A	N/A
+└─TableFullScan_4	10000.00	5	cop[tikv]	table:t	tikv_task:{time:<num>, loops:<num>, scan_detail: {total_process_keys:<num>, total_process_keys_size:<num>, total_keys:<num>, get_snapshot_time:<num>, rocksdb: {delete_skipped_count:<num>, key_skipped_count:<num>, block: {}}}, time_detail: {total_process_time:<num>, total_wait_time:<num>, tikv_wall_time:<num>}	keep order:false, stats:pseudo	N/A	N/A
 insert into t values (6, 6);
 affected rows: 1
 info: 
-Destination	Size	BackupTS	Queue Time	Execution Time
-/tmp/t_b7987183-d39f-4572-868d-75e25c8cd215	1876	451473879442653228	2024-07-29 14:56:13	2024-07-29 14:56:13
-affected rows: 0
-info: 
-affected rows: 0
-info: 
-affected rows: 0
-info: 
-Destination	Size	BackupTS	Cluster TS	Queue Time	Execution Time
-/tmp/t_b7987183-d39f-4572-868d-75e25c8cd215	1876	451473879442653228	451473880386371620	2024-07-29 14:56:17	2024-07-29 14:56:17
-affected rows: 0
-info: 
-affected rows: 0
-info: 
-affected rows: 0
-info: 
-affected rows: 0
-info: 
-Job_ID	Data_Source	Target_Table	Table_ID	Phase	Status	Source_File_Size	Imported_Rows	Result_Message	Create_Time	Start_Time	End_Time	Created_By
-3	/tmp/t_6cac1a43-c66c-4af9-962f-95287fa12432/example.t.000000000.csv	`example`.`td`	453		finished	30B	6		2024-07-29 14:56:17.619215	2024-07-29 14:56:18.125792	2024-07-29 14:56:19.640005	root@%
-affected rows: 0
-info: 
 DROP TABLE IF EXISTS t1;
 affected rows: 0
 info: 
@@ -71,3 +49,8 @@ affected rows: 3
 info: Records: 2  Duplicates: 1  Warnings: 0
 1
 use `test`;;
+use example;
+select * from t1;
+f1	f2
+1	1
+2	2

src/main.go

@@ -51,6 +51,14 @@ var (
 	checkErr         bool
 	pathBR           string
 	pathDumpling     string
+	pathCDC          string
+	downstream       string
+
+	downStreamHost     string
+	downStreamPort     string
+	downStreamUser     string
+	downStreamPassword string
+	downStreamDB       string
 )
 
 func init() {
@@ -67,8 +75,10 @@ func init() {
 	flag.IntVar(&retryConnCount, "retry-connection-count", 120, "The max number to retry to connect to the database.")
 	flag.BoolVar(&checkErr, "check-error", false, "if --error ERR does not match, return error instead of just warn")
 	flag.BoolVar(&collationDisable, "collation-disable", false, "run collation related-test with new-collation disabled")
-	flag.StringVar(&pathBR, "path-br", "", "Path of BR")
-	flag.StringVar(&pathDumpling, "path-dumpling", "", "Path of Dumpling")
+	flag.StringVar(&pathBR, "path-br", "", "Path of BR binary")
+	flag.StringVar(&pathDumpling, "path-dumpling", "", "Path of Dumpling binary")
+	flag.StringVar(&pathCDC, "path-cdc", "", "Path of TiCDC binary")
+	flag.StringVar(&downstream, "downstream", "", "Connection string of downstream TiDB cluster")
 }
 
 const (
@@ -165,6 +175,12 @@ type tester struct {
 
 	// dump and import context through --dump_and_import $SOURCE_TABLE as $TARGET_TABLE'
 	dumpAndImport *SourceAndTarget
+
+	// replication checkpoint database name
+	replicationCheckpointDB string
+
+	// replication checkpoint ID
+	replicationCheckpointID int
 }
 
 func newTester(name string) *tester {
@@ -179,6 +195,8 @@ func newTester(name string) *tester {
 	t.enableConcurrent = false
 	t.enableInfo = false
 
+	t.replicationCheckpointDB = "checkpoint-" + uuid.NewString()
+	t.replicationCheckpointID = 0
 	return t
 }
 
@@ -219,7 +237,7 @@ func isTiDB(db *sql.DB) bool {
 	return true
 }
 
-func (t *tester) addConnection(connName, hostName, userName, password, db string) {
+func (t *tester) addConnection(connName, hostName, port, userName, password, db string) {
 	var (
 		mdb *sql.DB
 		err error
@@ -285,6 +303,64 @@ func (t *tester) disconnect(connName string) {
 	t.currConnName = default_connection
 }
 
+func parseUserInfo(userInfo string) (string, string, error) {
+	colonIndex := strings.Index(userInfo, ":")
+	if colonIndex == -1 {
+		return "", "", fmt.Errorf("missing password in userinfo")
+	}
+	return userInfo[:colonIndex], userInfo[colonIndex+1:], nil
+}
+
+func parseHostPort(hostPort string) (string, string, error) {
+	colonIndex := strings.Index(hostPort, ":")
+	if colonIndex == -1 {
+		return "", "", fmt.Errorf("missing port in host:port")
+	}
+	return hostPort[:colonIndex], hostPort[colonIndex+1:], nil
+}
+
+func parseDownstream(connStr string) (dbname string, host string, port string, user string, password string) {
+	// Splitting into userinfo and network/database parts
+	parts := strings.SplitN(connStr, "@", 2)
+	if len(parts) != 2 {
+		fmt.Println("Invalid connection string format")
+		return
+	}
+
+	// Parsing userinfo
+	userInfo := parts[0]
+	user, password, err := parseUserInfo(userInfo)
+	if err != nil {
+		fmt.Println("Error parsing userinfo:", err)
+		return
+	}
+
+	// Splitting network type and database part
+	networkAndDB := parts[1]
+	networkTypeIndex := strings.Index(networkAndDB, "(")
+	if networkTypeIndex == -1 {
+		fmt.Println("Invalid connection string format: missing network type")
+		return
+	}
+
+	// Extracting host, port, and database name
+	hostPortDB := networkAndDB[networkTypeIndex+1:]
+	hostPortDBParts := strings.SplitN(hostPortDB, ")/", 2)
+	if len(hostPortDBParts) != 2 {
+		fmt.Println("Invalid connection string format")
+		return
+	}
+
+	host, port, err = parseHostPort(hostPortDBParts[0])
+	if err != nil {
+		fmt.Println("Error parsing host and port:", err)
+		return
+	}
+
+	dbname = hostPortDBParts[1]
+	return
+}
+
 func (t *tester) preProcess() {
 	dbName := "test"
 	mdb, err := OpenDBWithRetry("mysql", user+":"+passwd+"@tcp("+host+":"+port+")/"+dbName+"?time_zone=%27Asia%2FShanghai%27&allowAllFiles=true"+params, retryConnCount)
@@ -303,6 +379,7 @@ func (t *tester) preProcess() {
 		}
 		for rows.Next() {
 			rows.Scan(&dbName)
+			fmt.Println("Scanning database:", dbName)
 			t.originalSchemas[dbName] = struct{}{}
 		}
 	}
@@ -313,13 +390,32 @@ func (t *tester) preProcess() {
 		log.Fatalf("Executing create db %s err[%v]", dbName, err)
 	}
 	t.mdb = mdb
+
 	conn, err := initConn(mdb, user, passwd, host, dbName)
 	if err != nil {
 		log.Fatalf("Open db err %v", err)
 	}
 	t.conn[default_connection] = conn
 	t.curr = conn
 	t.currConnName = default_connection
+
+	if downstream != "" {
+		// create replication checkpoint database
+		if _, err := t.mdb.Exec(fmt.Sprintf("create database if not exists `%s`", t.replicationCheckpointDB)); err != nil {
+			log.Fatalf("Executing create db %s err[%v]", t.replicationCheckpointDB, err)
+		}
+
+		downStreamDB, downStreamHost, downStreamPort, downStreamUser, downStreamPassword = parseDownstream(downstream)
+
+		fmt.Println("downStreamDB:", downStreamDB)
+		fmt.Println("downStreamHost:", downStreamHost)
+		fmt.Println("downStreamPort:", downStreamPort)
+		fmt.Println("downStreamUser:", downStreamUser)
+		fmt.Println("downStreamPassword:", downStreamPassword)
+
+		t.addConnection("downstream", downStreamHost, downStreamPort, downStreamUser, downStreamPassword, downStreamDB)
+	}
+	t.switchConnection(default_connection)
 }
 
 func (t *tester) postProcess() {
@@ -329,6 +425,7 @@ func (t *tester) postProcess() {
 		}
 		t.mdb.Close()
 	}()
+	t.switchConnection(default_connection)
 	if !reserveSchema {
 		rows, err := t.mdb.Query("show databases")
 		if err != nil {
@@ -339,6 +436,7 @@ func (t *tester) postProcess() {
 		for rows.Next() {
 			rows.Scan(&dbName)
 			if _, exists := t.originalSchemas[dbName]; !exists {
+				fmt.Println("Dropping database:", dbName)
 				_, err := t.mdb.Exec(fmt.Sprintf("drop database `%s`", dbName))
 				if err != nil {
 					log.Errorf("failed to drop database: %s", err.Error())
@@ -421,6 +519,49 @@ func (t *tester) importTableStmt(path, target string) string {
     `, target, path)
 }
 
+func (t *tester) waitForReplicationCheckpoint() error {
+	curr := t.currConnName
+	defer t.switchConnection(curr)
+
+	if err := t.executeStmt(fmt.Sprintf("use `%s`", t.replicationCheckpointDB)); err != nil {
+		return err
+	}
+
+	markerTable := fmt.Sprintf("marker_%d", t.replicationCheckpointID)
+	if err := t.executeStmt(fmt.Sprintf("create table `%s`.`%s` (id int primary key)", t.replicationCheckpointDB, markerTable)); err != nil {
+		return err
+	}
+
+	t.switchConnection("downstream")
+
+	checkInterval := 1 * time.Second
+	queryTimeout := 10 * time.Second
+
+	// Keep querying until the table is found
+	for {
+		ctx, cancel := context.WithTimeout(context.Background(), queryTimeout)
+		defer cancel()
+
+		query := fmt.Sprintf("select * from information_schema.tables where table_schema = '%s' and table_name = '%s';", t.replicationCheckpointDB, markerTable)
+		rows, err := t.mdb.QueryContext(ctx, query)
+		if err != nil {
+			log.Printf("Error checking for table: %v", err)
+			return err
+		}
+
+		if rows.Next() {
+			fmt.Printf("Table '%s' found!\n", markerTable)
+			break
+		} else {
+			fmt.Printf("Table '%s' not found. Retrying in %v...\n", markerTable, checkInterval)
+		}
+
+		time.Sleep(checkInterval)
+	}
+
+	return nil
+}
+
 func (t *tester) Run() error {
 	t.preProcess()
 	defer t.postProcess()
@@ -543,7 +684,7 @@ func (t *tester) Run() error {
 			for i := 0; i < 4; i++ {
 				args = append(args, "")
 			}
-			t.addConnection(args[0], args[1], args[2], args[3], args[4])
+			t.addConnection(args[0], args[1], port, args[2], args[3], args[4])
 		case Q_CONNECTION:
 			q.Query = strings.TrimSpace(q.Query)
 			if q.Query[len(q.Query)-1] == ';' {
@@ -646,7 +787,10 @@ func (t *tester) Run() error {
 				return err
 			}
 			log.WithFields(log.Fields{"stmt": importStmt, "line": q.Line}).Warn("Restore end")
-
+		case Q_REPLICATION_CHECKPOINT:
+			if err := t.waitForReplicationCheckpoint(); err != nil {
+				return err
+			}
 		default:
 			log.WithFields(log.Fields{"command": q.firstWord, "arguments": q.Query, "line": q.Line}).Warn("command not implemented")
 		}

src/query.go

@@ -126,6 +126,7 @@ const (
 	Q_EMPTY_LINE
 	Q_BACKUP_AND_RESTORE
 	Q_DUMP_AND_IMPORT
+	Q_REPLICATION_CHECKPOINT
 )
 
 // ParseQueries parses an array of string into an array of query object.

src/type.go

@@ -115,7 +115,8 @@ var commandMap = map[string]int{
 	"begin_concurrent":           Q_BEGIN_CONCURRENT,
 	"end_concurrent":             Q_END_CONCURRENT,
 	"backup_and_restore":         Q_BACKUP_AND_RESTORE,
-	"dump_and_import":			Q_DUMP_AND_IMPORT,
+	"dump_and_import":            Q_DUMP_AND_IMPORT,
+	"replication_checkpoint":     Q_REPLICATION_CHECKPOINT,
 }
 
 func findType(cmdName string) int {

t/example.test

@@ -38,9 +38,9 @@ explain analyze select * from t;
 --enable_info
 insert into t values (6, 6);
 
---backup_and_restore t AS tt
+# --backup_and_restore t AS tt
 
---dump_and_import t AS td
+# --dump_and_import t AS td
 
 DROP TABLE IF EXISTS t1;
 CREATE TABLE t1 (f1 INT PRIMARY KEY, f2 INT NOT NULL UNIQUE);
@@ -52,3 +52,10 @@ INSERT t1 VALUES (1, 1), (1, 1) ON DUPLICATE KEY UPDATE f1 = 2, f2 = 2;
 --echo $a
 
 use `test`;;
+
+sleep 10;
+
+--replication_checkpoint
+connection default;
+use example;
+select * from t1;