tidb: update statement history in retry() (#5207)

* tidb: update statement history in retry()

Fix a bug in this order:
1. transaction commit meet schema change
2. retry and meet retryable error
3. retry again and success

It would insert stale data, because statement history is not updated,
we may using the wrong schema.

* update to fix ci

new_session_test.go

@@ -1470,8 +1470,39 @@ func (s *testSchemaSuite) TestCommitWhenSchemaChanged(c *C) {
 
 	tk.MustExec("alter table t drop column b")
 
-	// When s2 commit, it will find schema already changed.
+	// When tk1 commit, it will find schema already changed.
 	tk1.MustExec("insert into t values (4, 4)")
 	_, err := tk1.Exec("commit")
 	c.Assert(terror.ErrorEqual(err, executor.ErrWrongValueCountOnRow), IsTrue)
 }
+
+func (s *testSchemaSuite) TestRetrySchemaChange(c *C) {
+	tk := testkit.NewTestKitWithInit(c, s.store)
+	tk1 := testkit.NewTestKitWithInit(c, s.store)
+	tk.MustExec("create table t (a int primary key, b int)")
+	tk.MustExec("insert into t values (1, 1)")
+
+	tk1.MustExec("begin")
+	tk1.MustExec("update t set b = 5 where a = 1")
+
+	tk.MustExec("alter table t add index b_i (b)")
+
+	run := false
+	hook := func() {
+		if run == false {
+			tk.MustExec("update t set b = 3 where a = 1")
+			run = true
+		}
+	}
+
+	// In order to cover a bug that statement history is not updated during retry.
+	// See https://github.com/pingcap/tidb/pull/5202
+	// Step1: when tk1 commit, it find schema changed and retry().
+	// Step2: during retry, hook() is called, tk update primary key.
+	// Step3: tk1 continue commit in retry() meet a retryable error(write conflict), retry again.
+	// Step4: tk1 retry() success, if it use the stale statement, data and index will inconsistent.
+	tk1.Se.SetValue(tidb.PreCommitHook{}, hook)
+	err := tk1.Se.CommitTxn()
+	c.Assert(err, IsNil)
+	tk.MustQuery("select * from t where t.b = 5").Check(testkit.Rows("1 5"))
+}

session.go

@@ -400,6 +400,7 @@ func (s *session) retry(maxCnt int, infoSchemaChanged bool) error {
 		s.txn = nil
 		s.sessionVars.SetStatusFlag(mysql.ServerStatusInTrans, false)
 	}()
+
 	nh := GetHistory(s)
 	var err error
 	for {
@@ -416,6 +417,7 @@ func (s *session) retry(maxCnt int, infoSchemaChanged bool) error {
 				if err != nil {
 					return errors.Trace(err)
 				}
+				nh.history[i].st = st
 			}
 
 			if retryCnt == 0 {
@@ -432,6 +434,10 @@ func (s *session) retry(maxCnt int, infoSchemaChanged bool) error {
 				break
 			}
 		}
+		if hook := s.Value(PreCommitHook{}); hook != nil {
+			// For testing purpose.
+			hook.(func())()
+		}
 		if err == nil {
 			err = s.doCommit()
 			if err == nil {
@@ -1297,3 +1303,10 @@ func logCrucialStmt(node ast.StmtNode, user *auth.UserIdentity) {
 		}
 	}
 }
+
+// PreCommitHook implements the fmt.Stringer interface.
+type PreCommitHook struct{}
+
+func (h PreCommitHook) String() string {
+	return "preCommitHook"
+}