swap ColumnMapFrom, ColumnMapTo for a reversed job #1053

Author: ffffwh

api/handler/v2/job.go

@@ -11,21 +11,16 @@ import (
 	"strings"
 	"time"
 
-	mysql "github.com/actiontech/dtle/driver/mysql"
-
+	nomadApi "github.com/hashicorp/nomad/api"
 	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/labstack/echo/v4"
 
-	"github.com/actiontech/dtle/driver/kafka"
-
+	"github.com/actiontech/dtle/api/handler"
+	"github.com/actiontech/dtle/api/models"
 	"github.com/actiontech/dtle/driver/common"
-
+	"github.com/actiontech/dtle/driver/kafka"
+	mysql "github.com/actiontech/dtle/driver/mysql"
 	"github.com/actiontech/dtle/g"
-
-	"github.com/actiontech/dtle/api/models"
-
-	"github.com/actiontech/dtle/api/handler"
-	nomadApi "github.com/hashicorp/nomad/api"
-	"github.com/labstack/echo/v4"
 )
 
 // @Id MigrationJobListV2
@@ -1955,6 +1950,11 @@ func ReverseJobV2(c echo.Context, filterJobType DtleJobType) error {
 			return c.JSON(http.StatusInternalServerError, models.BuildBaseResp(err))
 		}
 
+		if originalJob.BasicTaskProfile.Configuration.SrcConfig.MysqlSrcTaskConfig.TwoWaySync {
+			return c.JSON(http.StatusConflict, models.BuildBaseResp(fmt.Errorf(
+				"job_id=%v; job can't be reversed with TwoWaySync = true", reqParam.JobId)))
+		}
+
 		reverseJobParam := new(models.CreateOrUpdateMysqlToMysqlJobParamV2)
 		reverseJobParam.JobId = fmt.Sprintf("%s-%s", "reverse", consulJobItem.JobId)
 		reverseJobParam.TaskStepName = mysql.JobIncrCopy
@@ -1982,6 +1982,15 @@ func ReverseJobV2(c echo.Context, filterJobType DtleJobType) error {
 				TwoWaySyncGtid:      originalJob.BasicTaskProfile.Configuration.SrcConfig.MysqlSrcTaskConfig.TwoWaySyncGtid,
 			},
 		}
+		for _, dbItem := range reverseJobParam.SrcTask.ReplicateDoDb {
+			for _, tbItem := range dbItem.Tables {
+				if len(tbItem.ColumnMapFrom) > 0 && len(tbItem.ColumnMapTo) == 0 {
+					return c.JSON(http.StatusConflict, models.BuildBaseResp(fmt.Errorf(
+						"job_id=%v; job can't be reversed with ColumnMapFrom not matching ColumnMapTo", reqParam.JobId)))
+				}
+				tbItem.ColumnMapFrom, tbItem.ColumnMapTo = tbItem.ColumnMapTo, tbItem.ColumnMapFrom
+			}
+		}
 		reverseJobParam.DestTask = &models.DestTaskConfig{
 			TaskName:         common.TaskTypeDest,
 			ConnectionConfig: &originalJob.BasicTaskProfile.ConnectionInfo.SrcDataBase,

driver/mysql/applier.go

@@ -304,9 +304,23 @@ func (a *Applier) Run() {
 			Subject:  a.subject + "_dtrev",
 			StateDir: a.stateDir,
 		}
-		var cfg2 = *a.mysqlContext
+		// cfg2 will be a deepcopy of a.mysqlContext
+		cfg2, err := a.storeManager.GetConfig(a.subject)
+		if err != nil {
+			a.onError(common.TaskStateDead, errors.Wrap(err, "GetConfig"))
+			return
+		}
 		cfg2.SrcConnectionConfig = a.mysqlContext.DestConnectionConfig
 		cfg2.DestConnectionConfig = a.mysqlContext.SrcConnectionConfig
+		for _, dbItem := range cfg2.ReplicateDoDb {
+			for _, tbItem := range dbItem.Tables {
+				if len(tbItem.ColumnMapFrom) > 0 && len(tbItem.ColumnMapTo) == 0 {
+					a.onError(common.TaskStateDead, errors.Wrap(err, "GetConfig"))
+					return
+				}
+				tbItem.ColumnMapFrom, tbItem.ColumnMapTo = tbItem.ColumnMapTo, tbItem.ColumnMapFrom
+			}
+		}
 
 		if strings.ToLower(a.mysqlContext.TwoWaySyncGtid) == "auto" {
 			cfg2.AutoGtid = true
@@ -318,7 +332,7 @@ func (a *Applier) Run() {
 		cfg2.TwoWaySync = false
 		cfg2.TwoWaySyncGtid = ""
 
-		a.revExtractor, err = NewExtractor(execCtx2, &cfg2, a.logger, a.storeManager, a.waitCh, a.ctx)
+		a.revExtractor, err = NewExtractor(execCtx2, cfg2, a.logger, a.storeManager, a.waitCh, a.ctx)
 		if err != nil {
 			a.onError(common.TaskStateDead, errors.Wrap(err, "reversed Extractor"))
 			return