.*: add compactor for syncer

chaos/cases/conf/task-optimistic.yaml

@@ -20,20 +20,24 @@ mysql-instances:
     black-white-list:  "instance"
     mydumper-thread: 4
     loader-thread: 16
-    syncer-thread: 16
+    syncer-config-name: "global"
   -
     source-id: "replica-02"
     black-white-list:  "instance"
     mydumper-thread: 4
     loader-thread: 16
-    syncer-thread: 16
+    syncer-config-name: "global"
   -
     source-id: "replica-03"
     black-white-list:  "instance"
     mydumper-thread: 4
     loader-thread: 16
-    syncer-thread: 16
+    syncer-config-name: "global"
 
 black-white-list:
   instance:
     do-dbs: ["db_optimistic"]
+
+syncers:
+  global:
+    compact: true

chaos/cases/conf/task-pessimistic.yaml

@@ -20,20 +20,24 @@ mysql-instances:
     black-white-list:  "instance"
     mydumper-thread: 4
     loader-thread: 16
-    syncer-thread: 16
+    syncer-config-name: "global"
   -
     source-id: "replica-02"
     black-white-list:  "instance"
     mydumper-thread: 4
     loader-thread: 16
-    syncer-thread: 16
+    syncer-config-name: "global"
   -
     source-id: "replica-03"
     black-white-list:  "instance"
     mydumper-thread: 4
     loader-thread: 16
-    syncer-thread: 16
+    syncer-config-name: "global"
 
 black-white-list:
   instance:
     do-dbs: ["db_pessimistic"]
+
+syncers:
+  global:
+    compact: true

chaos/cases/conf/task-single.yaml

@@ -14,8 +14,12 @@ mysql-instances:
     black-white-list:  "instance"
     mydumper-thread: 4
     loader-thread: 16
-    syncer-thread: 16
+    syncer-config-name: "global"
 
 black-white-list:
   instance:
     do-dbs: ["db_single"]
+
+syncers:
+  global:
+    compact: true

chaos/cases/task.go

@@ -79,7 +79,12 @@ func newTask(ctx context.Context, cli pb.MasterClient, taskFile string, schema s
 		sourceConns = make([]*dbConn, 0, len(taskCfg.MySQLInstances))
 		res         = make(results, 0, len(taskCfg.MySQLInstances))
 	)
-	for i := range taskCfg.MySQLInstances { // only use necessary part of sources.
+	for i, m := range taskCfg.MySQLInstances { // only use necessary part of sources.
+		// reset Syncer, otherwise will report ERROR 20017
+		if len(m.SyncerConfigName) > 0 && m.Syncer != nil {
+			m.Syncer = nil
+		}
+
 		cfg := sourcesCfg[i]
 		db, err2 := conn.DefaultDBProvider.Apply(cfg)
 		if err2 != nil {
@@ -270,7 +275,7 @@ func (t *task) genFullData() error {
 
 // createTask does `start-task` operation.
 func (t *task) createTask() error {
-	t.logger.Info("starting the task")
+	t.logger.Info("starting the task", zap.String("task cfg", t.taskCfg.String()))
 	resp, err := t.cli.StartTask(t.ctx, &pb.StartTaskRequest{
 		Task: t.taskCfg.String(),
 	})

dm/config/task.go

@@ -237,7 +237,8 @@ type SyncerConfig struct {
 	Batch       int    `yaml:"batch" toml:"batch" json:"batch"`
 	QueueSize   int    `yaml:"queue-size" toml:"queue-size" json:"queue-size"`
 	// checkpoint flush interval in seconds.
-	CheckpointFlushInterval int `yaml:"checkpoint-flush-interval" toml:"checkpoint-flush-interval" json:"checkpoint-flush-interval"`
+	CheckpointFlushInterval int  `yaml:"checkpoint-flush-interval" toml:"checkpoint-flush-interval" json:"checkpoint-flush-interval"`
+	Compact                 bool `yaml:"compact" toml:"compact" json:"compact"`
 
 	// deprecated
 	MaxRetry int `yaml:"max-retry" toml:"max-retry" json:"max-retry"`
@@ -857,34 +858,76 @@ func NewMySQLInstancesForDowngrade(mysqlInstances []*MySQLInstance) []*MySQLInst
 	return mysqlInstancesForDowngrade
 }
 
+// SyncerConfigForDowngrade is the base configuration for syncer in v2.0.
+// This config is used for downgrade(config export) from a higher dmctl version.
+// When we add any new config item into SyncerConfig, we should update it also.
+type SyncerConfigForDowngrade struct {
+	MetaFile                string `yaml:"meta-file"`
+	WorkerCount             int    `yaml:"worker-count"`
+	Batch                   int    `yaml:"batch"`
+	QueueSize               int    `yaml:"queue-size"`
+	CheckpointFlushInterval int    `yaml:"checkpoint-flush-interval"`
+	MaxRetry                int    `yaml:"max-retry"`
+	AutoFixGTID             bool   `yaml:"auto-fix-gtid"`
+	EnableGTID              bool   `yaml:"enable-gtid"`
+	DisableCausality        bool   `yaml:"disable-detect"`
+	SafeMode                bool   `yaml:"safe-mode"`
+	EnableANSIQuotes        bool   `yaml:"enable-ansi-quotes"`
+
+	Compact bool `yaml:"compact,omitempty"`
+}
+
+// NewSyncerConfigsForDowngrade converts SyncerConfig to SyncerConfigForDowngrade.
+func NewSyncerConfigsForDowngrade(syncerConfigs map[string]*SyncerConfig) map[string]*SyncerConfigForDowngrade {
+	syncerConfigsForDowngrade := make(map[string]*SyncerConfigForDowngrade, len(syncerConfigs))
+	for configName, syncerConfig := range syncerConfigs {
+		newSyncerConfig := &SyncerConfigForDowngrade{
+			MetaFile:                syncerConfig.MetaFile,
+			WorkerCount:             syncerConfig.WorkerCount,
+			Batch:                   syncerConfig.Batch,
+			QueueSize:               syncerConfig.QueueSize,
+			CheckpointFlushInterval: syncerConfig.CheckpointFlushInterval,
+			MaxRetry:                syncerConfig.MaxRetry,
+			AutoFixGTID:             syncerConfig.AutoFixGTID,
+			EnableGTID:              syncerConfig.EnableGTID,
+			DisableCausality:        syncerConfig.DisableCausality,
+			SafeMode:                syncerConfig.SafeMode,
+			EnableANSIQuotes:        syncerConfig.EnableANSIQuotes,
+			Compact:                 syncerConfig.Compact,
+		}
+		syncerConfigsForDowngrade[configName] = newSyncerConfig
+	}
+	return syncerConfigsForDowngrade
+}
+
 // TaskConfigForDowngrade is the base configuration for task in v2.0.
 // This config is used for downgrade(config export) from a higher dmctl version.
 // When we add any new config item into SourceConfig, we should update it also.
 type TaskConfigForDowngrade struct {
-	Name                    string                         `yaml:"name"`
-	TaskMode                string                         `yaml:"task-mode"`
-	IsSharding              bool                           `yaml:"is-sharding"`
-	ShardMode               string                         `yaml:"shard-mode"`
-	IgnoreCheckingItems     []string                       `yaml:"ignore-checking-items"`
-	MetaSchema              string                         `yaml:"meta-schema"`
-	EnableHeartbeat         bool                           `yaml:"enable-heartbeat"`
-	HeartbeatUpdateInterval int                            `yaml:"heartbeat-update-interval"`
-	HeartbeatReportInterval int                            `yaml:"heartbeat-report-interval"`
-	Timezone                string                         `yaml:"timezone"`
-	CaseSensitive           bool                           `yaml:"case-sensitive"`
-	TargetDB                *DBConfig                      `yaml:"target-database"`
-	OnlineDDLScheme         string                         `yaml:"online-ddl-scheme"`
-	Routes                  map[string]*router.TableRule   `yaml:"routes"`
-	Filters                 map[string]*bf.BinlogEventRule `yaml:"filters"`
-	ColumnMappings          map[string]*column.Rule        `yaml:"column-mappings"`
-	BWList                  map[string]*filter.Rules       `yaml:"black-white-list"`
-	BAList                  map[string]*filter.Rules       `yaml:"block-allow-list"`
-	Mydumpers               map[string]*MydumperConfig     `yaml:"mydumpers"`
-	Loaders                 map[string]*LoaderConfig       `yaml:"loaders"`
-	Syncers                 map[string]*SyncerConfig       `yaml:"syncers"`
-	CleanDumpFile           bool                           `yaml:"clean-dump-file"`
-	EnableANSIQuotes        bool                           `yaml:"ansi-quotes"`
-	RemoveMeta              bool                           `yaml:"remove-meta"`
+	Name                    string                               `yaml:"name"`
+	TaskMode                string                               `yaml:"task-mode"`
+	IsSharding              bool                                 `yaml:"is-sharding"`
+	ShardMode               string                               `yaml:"shard-mode"`
+	IgnoreCheckingItems     []string                             `yaml:"ignore-checking-items"`
+	MetaSchema              string                               `yaml:"meta-schema"`
+	EnableHeartbeat         bool                                 `yaml:"enable-heartbeat"`
+	HeartbeatUpdateInterval int                                  `yaml:"heartbeat-update-interval"`
+	HeartbeatReportInterval int                                  `yaml:"heartbeat-report-interval"`
+	Timezone                string                               `yaml:"timezone"`
+	CaseSensitive           bool                                 `yaml:"case-sensitive"`
+	TargetDB                *DBConfig                            `yaml:"target-database"`
+	OnlineDDLScheme         string                               `yaml:"online-ddl-scheme"`
+	Routes                  map[string]*router.TableRule         `yaml:"routes"`
+	Filters                 map[string]*bf.BinlogEventRule       `yaml:"filters"`
+	ColumnMappings          map[string]*column.Rule              `yaml:"column-mappings"`
+	BWList                  map[string]*filter.Rules             `yaml:"black-white-list"`
+	BAList                  map[string]*filter.Rules             `yaml:"block-allow-list"`
+	Mydumpers               map[string]*MydumperConfig           `yaml:"mydumpers"`
+	Loaders                 map[string]*LoaderConfig             `yaml:"loaders"`
+	Syncers                 map[string]*SyncerConfigForDowngrade `yaml:"syncers"`
+	CleanDumpFile           bool                                 `yaml:"clean-dump-file"`
+	EnableANSIQuotes        bool                                 `yaml:"ansi-quotes"`
+	RemoveMeta              bool                                 `yaml:"remove-meta"`
 	// new config item
 	MySQLInstances   []*MySQLInstanceForDowngrade `yaml:"mysql-instances"`
 	ExprFilter       map[string]*ExpressionFilter `yaml:"expression-filter,omitempty"`
@@ -916,7 +959,7 @@ func NewTaskConfigForDowngrade(taskConfig *TaskConfig) *TaskConfigForDowngrade {
 		BAList:                  taskConfig.BAList,
 		Mydumpers:               taskConfig.Mydumpers,
 		Loaders:                 taskConfig.Loaders,
-		Syncers:                 taskConfig.Syncers,
+		Syncers:                 NewSyncerConfigsForDowngrade(taskConfig.Syncers),
 		CleanDumpFile:           taskConfig.CleanDumpFile,
 		EnableANSIQuotes:        taskConfig.EnableANSIQuotes,
 		RemoveMeta:              taskConfig.RemoveMeta,

dm/config/task_test.go

@@ -1050,6 +1050,21 @@ func cloneValues(dest, src reflect.Value) {
 		srcType = srcType.Elem()
 	}
 
+	if destType.Kind() == reflect.Map {
+		destMap := reflect.MakeMap(destType)
+		for _, k := range src.MapKeys() {
+			if src.MapIndex(k).Type().Kind() == reflect.Ptr {
+				newVal := reflect.New(destType.Elem().Elem())
+				cloneValues(newVal, src.MapIndex(k))
+				destMap.SetMapIndex(k, newVal)
+			} else {
+				cloneValues(destMap.MapIndex(k).Addr(), src.MapIndex(k).Addr())
+			}
+		}
+		dest.Set(destMap)
+		return
+	}
+
 	if destType.Kind() == reflect.Slice {
 		slice := reflect.MakeSlice(destType, src.Len(), src.Cap())
 		for i := 0; i < src.Len(); i++ {