operate-schema: flush schema info and sync to master in optimistic mode

dm/ctl/master/operate_schema.go

@@ -27,12 +27,14 @@ import (
 // NewOperateSchemaCmd creates a OperateSchema command.
 func NewOperateSchemaCmd() *cobra.Command {
 	cmd := &cobra.Command{
-		Use:   "operate-schema <operate-type> <-s source ...> <task-name | task-file> <-d database> <-t table> [schema-file]",
+		Use:   "operate-schema <operate-type> <-s source ...> <task-name | task-file> <-d database> <-t table> [schema-file] [--flush] [--sync]",
 		Short: "`get`/`set`/`remove` the schema for an upstream table.",
 		RunE:  operateSchemaCmd,
 	}
 	cmd.Flags().StringP("database", "d", "", "database name of the table")
 	cmd.Flags().StringP("table", "t", "", "table name")
+	cmd.Flags().Bool("flush", false, "flush the table info and checkpoint imediately")
+	cmd.Flags().Bool("sync", false, "sync the table info to master to resolve shard ddl lock, only for optimistic mode now")
 	return cmd
 }
 
@@ -109,6 +111,21 @@ func operateSchemaCmd(cmd *cobra.Command, _ []string) (err error) {
 		return
 	}
 
+	flush, err := cmd.Flags().GetBool("flush")
+	if err != nil {
+		return
+	}
+	if flush && op != pb.SchemaOp_SetSchema {
+		err = errors.New("--flush flag is only used to set schema")
+	}
+	sync, err := cmd.Flags().GetBool("sync")
+	if err != nil {
+		return
+	}
+	if sync && op != pb.SchemaOp_SetSchema {
+		err = errors.New("--sync flag is only used to set schema")
+	}
+
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -123,6 +140,8 @@ func operateSchemaCmd(cmd *cobra.Command, _ []string) (err error) {
 			Database: database,
 			Table:    table,
 			Schema:   string(schemaContent),
+			Flush:    flush,
+			Sync:     sync,
 		},
 		&resp,
 	)

dm/master/server.go

@@ -1737,6 +1737,8 @@ func (s *Server) OperateSchema(ctx context.Context, req *pb.OperateSchemaRequest
 					Database: req.Database,
 					Table:    req.Table,
 					Schema:   req.Schema,
+					Flush:    req.Flush,
+					Sync:     req.Sync,
 				},
 			}
 

dm/master/shardddl/optimist.go

@@ -499,7 +499,9 @@ func (o *Optimist) handleOperationPut(ctx context.Context, opCh <-chan optimism.
 func (o *Optimist) handleLock(info optimism.Info, tts []optimism.TargetTable, skipDone bool) error {
 	lockID, newDDLs, err := o.lk.TrySync(info, tts)
 	var cfStage = optimism.ConflictNone
-	if err != nil {
+	if info.IgnoreConflict {
+		o.logger.Warn("handle lock in ignore conflict mode", zap.String("lock", lockID), zap.Stringer("info", info))
+	} else if err != nil {
 		cfStage = optimism.ConflictDetected // we treat any errors returned from `TrySync` as conflict detected now.
 		o.logger.Warn("error occur when trying to sync for shard DDL info, this often means shard DDL conflict detected",
 			zap.String("lock", lockID), zap.Stringer("info", info), zap.Bool("is deleted", info.IsDeleted), log.ShortError(err))
@@ -536,6 +538,10 @@ func (o *Optimist) handleLock(info optimism.Info, tts []optimism.TargetTable, sk
 		return nil
 	}
 
+	if info.IgnoreConflict {
+		return nil
+	}
+
 	op := optimism.NewOperation(lockID, lock.Task, info.Source, info.UpSchema, info.UpTable, newDDLs, cfStage, false)
 	rev, succ, err := optimism.PutOperation(o.cli, skipDone, op)
 	if err != nil {

dm/proto/dmmaster.proto

@@ -366,6 +366,8 @@ message OperateSchemaRequest {
     string database = 4; // database name
     string table = 5; // table name
     string schema = 6; // schema content, a `CREATE TABLE` statement
+    bool flush = 7; // flush table info and checkpoint
+    bool sync = 8; // sync the table info to master
 }
 
 message OperateSchemaResponse {

dm/proto/dmworker.proto

@@ -323,6 +323,8 @@ message OperateWorkerSchemaRequest {
     string database = 4; // database name
     string table = 5; // table name
     string schema = 6; // schema content, a `CREATE TABLE` statement
+    bool flush = 7; // flush table info and checkpoint
+    bool sync = 8; // sync the table info to master
 }
 
 // copied `TaskMeta` from release-1.0 branch.

pkg/shardddl/optimism/info.go

@@ -53,6 +53,9 @@ type Info struct {
 
 	// only set it when get/watch from etcd
 	Version int64 `json:"-"`
+
+	// use to resolve conflict
+	IgnoreConflict bool `json:"ignore-conflict"`
 }
 
 // NewInfo creates a new Info instance.

pkg/shardddl/optimism/info_test.go

@@ -77,7 +77,7 @@ func (t *testForEtcd) TestInfoJSON(c *C) {
 
 	j, err := i1.toJSON()
 	c.Assert(err, IsNil)
-	c.Assert(j, Equals, `{"task":"test","source":"mysql-replica-1","up-schema":"db-1","up-table":"tbl-1","down-schema":"db","down-table":"tbl","ddls":["ALTER TABLE tbl ADD COLUMN c1 INT","ALTER TABLE tbl ADD COLUMN c2 INT"],"table-info-before":null,"table-info-after":null}`)
+	c.Assert(j, Equals, `{"task":"test","source":"mysql-replica-1","up-schema":"db-1","up-table":"tbl-1","down-schema":"db","down-table":"tbl","ddls":["ALTER TABLE tbl ADD COLUMN c1 INT","ALTER TABLE tbl ADD COLUMN c2 INT"],"table-info-before":null,"table-info-after":null,"ignore-conflict":false}`)
 	c.Assert(j, Equals, i1.String())
 
 	i2, err := infoFromJSON(j)

pkg/shardddl/optimism/keeper.go

@@ -50,7 +50,7 @@ func (lk *LockKeeper) TrySync(info Info, tts []TargetTable) (string, []string, e
 		l = lk.locks[lockID]
 	}
 
-	newDDLs, err := l.TrySync(info.Source, info.UpSchema, info.UpTable, info.DDLs, info.TableInfosAfter, tts, info.Version)
+	newDDLs, err := l.TrySync(info, tts)
 	return lockID, newDDLs, err
 }
 

pkg/shardddl/optimism/lock.go

@@ -91,8 +91,16 @@ func NewLock(ID, task, downSchema, downTable string, ti *model.TableInfo, tts []
 // TODO: but both of these modes are difficult to be implemented in DM-worker now, try to do that later.
 // for non-intrusive, a broadcast mechanism needed to notify conflict tables after the conflict has resolved, or even a block mechanism needed.
 // for intrusive, a DML prune or transform mechanism needed for two different schemas (before and after the conflict resolved).
-func (l *Lock) TrySync(callerSource, callerSchema, callerTable string,
-	ddls []string, newTIs []*model.TableInfo, tts []TargetTable, infoVersion int64) (newDDLs []string, err error) {
+func (l *Lock) TrySync(info Info, tts []TargetTable) (newDDLs []string, err error) {
+	var (
+		callerSource   = info.Source
+		callerSchema   = info.UpSchema
+		callerTable    = info.UpTable
+		ddls           = info.DDLs
+		newTIs         = info.TableInfosAfter
+		infoVersion    = info.Version
+		ignoreConflict = info.IgnoreConflict
+	)
 	l.mu.Lock()
 	defer func() {
 		if len(newDDLs) > 0 {
@@ -157,18 +165,23 @@ func (l *Lock) TrySync(callerSource, callerSchema, callerTable string,
 	oldJoined := l.joined
 
 	lastTableInfo := schemacmp.Encode(newTIs[len(newTIs)-1])
+
+	defer func() {
+		// only update table info if no error or ignore conflict
+		if ignoreConflict || err == nil {
+			log.L().Info("update table info", zap.String("lock", l.ID), zap.String("source", callerSource), zap.String("schema", callerSchema), zap.String("table", callerTable),
+				zap.Stringer("from", l.tables[callerSource][callerSchema][callerTable]), zap.Stringer("to", lastTableInfo), zap.Strings("ddls", ddls))
+			l.tables[callerSource][callerSchema][callerTable] = lastTableInfo
+		}
+	}()
+
 	lastJoined, err := joinTable(lastTableInfo)
 	if err != nil {
 		return emptyDDLs, err
 	}
 
 	defer func() {
-		// only update table info and joined info if no error
 		if err == nil {
-			// update table info and joined info base on the last new table info
-			log.L().Info("update table info", zap.String("lock", l.ID), zap.String("source", callerSource), zap.String("schema", callerSchema), zap.String("table", callerTable),
-				zap.Stringer("from", l.tables[callerSource][callerSchema][callerTable]), zap.Stringer("to", lastTableInfo), zap.Strings("ddls", ddls))
-			l.tables[callerSource][callerSchema][callerTable] = lastTableInfo
 			// update the current joined table info, it should be logged in `if cmp != 0` block below.
 			l.joined = lastJoined
 		}