performance: use OPTIMISTIC transaction as the default

dm/config/task.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/pingcap/dm/pkg/log"
 	"github.com/pingcap/dm/pkg/terror"
+	"github.com/pingcap/dm/pkg/utils"
 
 	"github.com/dustin/go-humanize"
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
@@ -41,8 +42,10 @@ const (
 
 // shard DDL mode.
 const (
-	ShardPessimistic = "pessimistic"
-	ShardOptimistic  = "optimistic"
+	ShardPessimistic  = "pessimistic"
+	ShardOptimistic   = "optimistic"
+	tidbTxnMode       = "tidb_txn_mode"
+	tidbTxnOptimistic = "optimistic"
 )
 
 // default config item values
@@ -66,6 +69,15 @@ var (
 	defaultBatch                   = 100
 	defaultQueueSize               = 1024 // do not give too large default value to avoid OOM
 	defaultCheckpointFlushInterval = 30   // in seconds
+
+	// TargetDBConfig
+	defaultSessionCfg = []struct {
+		key        string
+		val        string
+		minVersion utils.TiDBVersion
+	}{
+		{tidbTxnMode, tidbTxnOptimistic, utils.TiDBVersion{3, 0, 0}},
+	}
 )
 
 // Meta represents binlog's meta pos
@@ -707,3 +719,18 @@ func checkDuplicateString(ruleNames []string) []string {
 	}
 	return dupeArray
 }
+
+// AdjustTargetDBSessionCfg adjust session cfg of TiDB
+func AdjustTargetDBSessionCfg(dbConfig *DBConfig, version utils.TiDBVersion) {
+	lowerMap := make(map[string]string, len(dbConfig.Session))
+	for k, v := range dbConfig.Session {
+		lowerMap[strings.ToLower(k)] = v
+	}
+	// all cfg in defaultSessionCfg should be lower case
+	for _, cfg := range defaultSessionCfg {
+		if _, ok := lowerMap[cfg.key]; !ok && version.Ge(cfg.minVersion) {
+			lowerMap[cfg.key] = cfg.val
+		}
+	}
+	dbConfig.Session = lowerMap
+}

dm/config/task_test.go

@@ -19,6 +19,7 @@ import (
 	"sort"
 
 	"github.com/pingcap/dm/pkg/terror"
+	"github.com/pingcap/dm/pkg/utils"
 
 	. "github.com/pingcap/check"
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
@@ -324,7 +325,10 @@ func (t *testConfig) TestGenAndFromSubTaskConfigs(c *C) {
 		heartbeatRI         = 21
 		timezone            = "Asia/Shanghai"
 		maxAllowedPacket    = 10244201
-		session             = map[string]string{
+		fromSession         = map[string]string{
+			"sql_mode": " NO_AUTO_VALUE_ON_ZERO,ANSI_QUOTES",
+		}
+		toSession = map[string]string{
 			"sql_mode": " NO_AUTO_VALUE_ON_ZERO,ANSI_QUOTES",
 		}
 		security = Security{
@@ -380,7 +384,7 @@ func (t *testConfig) TestGenAndFromSubTaskConfigs(c *C) {
 			User:             "user_from_1",
 			Password:         "123",
 			MaxAllowedPacket: &maxAllowedPacket,
-			Session:          session,
+			Session:          fromSession,
 			Security:         &security,
 			RawDBCfg:         &rawDBCfg,
 		}
@@ -390,7 +394,7 @@ func (t *testConfig) TestGenAndFromSubTaskConfigs(c *C) {
 			User:             "user_from_2",
 			Password:         "abc",
 			MaxAllowedPacket: &maxAllowedPacket,
-			Session:          session,
+			Session:          fromSession,
 			Security:         &security,
 			RawDBCfg:         &rawDBCfg,
 		}
@@ -421,7 +425,7 @@ func (t *testConfig) TestGenAndFromSubTaskConfigs(c *C) {
 				User:             "user_to",
 				Password:         "abc",
 				MaxAllowedPacket: &maxAllowedPacket,
-				Session:          session,
+				Session:          toSession,
 				Security:         &security,
 				RawDBCfg:         &rawDBCfg,
 			},
@@ -645,3 +649,37 @@ func (t *testConfig) TestMySQLInstance(c *C) {
 	c.Assert(m.VerifyAndAdjust(), IsNil)
 
 }
+
+func (t *testConfig) TestAdjustTargetDBConfig(c *C) {
+	testCases := []struct {
+		dbConfig DBConfig
+		result   DBConfig
+		version  utils.TiDBVersion
+	}{
+		{
+			DBConfig{},
+			DBConfig{Session: map[string]string{}},
+			utils.TiDBVersion{0, 0, 0},
+		},
+		{
+			DBConfig{Session: map[string]string{"ANSI_QUOTES": ""}},
+			DBConfig{Session: map[string]string{"ansi_quotes": ""}},
+			utils.TiDBVersion{2, 0, 7},
+		},
+		{
+			DBConfig{},
+			DBConfig{Session: map[string]string{tidbTxnMode: tidbTxnOptimistic}},
+			utils.TiDBVersion{3, 0, 1},
+		},
+		{
+			DBConfig{Session: map[string]string{"ANSI_QUOTES": "", tidbTxnMode: "pessimistic"}},
+			DBConfig{Session: map[string]string{"ansi_quotes": "", tidbTxnMode: "pessimistic"}},
+			utils.TiDBVersion{4, 0, 2},
+		},
+	}
+
+	for _, tc := range testCases {
+		AdjustTargetDBSessionCfg(&tc.dbConfig, tc.version)
+		c.Assert(tc.dbConfig, DeepEquals, tc.result)
+	}
+}

dm/master/server.go

@@ -1027,17 +1027,47 @@ func parseAndAdjustSourceConfig(contents []string) ([]*config.SourceConfig, erro
 			return cfgs, err
 		}
 		if err = cfg.Adjust(fromDB.DB); err != nil {
+			fromDB.Close()
 			return cfgs, err
 		}
 		if _, err = cfg.Yaml(); err != nil {
+			fromDB.Close()
 			return cfgs, err
 		}
 
+		fromDB.Close()
 		cfgs[i] = cfg
 	}
 	return cfgs, nil
 }
 
+func adjustTargetDB(ctx context.Context, dbConfig *config.DBConfig) error {
+	cfg := *dbConfig
+	if len(cfg.Password) > 0 {
+		cfg.Password = utils.DecryptOrPlaintext(cfg.Password)
+	}
+
+	toDB, err := conn.DefaultDBProvider.Apply(cfg)
+	if err != nil {
+		return err
+	}
+	defer toDB.Close()
+
+	value, err := dbutil.ShowVersion(ctx, toDB.DB)
+	if err != nil {
+		return err
+	}
+
+	version, err := utils.ToTiDBVersion(value)
+	// Do not adjust if not TiDB
+	if err == nil {
+		config.AdjustTargetDBSessionCfg(dbConfig, version)
+	} else {
+		log.L().Warn("get tidb version", log.ShortError(err))
+	}
+	return nil
+}
+
 // OperateSource will create or update an upstream source.
 func (s *Server) OperateSource(ctx context.Context, req *pb.OperateSourceRequest) (*pb.OperateSourceResponse, error) {
 	var (
@@ -1194,6 +1224,11 @@ func (s *Server) generateSubTask(ctx context.Context, task string) (*config.Task
 		return nil, nil, terror.WithClass(err, terror.ClassDMMaster)
 	}
 
+	err = adjustTargetDB(ctx, cfg.TargetDB)
+	if err != nil {
+		return nil, nil, terror.WithClass(err, terror.ClassDMMaster)
+	}
+
 	sourceCfgs, err := s.getSourceConfigs(cfg.MySQLInstances)
 	if err != nil {
 		return nil, nil, err

dm/master/server_test.go

@@ -356,6 +356,12 @@ func (t *testMaster) TestCheckTask(c *check.C) {
 	var wg sync.WaitGroup
 	ctx, cancel := context.WithCancel(context.Background())
 	server.scheduler, _ = testMockScheduler(ctx, &wg, c, sources, workers, "", t.workerClients)
+	mock := t.initVersionDB(c)
+	defer func() {
+		conn.DefaultDBProvider = &conn.DefaultDBProviderImpl{}
+	}()
+	mock.ExpectQuery("SHOW GLOBAL VARIABLES LIKE 'version'").WillReturnRows(sqlmock.NewRows([]string{"Variable_name", "Value"}).
+		AddRow("version", "5.7.25-TiDB-v4.0.2"))
 	resp, err := server.CheckTask(context.Background(), &pb.CheckTaskRequest{
 		Task: taskConfig,
 	})
@@ -373,6 +379,9 @@ func (t *testMaster) TestCheckTask(c *check.C) {
 	// simulate invalid password returned from scheduler, but config was supported plaintext mysql password, so cfg.SubTaskConfigs will success
 	ctx, cancel = context.WithCancel(context.Background())
 	server.scheduler, _ = testMockScheduler(ctx, &wg, c, sources, workers, "invalid-encrypt-password", t.workerClients)
+	mock = t.initVersionDB(c)
+	mock.ExpectQuery("SHOW GLOBAL VARIABLES LIKE 'version'").WillReturnRows(sqlmock.NewRows([]string{"Variable_name", "Value"}).
+		AddRow("version", "5.7.25-TiDB-v4.0.2"))
 	resp, err = server.CheckTask(context.Background(), &pb.CheckTaskRequest{
 		Task: taskConfig,
 	})
@@ -406,6 +415,12 @@ func (t *testMaster) TestStartTask(c *check.C) {
 	}
 	server.scheduler, _ = testMockScheduler(ctx, &wg, c, sources, workers, "",
 		makeWorkerClientsForHandle(ctrl, taskName, sources, workers, req))
+	mock := t.initVersionDB(c)
+	defer func() {
+		conn.DefaultDBProvider = &conn.DefaultDBProviderImpl{}
+	}()
+	mock.ExpectQuery("SHOW GLOBAL VARIABLES LIKE 'version'").WillReturnRows(sqlmock.NewRows([]string{"Variable_name", "Value"}).
+		AddRow("version", "5.7.25-TiDB-v4.0.2"))
 	resp, err = server.StartTask(context.Background(), req)
 	c.Assert(err, check.IsNil)
 	c.Assert(resp.Result, check.IsTrue)
@@ -420,6 +435,9 @@ func (t *testMaster) TestStartTask(c *check.C) {
 
 	// check start-task with an invalid source
 	invalidSource := "invalid-source"
+	mock = t.initVersionDB(c)
+	mock.ExpectQuery("SHOW GLOBAL VARIABLES LIKE 'version'").WillReturnRows(sqlmock.NewRows([]string{"Variable_name", "Value"}).
+		AddRow("version", "5.7.25-TiDB-v4.0.2"))
 	resp, err = server.StartTask(context.Background(), &pb.StartTaskRequest{
 		Task:    taskConfig,
 		Sources: []string{invalidSource},
@@ -438,6 +456,9 @@ func (t *testMaster) TestStartTask(c *check.C) {
 	defer func() {
 		checker.CheckSyncConfigFunc = bakCheckSyncConfigFunc
 	}()
+	mock = t.initVersionDB(c)
+	mock.ExpectQuery("SHOW GLOBAL VARIABLES LIKE 'version'").WillReturnRows(sqlmock.NewRows([]string{"Variable_name", "Value"}).
+		AddRow("version", "5.7.25-TiDB-v4.0.2"))
 	resp, err = server.StartTask(context.Background(), &pb.StartTaskRequest{
 		Task:    taskConfig,
 		Sources: sources,
@@ -448,13 +469,41 @@ func (t *testMaster) TestStartTask(c *check.C) {
 	clearSchedulerEnv(c, cancel, &wg)
 }
 
+// db use for remove data
+// verDB user for show version
 type mockDBProvider struct {
-	db *sql.DB
+	verDB *sql.DB
+	db    *sql.DB
 }
 
-// Apply will build BaseDB with DBConfig
+// return db if verDB was closed
 func (d *mockDBProvider) Apply(config config.DBConfig) (*conn.BaseDB, error) {
-	return conn.NewBaseDB(d.db, func() {}), nil
+	if err := d.verDB.Ping(); err != nil {
+		return conn.NewBaseDB(d.db, func() {}), nil
+	}
+	return conn.NewBaseDB(d.verDB, func() {}), nil
+}
+
+func (t *testMaster) initVersionDB(c *check.C) sqlmock.Sqlmock {
+	db, mock, err := sqlmock.New()
+	c.Assert(err, check.IsNil)
+	if mdbp, ok := conn.DefaultDBProvider.(*mockDBProvider); ok {
+		mdbp.verDB = db
+	} else {
+		conn.DefaultDBProvider = &mockDBProvider{verDB: db}
+	}
+	return mock
+}
+
+func (t *testMaster) initMockDB(c *check.C) sqlmock.Sqlmock {
+	db, mock, err := sqlmock.New()
+	c.Assert(err, check.IsNil)
+	if mdbp, ok := conn.DefaultDBProvider.(*mockDBProvider); ok {
+		mdbp.db = db
+	} else {
+		conn.DefaultDBProvider = &mockDBProvider{db: db}
+	}
+	return mock
 }
 
 func (t *testMaster) TestStartTaskWithRemoveMeta(c *check.C) {
@@ -503,12 +552,13 @@ func (t *testMaster) TestStartTaskWithRemoveMeta(c *check.C) {
 	c.Assert(server.pessimist.Start(ctx, etcdTestCli), check.IsNil)
 	c.Assert(server.optimist.Start(ctx, etcdTestCli), check.IsNil)
 
-	db, mock, err := sqlmock.New()
-	c.Assert(err, check.IsNil)
-	conn.DefaultDBProvider = &mockDBProvider{db: db}
+	verMock := t.initVersionDB(c)
 	defer func() {
 		conn.DefaultDBProvider = &conn.DefaultDBProviderImpl{}
 	}()
+	verMock.ExpectQuery("SHOW GLOBAL VARIABLES LIKE 'version'").WillReturnRows(sqlmock.NewRows([]string{"Variable_name", "Value"}).
+		AddRow("version", "5.7.25-TiDB-v4.0.2"))
+	mock := t.initMockDB(c)
 	mock.ExpectBegin()
 	mock.ExpectExec(fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`", cfg.MetaSchema, cputil.LoaderCheckpoint(cfg.Name))).WillReturnResult(sqlmock.NewResult(1, 1))
 	mock.ExpectExec(fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`", cfg.MetaSchema, cputil.SyncerCheckpoint(cfg.Name))).WillReturnResult(sqlmock.NewResult(1, 1))
@@ -523,6 +573,9 @@ func (t *testMaster) TestStartTaskWithRemoveMeta(c *check.C) {
 		defer wg.Done()
 		time.Sleep(10 * time.Microsecond)
 		// start another same task at the same time, should get err
+		verMock2 := t.initVersionDB(c)
+		verMock2.ExpectQuery("SHOW GLOBAL VARIABLES LIKE 'version'").WillReturnRows(sqlmock.NewRows([]string{"Variable_name", "Value"}).
+			AddRow("version", "5.7.25-TiDB-v4.0.2"))
 		resp1, err1 := server.StartTask(context.Background(), req)
 		c.Assert(err1, check.IsNil)
 		c.Assert(resp1.Result, check.IsFalse)
@@ -593,9 +646,10 @@ func (t *testMaster) TestStartTaskWithRemoveMeta(c *check.C) {
 	err = server.optimist.Start(ctx, etcdTestCli)
 	c.Assert(err, check.IsNil)
 
-	db, mock, err = sqlmock.New()
-	c.Assert(err, check.IsNil)
-	conn.DefaultDBProvider = &mockDBProvider{db: db}
+	verMock = t.initVersionDB(c)
+	verMock.ExpectQuery("SHOW GLOBAL VARIABLES LIKE 'version'").WillReturnRows(sqlmock.NewRows([]string{"Variable_name", "Value"}).
+		AddRow("version", "5.7.25-TiDB-v4.0.2"))
+	mock = t.initMockDB(c)
 	mock.ExpectBegin()
 	mock.ExpectExec(fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`", cfg.MetaSchema, cputil.LoaderCheckpoint(cfg.Name))).WillReturnResult(sqlmock.NewResult(1, 1))
 	mock.ExpectExec(fmt.Sprintf("DROP TABLE IF EXISTS `%s`.`%s`", cfg.MetaSchema, cputil.SyncerCheckpoint(cfg.Name))).WillReturnResult(sqlmock.NewResult(1, 1))
@@ -610,6 +664,9 @@ func (t *testMaster) TestStartTaskWithRemoveMeta(c *check.C) {
 		defer wg.Done()
 		time.Sleep(10 * time.Microsecond)
 		// start another same task at the same time, should get err
+		vermock2 := t.initVersionDB(c)
+		vermock2.ExpectQuery("SHOW GLOBAL VARIABLES LIKE 'version'").WillReturnRows(sqlmock.NewRows([]string{"Variable_name", "Value"}).
+			AddRow("version", "5.7.25-TiDB-v4.0.2"))
 		resp1, err1 := server.StartTask(context.Background(), req)
 		c.Assert(err1, check.IsNil)
 		c.Assert(resp1.Result, check.IsFalse)
@@ -692,6 +749,12 @@ func (t *testMaster) TestOperateTask(c *check.C) {
 	sourceResps := []*pb.CommonWorkerResponse{{Result: true, Source: sources[0]}, {Result: true, Source: sources[1]}}
 	server.scheduler, _ = testMockScheduler(ctx, &wg, c, sources, workers, "",
 		makeWorkerClientsForHandle(ctrl, taskName, sources, workers, startReq, pauseReq, resumeReq, stopReq1, stopReq2))
+	mock := t.initVersionDB(c)
+	defer func() {
+		conn.DefaultDBProvider = &conn.DefaultDBProviderImpl{}
+	}()
+	mock.ExpectQuery("SHOW GLOBAL VARIABLES LIKE 'version'").WillReturnRows(sqlmock.NewRows([]string{"Variable_name", "Value"}).
+		AddRow("version", "5.7.25-TiDB-v4.0.2"))
 	stResp, err := server.StartTask(context.Background(), startReq)
 	c.Assert(err, check.IsNil)
 	c.Assert(stResp.Result, check.IsTrue)
@@ -1328,6 +1391,12 @@ func (t *testMaster) TestGetTaskCfg(c *check.C) {
 		makeWorkerClientsForHandle(ctrl, taskName, sources, workers, req))
 
 	// start task
+	mock := t.initVersionDB(c)
+	defer func() {
+		conn.DefaultDBProvider = &conn.DefaultDBProviderImpl{}
+	}()
+	mock.ExpectQuery("SHOW GLOBAL VARIABLES LIKE 'version'").WillReturnRows(sqlmock.NewRows([]string{"Variable_name", "Value"}).
+		AddRow("version", "5.7.25-TiDB-v4.0.2"))
 	resp, err := server.StartTask(context.Background(), req)
 	c.Assert(err, check.IsNil)
 	c.Assert(resp.Result, check.IsTrue)

pkg/schema/tracker.go

@@ -63,9 +63,13 @@ func NewTracker(sessionCfg map[string]string, tidbConn *conn.BaseConn) (*Tracker
 
 	if len(sessionCfg) == 0 {
 		sessionCfg = make(map[string]string)
-		var ignoredColumn interface{}
-		for _, k := range sessionVars {
-			rows, err2 := tidbConn.QuerySQL(tcontext.Background(), fmt.Sprintf("show variables like '%s'", k))
+	}
+	// get variables if user doesn't specify
+	// all cfg in sessionVars should be lower case
+	for _, k := range sessionVars {
+		if _, ok := sessionCfg[k]; !ok {
+			var ignoredColumn interface{}
+			rows, err2 := tidbConn.QuerySQL(tcontext.Background(), fmt.Sprintf("SHOW VARIABLES LIKE '%s'", k))
 			if err2 != nil {
 				return nil, err2
 			}