shardddl/optimistic: warn add not fully dropped columns

dm/common/common.go

@@ -79,6 +79,15 @@ var (
 	// ShardDDLOptimismInitSchemaKeyAdapter is used to store the initial schema (before constructed the lock) of merged tables.
 	// k/v: Encode(task-name, downstream-schema-name, downstream-table-name) -> table schema.
 	ShardDDLOptimismInitSchemaKeyAdapter KeyAdapter = keyHexEncoderDecoder("/dm-master/shardddl-optimism/init-schema/")
+	// ShardDDLOptimismDroppedColumnsKeyAdapter is used to store the columns that are not fully dropped
+	// k/v: Encode(task-name, downstream-schema-name, downstream-table-name, column-name, source-id, upstream-schema-name, upstream-table-name) -> empty
+	// If we don't identify different upstream tables, we may report an error for tb2 in the following case.
+	// Time series: (+a/-a means add/drop column a)
+	//	    older ----------------> newer
+	// tb1: +a +b +c           -c
+	// tb2:                       +a +b +c
+	// tb3:          +a +b +c
+	ShardDDLOptimismDroppedColumnsKeyAdapter KeyAdapter = keyHexEncoderDecoder("/dm-master/shardddl-optimism/dropped-columns/")
 )
 
 func keyAdapterKeysLen(s KeyAdapter) int {
@@ -95,7 +104,8 @@ func keyAdapterKeysLen(s KeyAdapter) int {
 		return 3
 	case ShardDDLOptimismInfoKeyAdapter, ShardDDLOptimismOperationKeyAdapter:
 		return 4
-
+	case ShardDDLOptimismDroppedColumnsKeyAdapter:
+		return 7
 	}
 	return -1
 }

dm/master/server_test.go

@@ -641,7 +641,7 @@ func (t *testMaster) TestStartTaskWithRemoveMeta(c *check.C) {
 	c.Assert(err, check.IsNil)
 	_, err = optimism.PutInfo(etcdTestCli, info1)
 	c.Assert(err, check.IsNil)
-	_, succ, err = optimism.PutOperation(etcdTestCli, false, op1)
+	_, succ, err = optimism.PutOperation(etcdTestCli, false, op1, 0)
 	c.Assert(succ, check.IsTrue)
 	c.Assert(err, check.IsNil)
 
@@ -1470,7 +1470,7 @@ func (t *testMaster) TestOfflineMember(c *check.C) {
 	c.Assert(err, check.IsNil)
 	c.Assert(listResp.Members, check.HasLen, 3)
 
-	// make sure s3 is not the leader, otherwise it will take some time to campain a new leader after close s3, and it may cause timeout
+	// make sure s3 is not the leader, otherwise it will take some time to campaign a new leader after close s3, and it may cause timeout
 	c.Assert(utils.WaitSomething(20, 500*time.Millisecond, func() bool {
 		_, leaderID, _, err = s1.election.LeaderInfo(ctx)
 		if err != nil {

dm/master/shardddl/optimist.go

@@ -245,35 +245,49 @@ func (o *Optimist) rebuildLocks() (revSource, revInfo, revOperation int64, err e
 	}
 	o.logger.Info("get history shard DDL lock operation", zap.Int64("revision", revOperation))
 
+	colm, _, err := optimism.GetAllDroppedColumns(o.cli)
+	if err != nil {
+		// only log the error, and don't return it to forbid the startup of the DM-master leader.
+		// then these unexpected columns can be handled by the user.
+		o.logger.Error("fail to recover colms", log.ShortError(err))
+	}
+
 	// recover the shard DDL lock based on history shard DDL info & lock operation.
-	err = o.recoverLocks(ifm, opm)
+	err = o.recoverLocks(ifm, opm, colm)
 	if err != nil {
 		// only log the error, and don't return it to forbid the startup of the DM-master leader.
 		// then these unexpected locks can be handled by the user.
 		o.logger.Error("fail to recover locks", log.ShortError(err))
 	}
+
 	return revSource, revInfo, revOperation, nil
 }
 
 // recoverLocks recovers shard DDL locks based on shard DDL info and shard DDL lock operation.
 func (o *Optimist) recoverLocks(
 	ifm map[string]map[string]map[string]map[string]optimism.Info,
-	opm map[string]map[string]map[string]map[string]optimism.Operation) error {
+	opm map[string]map[string]map[string]map[string]optimism.Operation,
+	colm map[string]map[string]map[string]map[string]map[string]interface{}) error {
 	// construct locks based on the shard DDL info.
-	for task, ifTask := range ifm {
+	o.lk.SetColumnMap(colm)
+	defer o.lk.SetColumnMap(nil)
+	var firstErr error
+	setFirstErr := func(err error) {
+		if firstErr == nil && err != nil {
+			firstErr = err
+		}
+	}
+
+	for _, ifTask := range ifm {
 		for _, ifSource := range ifTask {
 			for _, ifSchema := range ifSource {
 				for _, info := range ifSchema {
-					tts := o.tk.FindTables(task, info.DownSchema, info.DownTable)
-					_, _, err := o.lk.TrySync(info, tts)
-					if err != nil {
-						return err
-					}
-					// never mark the lock operation from `done` to `not-done` when recovering.
-					err = o.handleLock(info, tts, true)
-					if err != nil {
-						return err
-					}
+					// We should return err after all infos are set up.
+					// If we stopped recovering locks once we meet an error,
+					// dm-master leader may not have the full information for the other "normal" locks,
+					// which will cause the sync error in dm-worker.
+					err := o.handleInfo(info)
+					setFirstErr(err)
 				}
 			}
 		}
@@ -291,6 +305,11 @@ func (o *Optimist) recoverLocks(
 					}
 					if op.Done {
 						lock.TryMarkDone(op.Source, op.UpSchema, op.UpTable)
+						err := lock.DeleteColumnsByDDLs(op.DDLs)
+						if err != nil {
+							o.logger.Error("fail to update lock columns", zap.Error(err))
+							continue
+						}
 					}
 				}
 			}
@@ -338,7 +357,7 @@ func (o *Optimist) watchSourceInfoOperation(
 	}()
 	go func() {
 		defer wg.Done()
-		o.handleInfo(ctx, infoCh)
+		o.handleInfoPut(ctx, infoCh)
 	}()
 
 	// watch for the shard DDL lock operation and handle them.
@@ -381,8 +400,8 @@ func (o *Optimist) handleSourceTables(ctx context.Context, sourceCh <-chan optim
 	}
 }
 
-// handleInfo handles PUT and DELETE for the shard DDL info.
-func (o *Optimist) handleInfo(ctx context.Context, infoCh <-chan optimism.Info) {
+// handleInfoPut handles PUT and DELETE for the shard DDL info.
+func (o *Optimist) handleInfoPut(ctx context.Context, infoCh <-chan optimism.Info) {
 	for {
 		select {
 		case <-ctx.Done():
@@ -414,35 +433,38 @@ func (o *Optimist) handleInfo(ctx context.Context, infoCh <-chan optimism.Info)
 				continue
 			}
 
-			added := o.tk.AddTable(info.Task, info.Source, info.UpSchema, info.UpTable, info.DownSchema, info.DownTable)
-			o.logger.Debug("a table added for info", zap.Bool("added", added), zap.Stringer("info", info))
-
-			tts := o.tk.FindTables(info.Task, info.DownSchema, info.DownTable)
-			if tts == nil {
-				// WATCH for SourceTables may fall behind WATCH for Info although PUT earlier,
-				// so we try to get SourceTables again.
-				// NOTE: check SourceTables for `info.Source` if needed later.
-				stm, _, err := optimism.GetAllSourceTables(o.cli)
-				if err != nil {
-					o.logger.Error("fail to get source tables", log.ShortError(err))
-				} else if tts2 := optimism.TargetTablesForTask(info.Task, info.DownSchema, info.DownTable, stm); tts2 != nil {
-					tts = tts2
-				}
-			}
-			// put operation for the table. we don't set `skipDone=true` now,
-			// because in optimism mode, one table may execute/done multiple DDLs but other tables may do nothing.
-			err := o.handleLock(info, tts, false)
-			if err != nil {
-				o.logger.Error("fail to handle the shard DDL lock", zap.Stringer("info", info), log.ShortError(err))
-				metrics.ReportDDLError(info.Task, metrics.InfoErrHandleLock)
-				o.mu.Unlock()
-				continue
-			}
+			_ = o.handleInfo(info)
 			o.mu.Unlock()
 		}
 	}
 }
 
+func (o *Optimist) handleInfo(info optimism.Info) error {
+	added := o.tk.AddTable(info.Task, info.Source, info.UpSchema, info.UpTable, info.DownSchema, info.DownTable)
+	o.logger.Debug("a table added for info", zap.Bool("added", added), zap.Stringer("info", info))
+
+	tts := o.tk.FindTables(info.Task, info.DownSchema, info.DownTable)
+	if tts == nil {
+		// WATCH for SourceTables may fall behind WATCH for Info although PUT earlier,
+		// so we try to get SourceTables again.
+		// NOTE: check SourceTables for `info.Source` if needed later.
+		stm, _, err := optimism.GetAllSourceTables(o.cli)
+		if err != nil {
+			o.logger.Error("fail to get source tables", log.ShortError(err))
+		} else if tts2 := optimism.TargetTablesForTask(info.Task, info.DownSchema, info.DownTable, stm); tts2 != nil {
+			tts = tts2
+		}
+	}
+	// put operation for the table. we don't set `skipDone=true` now,
+	// because in optimism mode, one table may execute/done multiple DDLs but other tables may do nothing.
+	err := o.handleLock(info, tts, false)
+	if err != nil {
+		o.logger.Error("fail to handle the shard DDL lock", zap.Stringer("info", info), log.ShortError(err))
+		metrics.ReportDDLError(info.Task, metrics.InfoErrHandleLock)
+	}
+	return err
+}
+
 // handleOperationPut handles PUT for the shard DDL lock operations.
 func (o *Optimist) handleOperationPut(ctx context.Context, opCh <-chan optimism.Operation) {
 	for {
@@ -469,6 +491,10 @@ func (o *Optimist) handleOperationPut(ctx context.Context, opCh <-chan optimism.
 				continue
 			}
 
+			err := lock.DeleteColumnsByDDLs(op.DDLs)
+			if err != nil {
+				o.logger.Error("fail to update lock columns", zap.Error(err))
+			}
 			// in optimistic mode, we always try to mark a table as done after received the `done` status of the DDLs operation.
 			// NOTE: even all tables have done their previous DDLs operations, the lock may still not resolved,
 			// because these tables may have different schemas.
@@ -497,7 +523,7 @@ func (o *Optimist) handleOperationPut(ctx context.Context, opCh <-chan optimism.
 
 // handleLock handles a single shard DDL lock.
 func (o *Optimist) handleLock(info optimism.Info, tts []optimism.TargetTable, skipDone bool) error {
-	lockID, newDDLs, err := o.lk.TrySync(info, tts)
+	lockID, newDDLs, err := o.lk.TrySync(o.cli, info, tts)
 	var cfStage = optimism.ConflictNone
 	if info.IgnoreConflict {
 		o.logger.Warn("error occur when trying to sync for shard DDL info, this often means shard DDL conflict detected",
@@ -544,7 +570,7 @@ func (o *Optimist) handleLock(info optimism.Info, tts []optimism.TargetTable, sk
 	}
 
 	op := optimism.NewOperation(lockID, lock.Task, info.Source, info.UpSchema, info.UpTable, newDDLs, cfStage, false)
-	rev, succ, err := optimism.PutOperation(o.cli, skipDone, op)
+	rev, succ, err := optimism.PutOperation(o.cli, skipDone, op, info.ModRevision)
 	if err != nil {
 		return err
 	}
@@ -621,7 +647,7 @@ func (o *Optimist) deleteInfosOps(lock *optimism.Lock) (bool, error) {
 	}
 	// NOTE: we rely on only `task`, `downSchema`, and `downTable` used for deletion.
 	initSchema := optimism.NewInitSchema(lock.Task, lock.DownSchema, lock.DownTable, nil)
-	rev, deleted, err := optimism.DeleteInfosOperationsSchema(o.cli, infos, ops, initSchema)
+	rev, deleted, err := optimism.DeleteInfosOperationsSchemaColumn(o.cli, infos, ops, initSchema)
 	if err != nil {
 		return deleted, err
 	}

dm/master/shardddl/optimist_test.go

@@ -272,7 +272,7 @@ func (t *testOptimist) testOptimist(c *C, restart int) {
 	// mark op11 as done.
 	op11c := op11
 	op11c.Done = true
-	_, putted, err := optimism.PutOperation(etcdTestCli, false, op11c)
+	_, putted, err := optimism.PutOperation(etcdTestCli, false, op11c, 0)
 	c.Assert(err, IsNil)
 	c.Assert(putted, IsTrue)
 	c.Assert(utils.WaitSomething(backOff, waitTime, func() bool {
@@ -330,7 +330,7 @@ func (t *testOptimist) testOptimist(c *C, restart int) {
 	// mark op12 as done, the lock should be resolved.
 	op12c := op12
 	op12c.Done = true
-	_, putted, err = optimism.PutOperation(etcdTestCli, false, op12c)
+	_, putted, err = optimism.PutOperation(etcdTestCli, false, op12c, 0)
 	c.Assert(err, IsNil)
 	c.Assert(putted, IsTrue)
 	c.Assert(utils.WaitSomething(backOff, waitTime, func() bool {
@@ -491,7 +491,7 @@ func (t *testOptimist) testOptimist(c *C, restart int) {
 	// mark op21 as done.
 	op21c := op21
 	op21c.Done = true
-	_, putted, err = optimism.PutOperation(etcdTestCli, false, op21c)
+	_, putted, err = optimism.PutOperation(etcdTestCli, false, op21c, 0)
 	c.Assert(err, IsNil)
 	c.Assert(putted, IsTrue)
 	c.Assert(utils.WaitSomething(backOff, waitTime, func() bool {
@@ -511,7 +511,7 @@ func (t *testOptimist) testOptimist(c *C, restart int) {
 	// mark op23 as done.
 	op23c := op23
 	op23c.Done = true
-	_, putted, err = optimism.PutOperation(etcdTestCli, false, op23c)
+	_, putted, err = optimism.PutOperation(etcdTestCli, false, op23c, 0)
 	c.Assert(err, IsNil)
 	c.Assert(putted, IsTrue)
 	c.Assert(utils.WaitSomething(backOff, waitTime, func() bool {
@@ -570,7 +570,7 @@ func (t *testOptimist) testOptimist(c *C, restart int) {
 	// mark op31 as done.
 	op31c := op31
 	op31c.Done = true
-	_, putted, err = optimism.PutOperation(etcdTestCli, false, op31c)
+	_, putted, err = optimism.PutOperation(etcdTestCli, false, op31c, 0)
 	c.Assert(err, IsNil)
 	c.Assert(putted, IsTrue)
 	c.Assert(utils.WaitSomething(backOff, waitTime, func() bool {
@@ -623,7 +623,7 @@ func (t *testOptimist) testOptimist(c *C, restart int) {
 	// mark op33 as done, the lock should be resolved.
 	op33c := op33
 	op33c.Done = true
-	_, putted, err = optimism.PutOperation(etcdTestCli, false, op33c)
+	_, putted, err = optimism.PutOperation(etcdTestCli, false, op33c, 0)
 	c.Assert(err, IsNil)
 	c.Assert(putted, IsTrue)
 	c.Assert(utils.WaitSomething(backOff, waitTime, func() bool {
@@ -866,12 +866,12 @@ func (t *testOptimist) TestOptimistLockMultipleTarget(c *C) {
 	op11c := op12
 	op11c.Done = true
 	op11c.UpTable = i11.UpTable // overwrite `UpTable`.
-	_, putted, err := optimism.PutOperation(etcdTestCli, false, op11c)
+	_, putted, err := optimism.PutOperation(etcdTestCli, false, op11c, 0)
 	c.Assert(err, IsNil)
 	c.Assert(putted, IsTrue)
 	op12c := op12
 	op12c.Done = true
-	_, putted, err = optimism.PutOperation(etcdTestCli, false, op12c)
+	_, putted, err = optimism.PutOperation(etcdTestCli, false, op12c, 0)
 	c.Assert(err, IsNil)
 	c.Assert(putted, IsTrue)
 	c.Assert(utils.WaitSomething(backOff, waitTime, func() bool {
@@ -900,12 +900,12 @@ func (t *testOptimist) TestOptimistLockMultipleTarget(c *C) {
 	op21c := op22
 	op21c.Done = true
 	op21c.UpTable = i21.UpTable // overwrite `UpTable`.
-	_, putted, err = optimism.PutOperation(etcdTestCli, false, op21c)
+	_, putted, err = optimism.PutOperation(etcdTestCli, false, op21c, 0)
 	c.Assert(err, IsNil)
 	c.Assert(putted, IsTrue)
 	op22c := op22
 	op22c.Done = true
-	_, putted, err = optimism.PutOperation(etcdTestCli, false, op22c)
+	_, putted, err = optimism.PutOperation(etcdTestCli, false, op22c, 0)
 	c.Assert(err, IsNil)
 	c.Assert(putted, IsTrue)
 	c.Assert(utils.WaitSomething(backOff, waitTime, func() bool {
@@ -999,12 +999,12 @@ func (t *testOptimist) TestOptimistInitSchema(c *C) {
 	op11c := op12
 	op11c.Done = true
 	op11c.UpTable = i11.UpTable // overwrite `UpTable`.
-	_, putted, err := optimism.PutOperation(etcdTestCli, false, op11c)
+	_, putted, err := optimism.PutOperation(etcdTestCli, false, op11c, 0)
 	c.Assert(err, IsNil)
 	c.Assert(putted, IsTrue)
 	op12c := op12
 	op12c.Done = true
-	_, putted, err = optimism.PutOperation(etcdTestCli, false, op12c)
+	_, putted, err = optimism.PutOperation(etcdTestCli, false, op12c, 0)
 	c.Assert(err, IsNil)
 	c.Assert(putted, IsTrue)
 	c.Assert(utils.WaitSomething(backOff, waitTime, func() bool {

go.sum

@@ -171,6 +171,7 @@ github.com/coocood/rtutil v0.0.0-20190304133409-c84515f646f2 h1:NnLfQ77q0G4k2Of2
 github.com/coocood/rtutil v0.0.0-20190304133409-c84515f646f2/go.mod h1:7qG7YFnOALvsx6tKTNmQot8d7cGFXM9TidzvRFLWYwM=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
+github.com/coreos/etcd v3.3.12+incompatible h1:pAWNwdf7QiT1zfaWyqCtNZQWCLByQyA3JrSQyuYAqnQ=
 github.com/coreos/etcd v3.3.12+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk=
 github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=

loader/util.go

@@ -123,7 +123,7 @@ func getDBAndTableFromFilename(filename string) (string, string, error) {
 	}
 	fields := strings.Split(filename[:idx], ".")
 	if len(fields) != 2 && len(fields) != 3 {
-		return "", "", fmt.Errorf("%s doesn't have correct `.` seperator", filename)
+		return "", "", fmt.Errorf("%s doesn't have correct `.` separator", filename)
 	}
 	return fields[0], fields[1], nil
 }

loader/util_test.go

@@ -110,7 +110,7 @@ func (t *testUtilSuite) TestGetDBAndTableFromFilename(c *C) {
 		{"sqldb.tbl.0.sql", "sqldb", "tbl", ""},
 		{"db.tbl.sql0.sql", "db", "tbl", ""},
 		{"db.tbl.0", "", "", ".*doesn't have a `.sql` suffix.*"},
-		{"db.sql", "", "", ".*doesn't have correct `.` seperator.*"},
+		{"db.sql", "", "", ".*doesn't have correct `.` separator.*"},
 		{"db.0.sql", "db", "0", ""}, // treat `0` as the table name.
 	}
 

monitoring/dashboards/dashboard.go

@@ -84,7 +84,7 @@ func filterDashboard(str string, dashboard string, title string) string {
 		str = strings.ReplaceAll(str, fmt.Sprintf("${%s}", datasource), datasourceName)
 	}
 
-	// delete input defination
+	// delete input definition
 	if gjson.Get(str, "__inputs").Exists() {
 		str, err = sjson.Delete(str, "__inputs")
 		checkErr(err, "delete path failed")

pkg/etcdutil/etcdutil.go

@@ -108,7 +108,7 @@ func DoOpsInOneTxnWithRetry(cli *clientv3.Client, ops ...clientv3.Op) (*clientv3
 	ret, _, err := etcdDefaultTxnStrategy.Apply(tctx, etcdDefaultTxnRetryParam, func(t *tcontext.Context) (ret interface{}, err error) {
 		resp, err := cli.Txn(ctx).Then(ops...).Commit()
 		if err != nil {
-			return nil, err
+			return nil, errors.Trace(err)
 		}
 		return resp, nil
 	})