*: add system variable to turn on/off the large transaction feature

kv/kv.go

@@ -53,6 +53,8 @@ const (
 	SnapshotTS
 	// Set replica read
 	ReplicaRead
+	// LargeTxn option means this txn is a large txn.
+	LargeTxn
 )
 
 // Priority value for transaction priority.

session/pessimistic_test.go

@@ -45,10 +45,13 @@ type testPessimisticSuite struct {
 	dom       *domain.Domain
 }
 
-func (s *testPessimisticSuite) SetUpSuite(c *C) {
-	testleak.BeforeTest()
+func init() {
 	// Set it to 300ms for testing lock resolve.
 	tikv.PessimisticLockTTL = 300
+}
+
+func (s *testPessimisticSuite) SetUpSuite(c *C) {
+	testleak.BeforeTest()
 	s.cluster = mocktikv.NewCluster()
 	mocktikv.BootstrapWithSingleStore(s.cluster)
 	s.mvccStore = mocktikv.MustNewMVCCStore()

session/session.go

@@ -437,6 +437,9 @@ func (s *session) doCommit(ctx context.Context) error {
 	// Set this option for 2 phase commit to validate schema lease.
 	s.txn.SetOption(kv.SchemaChecker, domain.NewSchemaChecker(domain.GetDomain(s), s.sessionVars.TxnCtx.SchemaVersion, tableIDs))
 
+	if s.sessionVars.EnableLargeTxn && s.txn.Size() > int(s.sessionVars.LargeTxnSize) {
+		s.txn.SetOption(kv.LargeTxn, true)
+	}
 	return s.txn.Commit(sessionctx.SetCommitCtx(ctx, s))
 }
 

session/session_test.go

@@ -2882,3 +2882,26 @@ func (s *testSessionSuite) TestStmtHints(c *C) {
 	c.Assert(tk.Se.GetSessionVars().StmtCtx.GetWarnings(), HasLen, 1)
 	c.Assert(tk.Se.GetSessionVars().GetReplicaRead(), Equals, kv.ReplicaReadFollower)
 }
+
+func (s *testSessionSuite) TestEnableLargeTxn(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("create table enable_large_txn (id int)")
+	tk.MustExec("set @@tidb_enable_large_txn = 0")
+	tk.MustQuery("select @@tidb_enable_large_txn").Check(testkit.Rows("0"))
+
+	tk.MustExec("set @@tidb_enable_large_txn = 1")
+	tk.MustQuery("select @@tidb_enable_large_txn").Check(testkit.Rows("1"))
+
+	tk.MustExec("set @@tidb_large_txn_size = 6666")
+	tk.MustQuery("select @@tidb_large_txn_size").Check(testkit.Rows("6666"))
+
+	tk.MustExec("set @@tidb_large_txn_size = 0")
+	tk.MustQuery("select @@tidb_large_txn_size").Check(testkit.Rows("0"))
+
+	var codeRun bool
+	ctx := context.WithValue(context.Background(), "checkLargeTxnEnable", &codeRun)
+	_, err := tk.Se.Execute(ctx, "insert into enable_large_txn values (1),(2),(3)")
+	c.Assert(err, IsNil)
+	c.Assert(codeRun, IsTrue)
+}

sessionctx/variable/session.go

@@ -366,6 +366,12 @@ type SessionVars struct {
 	// TODO: remove this after tidb-server configuration "enable-streaming' removed.
 	EnableStreaming bool
 
+	// EnableLargeTxn enables the large transaction feature.
+	EnableLargeTxn bool
+
+	// LargeTxnSize is the size threshold for a transaction to be considered as large transaction.
+	LargeTxnSize int64
+
 	// EnableChunkRPC indicates whether the coprocessor request can use chunk API.
 	EnableChunkRPC bool
 
@@ -529,6 +535,8 @@ func NewSessionVars() *SessionVars {
 		UsePlanBaselines:            DefTiDBUsePlanBaselines,
 		isolationReadEngines:        map[kv.StoreType]struct{}{kv.TiKV: {}, kv.TiFlash: {}},
 		LockWaitTimeout:             DefInnodbLockWaitTimeout * 1000,
+		EnableLargeTxn:              DefTiDBEnableLargeTxn,
+		LargeTxnSize:                DefTiDBLargeTxnSize,
 	}
 	vars.Concurrency = Concurrency{
 		IndexLookupConcurrency:     DefIndexLookupConcurrency,
@@ -920,6 +928,10 @@ func (s *SessionVars) SetSystemVar(name string, val string) error {
 		s.DisableTxnAutoRetry = TiDBOptOn(val)
 	case TiDBEnableStreaming:
 		s.EnableStreaming = TiDBOptOn(val)
+	case TiDBEnableLargeTxn:
+		s.EnableLargeTxn = TiDBOptOn(val)
+	case TiDBLargeTxnSize:
+		s.LargeTxnSize = tidbOptInt64(val, DefTiDBLargeTxnSize)
 	case TiDBEnableChunkRPC:
 		s.EnableChunkRPC = TiDBOptOn(val)
 	case TiDBEnableCascadesPlanner:

sessionctx/variable/sysvar.go

@@ -702,6 +702,8 @@ var defaultSysVars = []*SysVar{
 	{ScopeGlobal | ScopeSession, TiDBEnableVectorizedExpression, BoolToIntStr(DefEnableVectorizedExpression)},
 	{ScopeGlobal | ScopeSession, TiDBEnableFastAnalyze, BoolToIntStr(DefTiDBUseFastAnalyze)},
 	{ScopeGlobal | ScopeSession, TiDBSkipIsolationLevelCheck, BoolToIntStr(DefTiDBSkipIsolationLevelCheck)},
+	{ScopeGlobal | ScopeSession, TiDBEnableLargeTxn, BoolToIntStr(DefTiDBEnableLargeTxn)},
+	{ScopeGlobal | ScopeSession, TiDBLargeTxnSize, strconv.Itoa(DefTiDBLargeTxnSize)},
 	/* The following variable is defined as session scope but is actually server scope. */
 	{ScopeSession, TiDBGeneralLog, strconv.Itoa(DefTiDBGeneralLog)},
 	{ScopeSession, TiDBSlowLogThreshold, strconv.Itoa(logutil.DefaultSlowThreshold)},

sessionctx/variable/tidb_vars.go

@@ -140,6 +140,12 @@ const (
 	// off: always disable table partition.
 	TiDBEnableTablePartition = "tidb_enable_table_partition"
 
+	// tidb_enable_large_txn enables TiDB to use the large transaction feature.
+	TiDBEnableLargeTxn = "tidb_enable_large_txn"
+
+	// tidb_large_txn_size controls the size threshold for a transaction to use the large transaction feature.
+	TiDBLargeTxnSize = "tidb_large_txn_size"
+
 	// TiDBCheckMb4ValueInUTF8 is used to control whether to enable the check wrong utf8 value.
 	TiDBCheckMb4ValueInUTF8 = "tidb_check_mb4_value_in_utf8"
 
@@ -414,6 +420,8 @@ const (
 	DefTiDBAllowRemoveAutoInc          = false
 	DefTiDBUsePlanBaselines            = true
 	DefInnodbLockWaitTimeout           = 50 // 50s
+	DefTiDBEnableLargeTxn              = false
+	DefTiDBLargeTxnSize                = 4 << 20 // 4M
 )
 
 // Process global variables.

sessionctx/variable/varsutil_test.go

@@ -82,6 +82,8 @@ func (s *testVarsutilSuite) TestNewSessionVars(c *C) {
 	c.Assert(vars.AllowWriteRowID, Equals, DefOptWriteRowID)
 	c.Assert(vars.TiDBOptJoinReorderThreshold, Equals, DefTiDBOptJoinReorderThreshold)
 	c.Assert(vars.EnableFastAnalyze, Equals, DefTiDBUseFastAnalyze)
+	c.Assert(vars.EnableLargeTxn, Equals, DefTiDBEnableLargeTxn)
+	c.Assert(vars.LargeTxnSize, Equals, int64(DefTiDBLargeTxnSize))
 
 	assertFieldsGreaterThanZero(c, reflect.ValueOf(vars.Concurrency))
 	assertFieldsGreaterThanZero(c, reflect.ValueOf(vars.MemQuota))

store/tikv/2pc.go

@@ -489,6 +489,10 @@ func (c *twoPhaseCommitter) buildPrewriteRequest(batch batchKeys, txnSize uint64
 			isPessimisticLock[i] = true
 		}
 	}
+	var minCommitTS uint64
+	if c.txn.IsLargeTxn() {
+		minCommitTS = c.startTS + 1
+	}
 	req := &pb.PrewriteRequest{
 		Mutations:         mutations,
 		PrimaryLock:       c.primary(),
@@ -497,11 +501,12 @@ func (c *twoPhaseCommitter) buildPrewriteRequest(batch batchKeys, txnSize uint64
 		IsPessimisticLock: isPessimisticLock,
 		ForUpdateTs:       c.forUpdateTS,
 		TxnSize:           txnSize,
+		MinCommitTs:       minCommitTS,
 	}
 	return tikvrpc.NewRequest(tikvrpc.CmdPrewrite, req, pb.Context{Priority: c.priority, SyncLog: c.syncLog})
 }
 
-func (actionPrewrite) handleSingleBatch(c *twoPhaseCommitter, bo *Backoffer, batch batchKeys) error {
+func (action actionPrewrite) handleSingleBatch(c *twoPhaseCommitter, bo *Backoffer, batch batchKeys) error {
 	txnSize := uint64(c.regionTxnSize[batch.region.id])
 	// When we retry because of a region miss, we don't know the transaction size. We set the transaction size here
 	// to MaxUint64 to avoid unexpected "resolve lock lite".
@@ -510,6 +515,14 @@ func (actionPrewrite) handleSingleBatch(c *twoPhaseCommitter, bo *Backoffer, bat
 	}
 
 	req := c.buildPrewriteRequest(batch, txnSize)
+	if x := bo.ctx.Value("checkLargeTxnEnable"); x != nil {
+		codeRun := x.(*bool)
+		*codeRun = true
+		if req.Prewrite().MinCommitTs == 0 {
+			return errors.New("minCommitTS should greater then 0 in the large txn")
+		}
+	}
+
 	for {
 		resp, err := c.store.SendReq(bo, req, batch.region, readTimeoutShort)
 		if err != nil {

store/tikv/lock_test.go

@@ -391,6 +391,12 @@ func (s *testLockSuite) prewriteTxnWithTTL(c *C, txn *tikvTxn, ttl uint64) {
 	c.Assert(err, IsNil)
 }
 
+func mustNewTwoPhaseCommitter(c *C, txn *tikvTxn) *twoPhaseCommitter {
+	committer, err := newTwoPhaseCommitterWithInit(txn, 0)
+	c.Assert(err, IsNil)
+	return committer
+}
+
 func (s *testLockSuite) mustGetLock(c *C, key []byte) *Lock {
 	ver, err := s.store.CurrentVersion()
 	c.Assert(err, IsNil)
@@ -427,9 +433,12 @@ func (s *testLockSuite) TestLockTTL(c *C) {
 	c.Assert(err, IsNil)
 	txn.Set(kv.Key("key"), []byte("value"))
 	time.Sleep(time.Millisecond)
-	s.prewriteTxnWithTTL(c, txn.(*tikvTxn), 1000)
+	bo := NewBackoffer(context.Background(), PrewriteMaxBackoff)
+	committer := mustNewTwoPhaseCommitter(c, txn.(*tikvTxn))
+	c.Assert(committer.prewriteKeys(bo, committer.keys), IsNil)
 	l := s.mustGetLock(c, []byte("key"))
 	c.Assert(l.TTL >= defaultLockTTL, IsTrue)
+	committer.cleanupKeys(bo, committer.keys)
 
 	// Huge txn has a greater TTL.
 	txn, err = s.store.Begin()
@@ -440,9 +449,11 @@ func (s *testLockSuite) TestLockTTL(c *C) {
 		k, v := randKV(1024, 1024)
 		txn.Set(kv.Key(k), []byte(v))
 	}
-	s.prewriteTxn(c, txn.(*tikvTxn))
+	committer = mustNewTwoPhaseCommitter(c, txn.(*tikvTxn))
+	c.Assert(committer.prewriteKeys(bo, committer.keys), IsNil)
 	l = s.mustGetLock(c, []byte("key"))
 	s.ttlEquals(c, l.TTL, uint64(ttlFactor*2)+uint64(time.Since(start)/time.Millisecond))
+	committer.cleanupKeys(bo, committer.keys)
 
 	// Txn with long read time.
 	start = time.Now()

store/tikv/txn.go

@@ -259,6 +259,10 @@ func (txn *tikvTxn) DelOption(opt kv.Option) {
 	txn.us.DelOption(opt)
 }
 
+func (txn *tikvTxn) IsLargeTxn() bool {
+	return txn.us.GetOption(kv.LargeTxn) != nil
+}
+
 func (txn *tikvTxn) IsPessimistic() bool {
 	return txn.us.GetOption(kv.Pessimistic) != nil
 }