tracker(dm): fix tracker panic when pk of downstream table orders beh…

…ind (pingcap#5165) (pingcap#5167)

close pingcap#5159

dm/pkg/schema/tracker.go

@@ -533,7 +533,7 @@ func GetDownStreamTi(ti *model.TableInfo, originTi *model.TableInfo) *Downstream
 		return mysql.HasNotNullFlag(ti.Columns[i].Flag)
 	}
 
-	for i, idx := range ti.Indices {
+	for _, idx := range ti.Indices {
 		if !idx.Primary && !idx.Unique {
 			continue
 		}
@@ -544,12 +544,12 @@ func GetDownStreamTi(ti *model.TableInfo, originTi *model.TableInfo) *Downstream
 		availableUKIndexList = append(availableUKIndexList, indexRedirect)
 		if idx.Primary {
 			absoluteUKIndexInfo = indexRedirect
-			absoluteUKPosition = i
+			absoluteUKPosition = len(availableUKIndexList) - 1
 			hasPk = true
 		} else if absoluteUKIndexInfo == nil && isSpecifiedIndexColumn(idx, fn) {
 			// second check not null unique key
 			absoluteUKIndexInfo = indexRedirect
-			absoluteUKPosition = i
+			absoluteUKPosition = len(availableUKIndexList) - 1
 		}
 	}
 

dm/tests/downstream_index_order/conf/diff_config.toml

@@ -0,0 +1,30 @@
+# diff Configuration.
+
+check-thread-count = 4
+
+export-fix-sql = true
+
+check-struct-only = false
+
+[task]
+    output-dir = "/tmp/ticdc_dm_test/output"
+
+    source-instances = ["mysql1"]
+
+    target-instance = "tidb0"
+
+    target-check-tables = ["dm_syncer_tracker.*"]
+
+
+[data-sources]
+[data-sources.mysql1]
+host = "127.0.0.1"
+port = 3306
+user = "root"
+password = "123456"
+
+[data-sources.tidb0]
+host = "127.0.0.1"
+port = 4000
+user = "test"
+password = "123456"

dm/tests/downstream_index_order/conf/dm-master.toml

@@ -0,0 +1,4 @@
+# Master Configuration.
+master-addr = ":8261"
+advertise-addr = "127.0.0.1:8261"
+auto-compaction-retention = "3s"

dm/tests/downstream_index_order/conf/dm-task.yaml

@@ -0,0 +1,46 @@
+---
+name: test
+task-mode: "all"
+is-sharding: false
+meta-schema: "dm_meta"
+# enable-heartbeat: true
+heartbeat-update-interval: 1
+heartbeat-report-interval: 1
+clean-dump-file: false
+
+target-database:
+  host: "127.0.0.1"
+  port: 4000
+  user: "root"
+  password: ""
+
+mysql-instances:
+  - source-id: "mysql-replica-01"
+    meta:
+      binlog-name: binlog-name-placeholder-1
+      binlog-pos: 4
+    block-allow-list:  "instance"
+    mydumper-config-name: "global"
+    loader-config-name: "global"
+    syncer-config-name: "global"
+
+block-allow-list:
+  instance:
+    do-dbs: ["dm_syncer_tracker"]
+
+mydumpers:
+  global:
+    threads: 4
+    chunk-filesize: 64
+    skip-tz-utc: true
+    extra-args: ""
+
+loaders:
+  global:
+    pool-size: 16
+    dir: "./dumped_data"
+
+syncers:
+  global:
+    worker-count: 16
+    batch: 100

dm/tests/downstream_index_order/conf/dm-worker1.toml

@@ -0,0 +1,2 @@
+name = "worker1"
+join = "127.0.0.1:8261"

dm/tests/downstream_index_order/conf/source1.yaml

@@ -0,0 +1,11 @@
+source-id: mysql-replica-01
+flavor: ''
+enable-gtid: false
+enable-relay: false
+relay-binlog-name: ''
+relay-binlog-gtid: ''
+from:
+  host: 127.0.0.1
+  user: root
+  password: '123456'
+  port: 3306

dm/tests/downstream_index_order/data/db1.increment.sql

@@ -0,0 +1,15 @@
+use dm_syncer_tracker;
+create table t(
+  a int not null,
+  b int not null,
+  c int not null,
+  d int not null,
+  e int not null,
+  index (a),
+  index(b),
+  index(c)
+);
+
+alter table t add unique(d);
+alter table t add unique(e);
+insert into t values(1,1,1,1,1);

dm/tests/downstream_index_order/data/db1.prepare.sql

@@ -0,0 +1,2 @@
+drop database if exists `dm_syncer_tracker`;
+create database `dm_syncer_tracker`;

dm/tests/downstream_index_order/run.sh

@@ -0,0 +1,53 @@
+#!/bin/bash
+
+set -eu
+
+cur=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $cur/../_utils/test_prepare
+WORK_DIR=$TEST_DIR/$TEST_NAME
+db1="downstream_more_column1"
+tb1="t1"
+db="downstream_more_column"
+tb="t"
+
+# pk/uk added using alter stmt in tidb will be appended to the table
+# unlike mysql which will put pk/uk at the beginning
+# this test make sure we can handle this case
+# may panic previously since tracker uses the wrong index to get pk
+# see https://github.com/pingcap/tiflow/issues/5159
+function run() {
+	run_sql_file $cur/data/db1.prepare.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+
+	run_dm_master $WORK_DIR/master $MASTER_PORT $cur/conf/dm-master.toml
+	check_rpc_alive $cur/../bin/check_master_online 127.0.0.1:$MASTER_PORT
+	check_metric $MASTER_PORT 'start_leader_counter' 3 0 2
+	run_dm_worker $WORK_DIR/worker1 $WORKER1_PORT $cur/conf/dm-worker1.toml
+	check_rpc_alive $cur/../bin/check_worker_online 127.0.0.1:$WORKER1_PORT
+
+	cp $cur/conf/source1.yaml $WORK_DIR/source1.yaml
+	sed -i "/relay-binlog-name/i\relay-dir: $WORK_DIR/worker1/relay_log" $WORK_DIR/source1.yaml
+	dmctl_operate_source create $WORK_DIR/source1.yaml $SOURCE_ID1
+
+	# start a task
+	cat $cur/conf/dm-task.yaml >$WORK_DIR/dm-task.yaml
+	dmctl_start_task_standalone $WORK_DIR/dm-task.yaml "--remove-meta"
+
+	# make sure we're in sync unit
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"\"result\": true" 2 \
+		"\"unit\": \"Sync\"" 1
+
+	run_sql_file $cur/data/db1.increment.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+	check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
+	run_sql "select * from dm_syncer_tracker.t" $TIDB_PORT $TIDB_PASSWORD
+	check_not_contains "dm_syncer_ignore_db"
+}
+
+cleanup_data downstream_more_column
+# also cleanup dm processes in case of last run failed
+cleanup_process $*
+run $*
+cleanup_process $*
+
+echo "[$(date)] <<<<<< test case $TEST_NAME success! >>>>>>"

dm/tests/others_integration_1.txt

@@ -10,3 +10,4 @@ checkpoint_transaction
 lightning_mode
 slow_relay_writer
 sync_collation
+downstream_index_order