*: Make column test stable and fix prepare statement issue

ddl/ddl_db_test.go

@@ -597,13 +597,19 @@ LOOP:
 			// delete some rows, and add some data
 			for i := num; i < num+step; i++ {
 				n := rand.Intn(num)
+				s.tk.MustExec("begin")
 				s.tk.MustExec("delete from t2 where c1 = ?", n)
+				s.tk.MustExec("commit")
+
+				// Make sure that statement of insert and show use the same infoSchema.
+				s.tk.MustExec("begin")
 				_, err := s.tk.Exec("insert into t2 values (?, ?, ?)", i, i, i)
 				if err != nil {
 					// if err is failed, the column number must be 4 now.
 					values := s.showColumns(c, "t2")
-					c.Assert(values, HasLen, 4, Commentf("err:%v", err))
+					c.Assert(values, HasLen, 4, Commentf("err:%v", errors.ErrorStack(err)))
 				}
+				s.tk.MustExec("commit")
 			}
 			num += step
 		}
@@ -681,16 +687,15 @@ LOOP:
 		case <-ticker.C:
 			// delete some rows, and add some data
 			for i := num; i < num+step; i++ {
+				// Make sure that statement of insert and show use the same infoSchema.
+				s.tk.MustExec("begin")
 				_, err := s.tk.Exec("insert into t2 values (?, ?, ?)", i, i, i)
-				if err == nil {
-					continue
-				}
-				// If executing is failed, the column number must be 4 now.
-				values := s.showColumns(c, "t2")
-				if len(values) != 4 {
-					c.Log(errors.ErrorStack(err))
-					c.FailNow()
+				if err != nil {
+					// If executing is failed, the column number must be 4 now.
+					values := s.showColumns(c, "t2")
+					c.Assert(values, HasLen, 4, Commentf("err:%v", errors.ErrorStack(err)))
 				}
+				s.tk.MustExec("commit")
 			}
 			num += step
 		}

session.go

@@ -125,7 +125,11 @@ type session struct {
 
 func (s *session) cleanRetryInfo() {
 	if !s.sessionVars.RetryInfo.Retrying {
-		s.sessionVars.RetryInfo.Clean()
+		retryInfo := s.sessionVars.RetryInfo
+		for _, stmtID := range retryInfo.DroppedPreparedStmtIDs {
+			delete(s.sessionVars.PreparedStmts, stmtID)
+		}
+		retryInfo.Clean()
 	}
 }
 
@@ -567,7 +571,7 @@ func (s *session) DropPreparedStmt(stmtID uint32) error {
 	if _, ok := vars.PreparedStmts[stmtID]; !ok {
 		return executor.ErrStmtNotFound
 	}
-	delete(vars.PreparedStmts, stmtID)
+	vars.RetryInfo.DroppedPreparedStmtIDs = append(vars.RetryInfo.DroppedPreparedStmtIDs, stmtID)
 	return nil
 }
 

sessionctx/variable/session.go

@@ -36,9 +36,10 @@ var (
 
 // RetryInfo saves retry information.
 type RetryInfo struct {
-	Retrying         bool
-	currRetryOff     int
-	autoIncrementIDs []int64
+	Retrying               bool
+	DroppedPreparedStmtIDs []uint32
+	currRetryOff           int
+	autoIncrementIDs       []int64
 }
 
 // Clean does some clean work.
@@ -47,6 +48,9 @@ func (r *RetryInfo) Clean() {
 	if len(r.autoIncrementIDs) > 0 {
 		r.autoIncrementIDs = r.autoIncrementIDs[:0]
 	}
+	if len(r.DroppedPreparedStmtIDs) > 0 {
+		r.DroppedPreparedStmtIDs = r.DroppedPreparedStmtIDs[:0]
+	}
 }
 
 // AddAutoIncrementID adds id to AutoIncrementIDs.