operate-schema: flush schema info and sync to master in optimistic mode (#1506)

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 immediately")
+	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,10 @@ 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("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))
+	} 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 +539,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

@@ -368,6 +368,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.