session: add a global/session variable 'tidb_disable_txn_auto_retry' (#6877)

Author: coocood

session/session.go

@@ -333,6 +333,16 @@ func (s *session) doCommitWithRetry(ctx context.Context) error {
 	}
 	err := s.doCommit(ctx)
 	if err != nil {
+		commitRetryLimit := globalCommitRetryLimit
+		if s.sessionVars.DisableTxnAutoRetry && !s.sessionVars.InRestrictedSQL {
+			// Do not retry non-autocommit transactions.
+			// For autocommit single statement transactions, the history count is always 1.
+			// For explicit transactions, the statement count is more than 1.
+			history := GetHistory(s)
+			if history.Count() > 1 {
+				commitRetryLimit = 0
+			}
+		}
 		// Don't retry in BatchInsert mode. As a counter-example, insert into t1 select * from t2,
 		// BatchInsert already commit the first batch 1000 rows, then it commit 1000-2000 and retry the statement,
 		// Finally t1 will have more data than t2, with no errors return to user!
@@ -1269,7 +1279,8 @@ const loadCommonGlobalVarsSQL = "select HIGH_PRIORITY * from mysql.global_variab
 	variable.TiDBIndexLookupJoinConcurrency + quoteCommaQuote +
 	variable.TiDBIndexSerialScanConcurrency + quoteCommaQuote +
 	variable.TiDBHashJoinConcurrency + quoteCommaQuote +
-	variable.TiDBDistSQLScanConcurrency + "')"
+	variable.TiDBDistSQLScanConcurrency + quoteCommaQuote +
+	variable.TiDBDisableTxnAutoRetry + "')"
 
 // loadCommonGlobalVariablesIfNeeded loads and applies commonly used global variables for the session.
 func (s *session) loadCommonGlobalVariablesIfNeeded() error {

session/session_test.go

@@ -2008,3 +2008,40 @@ func (s *testSessionSuite) TestDBUserNameLength(c *C) {
 	tk.MustExec(`grant all privileges on test.* to 'abcddfjakldfjaldddds'@'%' identified by ''`)
 	tk.MustExec(`grant all privileges on test.t to 'abcddfjakldfjaldddds'@'%' identified by ''`)
 }
+
+func (s *testSessionSuite) TestDisableTxnAutoRetry(c *C) {
+	tk1 := testkit.NewTestKitWithInit(c, s.store)
+	tk2 := testkit.NewTestKitWithInit(c, s.store)
+	tk1.MustExec("create table no_retry (id int)")
+	tk1.MustExec("insert into no_retry values (1)")
+	tk1.MustExec("set @@tidb_disable_txn_auto_retry = 1")
+
+	tk1.MustExec("begin")
+	tk1.MustExec("update no_retry set id = 2")
+
+	tk2.MustExec("begin")
+	tk2.MustExec("update no_retry set id = 3")
+	tk2.MustExec("commit")
+
+	// No auto retry because tidb_disable_txn_auto_retry is set to 1.
+	_, err := tk1.Se.Execute(context.Background(), "commit")
+	c.Assert(err, NotNil)
+
+	// session 1 starts a transaction early.
+	// execute a select statement to clear retry history.
+	tk1.MustExec("select 1")
+	tk1.Se.NewTxn()
+	// session 2 update the value.
+	tk2.MustExec("update no_retry set id = 4")
+	// Autocommit update will retry, so it would not fail.
+	tk1.MustExec("update no_retry set id = 5")
+
+	// RestrictedSQL should retry.
+	tk1.Se.GetSessionVars().InRestrictedSQL = true
+	tk1.MustExec("begin")
+
+	tk2.MustExec("update no_retry set id = 6")
+
+	tk1.MustExec("update no_retry set id = 7")
+	tk1.MustExec("commit")
+}

session/tidb.go

@@ -108,7 +108,7 @@ var (
 	statsLease = 3 * time.Second
 
 	// The maximum number of retries to recover from retryable errors.
-	commitRetryLimit uint = 10
+	globalCommitRetryLimit uint = 10
 )
 
 // SetSchemaLease changes the default schema lease time for DDL.
@@ -129,7 +129,7 @@ func SetStatsLease(lease time.Duration) {
 // reinstated by retry, including network interruption, transaction conflicts, and
 // so on.
 func SetCommitRetryLimit(limit uint) {
-	commitRetryLimit = limit
+	globalCommitRetryLimit = limit
 }
 
 // Parse parses a query string to raw ast.StmtNode.

sessionctx/variable/session.go

@@ -327,6 +327,8 @@ type SessionVars struct {
 	EnableStreaming bool
 
 	writeStmtBufs WriteStmtBufs
+
+	DisableTxnAutoRetry bool
 }
 
 // NewSessionVars creates a session vars object.
@@ -362,6 +364,7 @@ func NewSessionVars() *SessionVars {
 		MemQuotaIndexLookupJoin:    DefTiDBMemQuotaIndexLookupJoin,
 		MemQuotaNestedLoopApply:    DefTiDBMemQuotaNestedLoopApply,
 		OptimizerSelectivityLevel:  DefTiDBOptimizerSelectivityLevel,
+		DisableTxnAutoRetry:        DefTiDBDisableTxnAutoRetry,
 	}
 	var enableStreaming string
 	if config.GetGlobalConfig().EnableStreaming {
@@ -562,6 +565,8 @@ func (s *SessionVars) SetSystemVar(name string, val string) error {
 		s.EnableStreaming = TiDBOptOn(val)
 	case TiDBOptimizerSelectivityLevel:
 		s.OptimizerSelectivityLevel = tidbOptPositiveInt32(val, DefTiDBOptimizerSelectivityLevel)
+	case TiDBDisableTxnAutoRetry:
+		s.DisableTxnAutoRetry = TiDBOptOn(val)
 	}
 	s.systems[name] = val
 	return nil

sessionctx/variable/sysvar.go

@@ -637,6 +637,7 @@ var defaultSysVars = []*SysVar{
 	{ScopeSession, TxnIsolationOneShot, ""},
 	{ScopeGlobal | ScopeSession, TiDBHashJoinConcurrency, strconv.Itoa(DefTiDBHashJoinConcurrency)},
 	{ScopeSession, TiDBOptimizerSelectivityLevel, strconv.Itoa(DefTiDBOptimizerSelectivityLevel)},
+	{ScopeGlobal | ScopeSession, TiDBDisableTxnAutoRetry, boolToIntStr(DefTiDBDisableTxnAutoRetry)},
 	/* The following variable is defined as session scope but is actually server scope. */
 	{ScopeSession, TiDBGeneralLog, strconv.Itoa(DefTiDBGeneralLog)},
 	{ScopeSession, TiDBConfig, ""},

sessionctx/variable/tidb_vars.go

@@ -150,6 +150,9 @@ const (
 	// tidb_hash_join_concurrency is used for hash join executor.
 	// The hash join outer executor starts multiple concurrent join workers to probe the hash table.
 	TiDBHashJoinConcurrency = "tidb_hash_join_concurrency"
+
+	// tidb_disable_txn_auto_retry disables transaction auto retry.
+	TiDBDisableTxnAutoRetry = "tidb_disable_txn_auto_retry"
 )
 
 // Default TiDB system variable values.
@@ -181,6 +184,7 @@ const (
 	DefTiDBGeneralLog                = 0
 	DefTiDBHashJoinConcurrency       = 5
 	DefTiDBOptimizerSelectivityLevel = 0
+	DefTiDBDisableTxnAutoRetry       = false
 )
 
 // Process global variables.