performance: use OPTIMISTIC transaction as the default

dm/config/task.go

@@ -41,8 +41,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 +68,11 @@ var (
 	defaultBatch                   = 100
 	defaultQueueSize               = 1024 // do not give too large default value to avoid OOM
 	defaultCheckpointFlushInterval = 30   // in seconds
+
+	// TargetDBConfig
+	defaultSessionCfg = map[string]string{
+		tidbTxnMode: tidbTxnOptimistic,
+	}
 )
 
 // Meta represents binlog's meta pos
@@ -411,6 +418,7 @@ func (c *TaskConfig) adjust() error {
 	if c.TargetDB == nil {
 		return terror.ErrConfigNeedTargetDB.Generate()
 	}
+	adjustTargetDB(c.TargetDB)
 
 	if len(c.MySQLInstances) == 0 {
 		return terror.ErrConfigMySQLInstsAtLeastOne.Generate()
@@ -707,3 +715,14 @@ func checkDuplicateString(ruleNames []string) []string {
 	}
 	return dupeArray
 }
+
+func adjustTargetDB(dbConfig *DBConfig) {
+	if dbConfig.Session == nil {
+		dbConfig.Session = make(map[string]string, len(defaultSessionCfg))
+	}
+	for k, v := range defaultSessionCfg {
+		if _, ok := dbConfig.Session[k]; !ok {
+			dbConfig.Session[k] = v
+		}
+	}
+}

dm/config/task_test.go

@@ -324,7 +324,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 +383,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 +393,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 +424,7 @@ func (t *testConfig) TestGenAndFromSubTaskConfigs(c *C) {
 				User:             "user_to",
 				Password:         "abc",
 				MaxAllowedPacket: &maxAllowedPacket,
-				Session:          session,
+				Session:          toSession,
 				Security:         &security,
 				RawDBCfg:         &rawDBCfg,
 			},
@@ -576,6 +579,8 @@ func (t *testConfig) TestGenAndFromSubTaskConfigs(c *C) {
 	stCfgs[0].EnableANSIQuotes = stCfg1.EnableANSIQuotes
 	stCfgs[1].EnableANSIQuotes = stCfg2.EnableANSIQuotes
 	c.Assert(stCfgs[0].String(), Equals, stCfg1.String())
+	// subtask session cfg also changed because we ref DBConfig when merge from subtask config
+	stCfg2.To.Session[tidbTxnMode] = tidbTxnOptimistic
 	c.Assert(stCfgs[1].String(), Equals, stCfg2.String())
 }
 
@@ -645,3 +650,21 @@ func (t *testConfig) TestMySQLInstance(c *C) {
 	c.Assert(m.VerifyAndAdjust(), IsNil)
 
 }
+
+func (t *testConfig) TestAdjustSessionCfg(c *C) {
+	sessionCfg := map[string]string{tidbTxnMode: tidbTxnOptimistic}
+	dbCfg := &DBConfig{}
+	adjustTargetDB(dbCfg)
+	c.Assert(dbCfg.Session, DeepEquals, sessionCfg)
+
+	sessionCfg["sql_mode"] = "ANSI_QUOTES"
+	dbCfg.Session["sql_mode"] = "ANSI_QUOTES"
+	adjustTargetDB(dbCfg)
+	c.Assert(dbCfg.Session, DeepEquals, sessionCfg)
+
+	tidbTxnPessimistic := "pessimistic"
+	sessionCfg[tidbTxnMode] = tidbTxnPessimistic
+	dbCfg.Session[tidbTxnMode] = tidbTxnPessimistic
+	adjustTargetDB(dbCfg)
+	c.Assert(dbCfg.Session, DeepEquals, sessionCfg)
+}

pkg/schema/tracker.go

@@ -63,9 +63,12 @@ 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
+	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
 			}

syncer/checkpoint_test.go

@@ -70,7 +70,10 @@ func (s *testCheckpointSuite) SetUpSuite(c *C) {
 	log.SetLevel(zapcore.ErrorLevel)
 	var (
 		err                   error
-		defaultTestSessionCfg = map[string]string{"sql_mode": "ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"}
+		defaultTestSessionCfg = map[string]string{
+			"sql_mode":             "ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION",
+			"tidb_skip_utf8_check": "0",
+		}
 	)
 
 	s.tracker, err = schema.NewTracker(defaultTestSessionCfg, nil)

syncer/syncer_test.go

@@ -59,7 +59,10 @@ import (
 var _ = Suite(&testSyncerSuite{})
 
 var (
-	defaultTestSessionCfg = map[string]string{"sql_mode": "ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"}
+	defaultTestSessionCfg = map[string]string{
+		"sql_mode":             "ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION",
+		"tidb_skip_utf8_check": "0",
+	}
 )
 
 func TestSuite(t *testing.T) {