openapi(dm): add more task config field

dm/dm/config/task_converters.go

@@ -167,11 +167,18 @@ func OpenAPITaskToSubTaskConfigs(task *openapi.Task, toDBCfg *DBConfig, sourceCf
 		// set shard config
 		if task.ShardMode != nil {
 			subTaskCfg.IsSharding = true
-			mode := *task.ShardMode
-			subTaskCfg.ShardMode = string(mode)
+			subTaskCfg.ShardMode = string(*task.ShardMode)
 		} else {
 			subTaskCfg.IsSharding = false
 		}
+		// set timezone
+		if task.TimeZone != nil {
+			subTaskCfg.Timezone = *task.TimeZone
+		}
+		// set collation_compatible
+		if task.CollationCompatible != nil {
+			subTaskCfg.CollationCompatible = string(*task.CollationCompatible)
+		}
 		// set online ddl plugin config
 		subTaskCfg.OnlineDDL = task.EnhanceOnlineSchemaChange
 		// set case sensitive from source
@@ -206,6 +213,12 @@ func OpenAPITaskToSubTaskConfigs(task *openapi.Task, toDBCfg *DBConfig, sourceCf
 			if incrCfg.ReplBatch != nil {
 				subTaskCfg.SyncerConfig.Batch = *incrCfg.ReplBatch
 			}
+			if incrCfg.Compact != nil {
+				subTaskCfg.SyncerConfig.Compact = *incrCfg.Compact
+			}
+			if incrCfg.MultipleRows != nil {
+				subTaskCfg.SyncerConfig.MultipleRows = *incrCfg.MultipleRows
+			}
 		}
 		// set route,blockAllowList,filter config
 		doCnt := len(tableMigrateRuleMap[sourceCfg.SourceName])
@@ -422,8 +435,10 @@ func SubTaskConfigsToOpenAPITask(subTaskConfigMap map[string]map[string]SubTaskC
 			ImportThreads: &oneSubtaskConfig.LoaderConfig.PoolSize,
 		}
 		taskSourceConfig.IncrMigrateConf = &openapi.TaskIncrMigrateConf{
-			ReplBatch:   &oneSubtaskConfig.SyncerConfig.Batch,
-			ReplThreads: &oneSubtaskConfig.SyncerConfig.WorkerCount,
+			ReplBatch:    &oneSubtaskConfig.SyncerConfig.Batch,
+			Compact:      &oneSubtaskConfig.SyncerConfig.Compact,
+			ReplThreads:  &oneSubtaskConfig.SyncerConfig.WorkerCount,
+			MultipleRows: &oneSubtaskConfig.SyncerConfig.MultipleRows,
 		}
 		// set filter rules
 		filterRuleMap := openapi.Task_BinlogFilterRule{}
@@ -494,6 +509,9 @@ func SubTaskConfigsToOpenAPITask(subTaskConfigMap map[string]map[string]SubTaskC
 		if len(filterMap) > 0 {
 			task.BinlogFilterRule = &filterRuleMap
 		}
+		task.TimeZone = &oneSubtaskConfig.Timezone
+		collationCompatible := openapi.TaskCollationCompatible(oneSubtaskConfig.CollationCompatible)
+		task.CollationCompatible = &collationCompatible
 		task.TableMigrateRule = tableMigrateRuleList
 		taskList = append(taskList, task)
 	}

dm/dm/config/task_converters_test.go

@@ -75,6 +75,10 @@ func testNoShardTaskToSubTaskConfigs(c *check.C) {
 	c.Assert(subTaskConfig.ShardMode, check.Equals, "")
 	// check online schema change
 	c.Assert(subTaskConfig.OnlineDDL, check.Equals, true)
+	// check timezone
+	c.Assert(subTaskConfig.Timezone, check.Equals, *task.TimeZone)
+	// check collation
+	c.Assert(subTaskConfig.CollationCompatible, check.Equals, string(*task.CollationCompatible))
 	// check case sensitive
 	c.Assert(subTaskConfig.CaseSensitive, check.Equals, sourceCfg1.CaseSensitive)
 	// check from
@@ -86,8 +90,11 @@ func testNoShardTaskToSubTaskConfigs(c *check.C) {
 	c.Assert(subTaskConfig.LoaderConfig.Dir, check.Equals, fmt.Sprintf(
 		"%s.%s", *task.SourceConfig.FullMigrateConf.DataDir, task.Name))
 	c.Assert(subTaskConfig.LoaderConfig.PoolSize, check.Equals, *task.SourceConfig.FullMigrateConf.ImportThreads)
-	c.Assert(subTaskConfig.SyncerConfig.WorkerCount, check.Equals, *task.SourceConfig.IncrMigrateConf.ReplThreads)
+	c.Assert(subTaskConfig.SyncerConfig.Compact, check.Equals, *task.SourceConfig.IncrMigrateConf.Compact)
 	c.Assert(subTaskConfig.SyncerConfig.Batch, check.Equals, *task.SourceConfig.IncrMigrateConf.ReplBatch)
+	c.Assert(subTaskConfig.SyncerConfig.WorkerCount, check.Equals, *task.SourceConfig.IncrMigrateConf.ReplThreads)
+	c.Assert(subTaskConfig.SyncerConfig.MultipleRows, check.Equals, *task.SourceConfig.IncrMigrateConf.MultipleRows)
+
 	// check route
 	c.Assert(subTaskConfig.RouteRules, check.HasLen, 1)
 	rule := subTaskConfig.RouteRules[0]
@@ -280,6 +287,21 @@ func testNoShardSubTaskConfigsToOpenAPITask(c *check.C) {
 	c.Assert(taskList, check.HasLen, 1)
 	newTask := taskList[0]
 
+	t1, _ := newTask.ToJSON()
+	t2, _ := task.ToJSON()
+	c.Assert(string(t1), check.Equals, string(t2))
+
+	// println("~~~~~~t1~~~~~~~", task.CollationCompatible, task.TimeZone)
+	// println("~~~~~~t2~~~~~~~", newTask.CollationCompatible, newTask.TimeZone)
+
+	// c.Assert(*task.CollationCompatible, check.Equals, *newTask.CollationCompatible)
+	// c.Assert(*task.TimeZone, check.Equals, *newTask.TimeZone)
+	// c.Assert(task.SourceConfig.IncrMigrateConf, check.DeepEquals, newTask.SourceConfig.IncrMigrateConf)
+
+	// newTask.TimeZone = task.TimeZone
+	// newTask.CollationCompatible = task.CollationCompatible
+	// newTask.SourceConfig.IncrMigrateConf = task.SourceConfig.IncrMigrateConf
+
 	c.Assert(task, check.DeepEquals, newTask)
 }
 

dm/openapi/fixtures/task.go

@@ -25,14 +25,21 @@ var (
 		"enhance_online_schema_change": true,
 		"meta_schema": "dm_meta",
 		"name": "test",
+		"time_zone": "Asia/Shanghai",
+		"collation_compatible": "loose",
 		"on_duplicate": "error",
 		"source_config": {
 		  "full_migrate_conf": {
 			"data_dir": "./exported_data",
 			"export_threads": 4,
 			"import_threads": 16
 		  },
-		  "incr_migrate_conf": { "repl_batch": 200, "repl_threads": 32 },
+		  "incr_migrate_conf": {
+			"repl_threads": 16,
+			"repl_batch": 100,
+			"multiple_rows": false,
+			"compact": false
+		  },
 		  "source_conf": [{ "source_name": "mysql-replica-01" }]
 		},
 		"table_migrate_rule": [