syncer: support two online-ddl pt and gh-ost on one task (#1904)

dm/config/subtask.go

@@ -131,8 +131,10 @@ type SubTaskConfig struct {
 	flagSet *flag.FlagSet
 
 	// when in sharding, multi dm-workers do one task
-	IsSharding      bool   `toml:"is-sharding" json:"is-sharding"`
-	ShardMode       string `toml:"shard-mode" json:"shard-mode"`
+	IsSharding bool   `toml:"is-sharding" json:"is-sharding"`
+	ShardMode  string `toml:"shard-mode" json:"shard-mode"`
+	OnlineDDL  bool   `toml:"online-ddl" json:"online-ddl"`
+	// deprecated
 	OnlineDDLScheme string `toml:"online-ddl-scheme" json:"online-ddl-scheme"`
 
 	// handle schema/table name mode, and only for schema/table name/pattern

dm/config/subtask_test.go

@@ -20,11 +20,11 @@ import (
 
 func (t *testConfig) TestSubTask(c *C) {
 	cfg := &SubTaskConfig{
-		Name:            "test-task",
-		IsSharding:      true,
-		ShardMode:       "optimistic",
-		SourceID:        "mysql-instance-01",
-		OnlineDDLScheme: "pt",
+		Name:       "test-task",
+		IsSharding: true,
+		ShardMode:  "optimistic",
+		SourceID:   "mysql-instance-01",
+		OnlineDDL:  true,
 		From: DBConfig{
 			Host:     "127.0.0.1",
 			Port:     3306,
@@ -70,9 +70,9 @@ func (t *testConfig) TestSubTask(c *C) {
 func (t *testConfig) TestSubTaskAdjustFail(c *C) {
 	newSubTaskConfig := func() *SubTaskConfig {
 		return &SubTaskConfig{
-			Name:            "test-task",
-			SourceID:        "mysql-instance-01",
-			OnlineDDLScheme: "pt",
+			Name:      "test-task",
+			SourceID:  "mysql-instance-01",
+			OnlineDDL: true,
 			From: DBConfig{
 				Host:     "127.0.0.1",
 				Port:     3306,

dm/config/task.go

@@ -298,6 +298,8 @@ type TaskConfig struct {
 
 	MySQLInstances []*MySQLInstance `yaml:"mysql-instances" toml:"mysql-instances" json:"mysql-instances"`
 
+	OnlineDDL bool `yaml:"online-ddl" toml:"online-ddl" json:"online-ddl"`
+	// deprecated
 	OnlineDDLScheme string `yaml:"online-ddl-scheme" toml:"online-ddl-scheme" json:"online-ddl-scheme"`
 
 	Routes         map[string]*router.TableRule   `yaml:"routes" toml:"routes" json:"routes"`
@@ -426,6 +428,9 @@ func (c *TaskConfig) adjust() error {
 
 	if c.OnlineDDLScheme != "" && c.OnlineDDLScheme != PT && c.OnlineDDLScheme != GHOST {
 		return terror.ErrConfigOnlineSchemeNotSupport.Generate(c.OnlineDDLScheme)
+	} else if c.OnlineDDLScheme == PT || c.OnlineDDLScheme == GHOST {
+		c.OnlineDDL = true
+		log.L().Warn("'online-ddl-scheme' will be deprecated soon. Recommend that use online-ddl instead of online-ddl-scheme.")
 	}
 
 	if c.TargetDB == nil {
@@ -689,7 +694,7 @@ func (c *TaskConfig) SubTaskConfigs(sources map[string]DBConfig) ([]*SubTaskConf
 		cfg := NewSubTaskConfig()
 		cfg.IsSharding = c.IsSharding
 		cfg.ShardMode = c.ShardMode
-		cfg.OnlineDDLScheme = c.OnlineDDLScheme
+		cfg.OnlineDDL = c.OnlineDDL
 		cfg.IgnoreCheckingItems = c.IgnoreCheckingItems
 		cfg.Name = c.Name
 		cfg.Mode = c.TaskMode
@@ -785,6 +790,7 @@ func FromSubTaskConfigs(stCfgs ...*SubTaskConfig) *TaskConfig {
 	c.HeartbeatReportInterval = stCfg0.HeartbeatReportInterval
 	c.CaseSensitive = stCfg0.CaseSensitive
 	c.TargetDB = &stCfg0.To // just ref
+	c.OnlineDDL = stCfg0.OnlineDDL
 	c.OnlineDDLScheme = stCfg0.OnlineDDLScheme
 	c.CleanDumpFile = stCfg0.CleanDumpFile
 	c.MySQLInstances = make([]*MySQLInstance, 0, len(stCfgs))

dm/config/task_test.go

@@ -36,7 +36,7 @@ task-mode: all
 shard-mode: "pessimistic"       
 meta-schema: "dm_meta"         
 case-sensitive: false        
-online-ddl-scheme: "gh-ost" 
+online-ddl: true
 clean-dump-file: true     
 
 target-database:
@@ -144,7 +144,7 @@ task-mode: all
 shard-mode: "pessimistic"       
 meta-schema: "dm_meta"         
 case-sensitive: false        
-online-ddl-scheme: "gh-ost" 
+online-ddl: true
 clean-dump-file: true     
 
 target-database:
@@ -540,7 +540,7 @@ func WordCount(s string) map[string]int {
 func (t *testConfig) TestGenAndFromSubTaskConfigs(c *C) {
 	var (
 		shardMode           = ShardOptimistic
-		onlineDDLScheme     = "pt"
+		onlineDDL           = true
 		name                = "from-sub-tasks"
 		taskMode            = "incremental"
 		ignoreCheckingItems = []string{VersionChecking, BinlogRowImageChecking}
@@ -645,7 +645,7 @@ func (t *testConfig) TestGenAndFromSubTaskConfigs(c *C) {
 		stCfg1 = &SubTaskConfig{
 			IsSharding:              true,
 			ShardMode:               shardMode,
-			OnlineDDLScheme:         onlineDDLScheme,
+			OnlineDDL:               onlineDDL,
 			CaseSensitive:           true,
 			Name:                    name,
 			Mode:                    taskMode,
@@ -770,7 +770,7 @@ func (t *testConfig) TestGenAndFromSubTaskConfigs(c *C) {
 				ExpressionFilters:  []string{"expr-filter-01"},
 			},
 		},
-		OnlineDDLScheme: onlineDDLScheme,
+		OnlineDDL: onlineDDL,
 		Routes: map[string]*router.TableRule{
 			"route-01": &routeRule1,
 			"route-02": &routeRule2,

syncer/ddl_test.go

@@ -417,8 +417,7 @@ func (s *testSyncerSuite) TestResolveOnlineDDL(c *C) {
 	}
 
 	for _, ca := range cases {
-		fn := onlineddl.OnlineDDLSchemes[ca.onlineType]
-		plugin, err := fn(tctx, s.cfg)
+		plugin, err := onlineddl.NewRealOnlinePlugin(tctx, s.cfg)
 		c.Assert(err, IsNil)
 		syncer := NewSyncer(s.cfg, nil)
 		syncer.onlineDDL = plugin