session: add a batch commit session variable for the large transaction

config/config.go

@@ -155,6 +155,7 @@ type Performance struct {
 	StatsLease          string  `toml:"stats-lease" json:"stats-lease"`
 	RunAutoAnalyze      bool    `toml:"run-auto-analyze" json:"run-auto-analyze"`
 	StmtCountLimit      uint    `toml:"stmt-count-limit" json:"stmt-count-limit"`
+	BatchCommit         bool    `toml:"batch-commit" json:"batch-commit"`
 	FeedbackProbability float64 `toml:"feedback-probability" json:"feedback-probability"`
 	QueryFeedbackLimit  uint    `toml:"query-feedback-limit" json:"query-feedback-limit"`
 	PseudoEstimateRatio float64 `toml:"pseudo-estimate-ratio" json:"pseudo-estimate-ratio"`
@@ -292,6 +293,7 @@ var defaultConf = Config{
 		StatsLease:          "3s",
 		RunAutoAnalyze:      true,
 		StmtCountLimit:      5000,
+		BatchCommit:         false,
 		FeedbackProbability: 0.05,
 		QueryFeedbackLimit:  1024,
 		PseudoEstimateRatio: 0.8,

config/config.toml.example

@@ -124,6 +124,8 @@ metrics-interval = 15
 max-procs = 0
 # StmtCountLimit limits the max count of statement inside a transaction.
 stmt-count-limit = 5000
+# If true, the transaction will be committed when it reaches stmt-count-limit and starts a new transaction.
+batch-commit = false
 
 # Set keep alive option for tcp connection.
 tcp-keep-alive = true

session/session_test.go

@@ -1925,13 +1925,13 @@ func (s *testSessionSuite) TestStatementErrorInTransaction(c *C) {
 	tk.MustQuery("select * from test where a = 1 and b = 11").Check(testkit.Rows())
 }
 
-func (s *testSessionSuite) TestStatementCountLimit(c *C) {
+func (s *testSessionSuite) TestStatementCountLimitAndBatchCommit(c *C) {
 	tk := testkit.NewTestKitWithInit(c, s.store)
 	tk.MustExec("create table stmt_count_limit (id int)")
-	saved := config.GetGlobalConfig().Performance.StmtCountLimit
+	savedLimit := config.GetGlobalConfig().Performance.StmtCountLimit
 	config.GetGlobalConfig().Performance.StmtCountLimit = 3
 	defer func() {
-		config.GetGlobalConfig().Performance.StmtCountLimit = saved
+		config.GetGlobalConfig().Performance.StmtCountLimit = savedLimit
 	}()
 	tk.MustExec("begin")
 	tk.MustExec("insert into stmt_count_limit values (1)")
@@ -1946,6 +1946,17 @@ func (s *testSessionSuite) TestStatementCountLimit(c *C) {
 	tk.MustExec("insert into stmt_count_limit values (3)")
 	_, err = tk.Exec("insert into stmt_count_limit values (4)")
 	c.Assert(err, NotNil)
+
+	savedBatchFlag := config.GetGlobalConfig().Performance.BatchCommit
+	config.GetGlobalConfig().Performance.BatchCommit = true
+	defer func() {
+		config.GetGlobalConfig().Performance.BatchCommit = savedBatchFlag
+	}()
+	tk.MustExec("begin")
+	tk.MustExec("insert into stmt_count_limit values (1)")
+	tk.MustExec("insert into stmt_count_limit values (2)")
+	tk.MustExec("insert into stmt_count_limit values (3)")
+	tk.MustExec("commit")
 }
 
 func (s *testSessionSuite) TestCastTimeToDate(c *C) {

session/tidb.go

@@ -171,11 +171,15 @@ func runStmt(ctx context.Context, sctx sessionctx.Context, s sqlexec.Statement)
 		// If the user insert, insert, insert ... but never commit, TiDB would OOM.
 		// So we limit the statement count in a transaction here.
 		history := GetHistory(sctx)
-		if history.Count() > int(config.GetGlobalConfig().Performance.StmtCountLimit) {
-			err1 := se.RollbackTxn(ctx)
-			terror.Log(errors.Trace(err1))
-			return rs, errors.Errorf("statement count %d exceeds the transaction limitation, autocommit = %t",
-				history.Count(), sctx.GetSessionVars().IsAutocommit())
+		perfConfig := config.GetGlobalConfig().Performance
+		if history.Count() > int(perfConfig.StmtCountLimit) {
+			if !perfConfig.BatchCommit {
+				err1 := se.RollbackTxn(ctx)
+				terror.Log(errors.Trace(err1))
+				return rs, errors.Errorf("statement count %d exceeds the transaction limitation, autocommit = %t",
+					history.Count(), sctx.GetSessionVars().IsAutocommit())
+			}
+			err = se.NewTxn()
 		}
 	}
 	return rs, errors.Trace(err)