ha: refactor online DDL

dm/master/shardddl/pessimist_test.go

@@ -226,7 +226,7 @@ func (t *testPessimist) TestPessimist(c *C) {
 	op21c.Done = true
 	_, err = pessimism.PutOperationDeleteInfo(etcdTestCli, op21c, i21)
 	c.Assert(err, IsNil)
-	c.Assert(utils.WaitSomething(30, 10*time.Millisecond, func() bool {
+	c.Assert(utils.WaitSomething(30, 100*time.Millisecond, func() bool {
 		return p.Locks()[ID2].IsDone(source1)
 	}), IsTrue)
 

syncer/online_ddl.go

@@ -16,7 +16,6 @@ package syncer
 import (
 	"encoding/json"
 	"fmt"
-	"strconv"
 	"strings"
 	"sync"
 
@@ -87,7 +86,7 @@ type OnlineDDLStorage struct {
 	dbConn *DBConn
 	schema string // schema name, set through task config
 	table  string // table name, now it's task name
-	id     string // now it is `server-id` used as MySQL slave
+	id     string // the source ID of the upstream MySQL/MariaDB replica.
 
 	// map ghost schema => [ghost table => ghost ddl info, ...]
 	ddls map[string]map[string]*GhostDDLInfo
@@ -101,7 +100,7 @@ func NewOnlineDDLStorage(logCtx *tcontext.Context, cfg *config.SubTaskConfig) *O
 		cfg:    cfg,
 		schema: cfg.MetaSchema,
 		table:  fmt.Sprintf("%s_onlineddl", cfg.Name),
-		id:     strconv.FormatUint(uint64(cfg.ServerID), 10),
+		id:     cfg.SourceID,
 		ddls:   make(map[string]map[string]*GhostDDLInfo),
 		logCtx: logCtx,
 	}

tests/incremental_mode/conf/dm-task.yaml

@@ -3,7 +3,7 @@ name: test
 task-mode: task-mode-placeholder
 is-sharding: false
 meta-schema: "dm_meta"
-remove-meta: true
+remove-meta: false
 # enable-heartbeat: true
 heartbeat-update-interval: 1
 heartbeat-report-interval: 1

tests/online_ddl/conf/dm-task.yaml

@@ -3,7 +3,7 @@ name: test
 task-mode: all
 is-sharding: true
 meta-schema: "dm_meta"
-remove-meta: true
+remove-meta: false
 enable-heartbeat: false
 timezone: "Asia/Shanghai"
 online-ddl-scheme: online-ddl-scheme-placeholder

tests/online_ddl/conf/dm-worker3.toml

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

tests/online_ddl/data/db1.increment2.sql

@@ -0,0 +1,11 @@
+use online_ddl;
+insert into t1 (uid, name, info) values (10006, 'name of 10006', '{"age": 10006}');
+insert into t2 (uid, name, info) values (20008, 'name of 20008', '{"age": 20008}');
+alter table t1 add column address varchar(255);
+alter table t2 add column address varchar(255);
+insert into t2 (uid, name, info, address) values (20009, 'name of 20009', '{"age": 20009}', 'address of 20009');
+insert into t1 (uid, name, info, address) values (10007, 'name of 10007', '{"age": 10007}', 'address of 10007');
+alter table t2 drop column age;
+alter table t1 drop column age;
+insert into t1 (uid, name, info, address) values (10008, 'name of 10008', '{"age": 10008}', 'address of 10008');
+insert into t2 (uid, name, info, address) values (20010, 'name of 20010', '{"age": 20010}', 'address of 20010');

tests/online_ddl/data/db2.increment2.sql

@@ -0,0 +1,11 @@
+use online_ddl;
+insert into t3 (uid, name, info) values (30004, 'name of 30004', '{"age": 30004}');
+insert into t2 (uid, name, info) values (50002, 'name of 50002', '{"age": 50002}');
+alter table t3 add column address varchar(255);
+alter table t2 add column address varchar(255);
+insert into t2 (uid, name, info, address) values (50003, 'name of 50003', '{"age": 50003}', 'address of 50003');
+insert into t3 (uid, name, info, address) values (30005, 'name of 30005', '{"age": 30005}', 'address of 30005');
+alter table t2 drop column age;
+alter table t3 drop column age;
+insert into t3 (uid, name, info, address) values (30006, 'name of 30006', '{"age": 30006}', 'address of 30006');
+insert into t2 (uid, name, info, address) values (50004, 'name of 50004', '{"age": 50004}', 'address of 50004');

tests/online_ddl/run.sh

@@ -33,15 +33,33 @@ function real_run() {
     sed -i "s/online-ddl-scheme-placeholder/${online_ddl_scheme}/g" $WORK_DIR/dm-task-${online_ddl_scheme}.yaml
     dmctl_start_task "$WORK_DIR/dm-task-${online_ddl_scheme}.yaml"
 
-    # use sync_diff_inspector to check full dump loader
+    echo "use sync_diff_inspector to check full dump data"
     check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
 
-    # skip online ddl test
-    # run_sql_file_online_ddl $cur/data/db1.increment.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1 online_ddl $online_ddl_scheme 
-    # run_sql_file_online_ddl $cur/data/db2.increment.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2 online_ddl $online_ddl_scheme
+    run_sql_file_online_ddl $cur/data/db1.increment.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1 online_ddl $online_ddl_scheme
+    run_sql_file_online_ddl $cur/data/db2.increment.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2 online_ddl $online_ddl_scheme
 
-    # use sync_diff_inspector to check data now!
-    # check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
+    echo "use sync_diff_inspector to check increment data"
+    check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
+
+    echo "start dm-worker3 and kill dm-worker2"
+    ps aux | grep dm-worker2 |awk '{print $2}'|xargs kill || true
+    check_port_offline $WORKER2_PORT 20
+
+    run_dm_worker $WORK_DIR/worker3 $WORKER3_PORT $cur/conf/dm-worker3.toml
+    check_rpc_alive $cur/../bin/check_worker_online 127.0.0.1:$WORKER3_PORT
+
+    echo "wait and check task running"
+    run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+        "query-status test" \
+        "\"stage\": \"Running\"" 2
+
+    run_sql_file $cur/data/db1.increment2.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
+    run_sql_file $cur/data/db2.increment2.sql $MYSQL_HOST2 $MYSQL_PORT2 $MYSQL_PASSWORD2
+    sleep 2
+
+    echo "use sync_diff_inspector to check increment2 data now!"
+    check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
 }
 
 function run() {