session: move more session vars to stmt context for retrying

session/session.go

@@ -97,10 +97,9 @@ var (
 )
 
 type stmtRecord struct {
-	stmtID  uint32
-	st      sqlexec.Statement
-	stmtCtx *stmtctx.StatementContext
-	params  []interface{}
+	stmtID uint32
+	st     sqlexec.Statement
+	params []interface{}
 }
 
 // StmtHistory holds all histories of statements in a txn.
@@ -111,10 +110,9 @@ type StmtHistory struct {
 // Add appends a stmt to history list.
 func (h *StmtHistory) Add(stmtID uint32, st sqlexec.Statement, stmtCtx *stmtctx.StatementContext, params ...interface{}) {
 	s := &stmtRecord{
-		stmtID:  stmtID,
-		st:      st,
-		stmtCtx: stmtCtx,
-		params:  append(([]interface{})(nil), params...),
+		stmtID: stmtID,
+		st:     st,
+		params: append(([]interface{})(nil), params...),
 	}
 	h.history = append(h.history, s)
 }
@@ -477,6 +475,7 @@ func (s *session) retry(ctx context.Context, maxCnt uint) error {
 			if st.IsReadOnly() {
 				continue
 			}
+			executor.ResetContextOfStmt(s, st.(*executor.ExecStmt).StmtNode)
 			schemaVersion, err = st.RebuildPlan()
 			if err != nil {
 				return errors.Trace(err)
@@ -489,8 +488,6 @@ func (s *session) retry(ctx context.Context, maxCnt uint) error {
 			} else {
 				log.Warnf("con:%d schema_ver:%d retry_cnt:%d query_num:%d", connID, schemaVersion, retryCnt, i)
 			}
-			s.sessionVars.StmtCtx = sr.stmtCtx
-			s.sessionVars.StmtCtx.ResetForRetry()
 			_, err = st.Exec(ctx)
 			if err != nil {
 				s.StmtRollback()

session/session_test.go

@@ -1255,6 +1255,24 @@ func (s *testSessionSuite) TestDelete(c *C) {
 	tk1.MustQuery("select * from t;").Check(testkit.Rows("1"))
 }
 
+func (s *testSessionSuite) TestResetCtx(c *C) {
+	tk := testkit.NewTestKitWithInit(c, s.store)
+	tk1 := testkit.NewTestKitWithInit(c, s.store)
+
+	tk.MustExec("create table t (i int);")
+	tk.MustExec("insert into t values (1);")
+	tk.MustExec("begin;")
+	tk.MustExec("insert into t values (10);")
+	tk.MustExec("update t set i = i + row_count();")
+	tk.MustQuery("select * from t;").Check(testkit.Rows("2", "11"))
+
+	tk1.MustExec("update t set i = 0 where i = 1;")
+	tk1.MustQuery("select * from t;").Check(testkit.Rows("0"))
+
+	tk.MustExec("commit;")
+	tk.MustQuery("select * from t;").Check(testkit.Rows("1", "11"))
+}
+
 func (s *testSessionSuite) TestUnique(c *C) {
 	// test for https://github.com/pingcap/tidb/pull/461
 

sessionctx/stmtctx/stmtctx.go

@@ -234,15 +234,6 @@ func (sc *StatementContext) HandleOverflow(err error, warnErr error) error {
 	return err
 }
 
-// ResetForRetry resets the changed states during execution.
-func (sc *StatementContext) ResetForRetry() {
-	sc.mu.Lock()
-	sc.mu.affectedRows = 0
-	sc.mu.foundRows = 0
-	sc.mu.warnings = nil
-	sc.mu.Unlock()
-}
-
 // MergeExecDetails merges a single region execution details into self, used to print
 // the information in slow query log.
 func (sc *StatementContext) MergeExecDetails(details *execdetails.ExecDetails) {

sessionctx/variable/session_test.go

@@ -44,9 +44,4 @@ func (*testSessionSuite) TestSession(c *C) {
 	// For last insert id
 	ctx.GetSessionVars().SetLastInsertID(1)
 	c.Assert(ctx.GetSessionVars().LastInsertID, Equals, uint64(1))
-
-	ss.ResetForRetry()
-	c.Assert(ss.AffectedRows(), Equals, uint64(0))
-	c.Assert(ss.FoundRows(), Equals, uint64(0))
-	c.Assert(ss.WarningCount(), Equals, uint16(0))
 }