session: remove extra useless tso request

executor/stale_txn_test.go

@@ -576,10 +576,13 @@ func (s *testStaleTxnSerialSuite) TestSetTransactionReadOnlyAsOf(c *C) {
 }
 
 func (s *testStaleTxnSerialSuite) TestValidateReadOnlyInStalenessTransaction(c *C) {
+	errMsg1 := ".*only support read-only statement during read-only staleness transactions.*"
+	errMsg2 := ".*select lock hasn't been supported in stale read yet.*"
 	testcases := []struct {
 		name       string
 		sql        string
 		isValidate bool
+		errMsg     string
 	}{
 		{
 			name:       "select statement",
@@ -595,6 +598,7 @@ func (s *testStaleTxnSerialSuite) TestValidateReadOnlyInStalenessTransaction(c *
 			name:       "explain analyze insert statement",
 			sql:        `explain analyze insert into t (id) values (1);`,
 			isValidate: false,
+			errMsg:     errMsg1,
 		},
 		{
 			name:       "explain analyze select statement",
@@ -605,6 +609,7 @@ func (s *testStaleTxnSerialSuite) TestValidateReadOnlyInStalenessTransaction(c *
 			name:       "execute insert statement",
 			sql:        `EXECUTE stmt1;`,
 			isValidate: false,
+			errMsg:     errMsg1,
 		},
 		{
 			name:       "execute select statement",
@@ -625,16 +630,19 @@ func (s *testStaleTxnSerialSuite) TestValidateReadOnlyInStalenessTransaction(c *
 			name:       "insert",
 			sql:        `insert into t (id) values (1);`,
 			isValidate: false,
+			errMsg:     errMsg1,
 		},
 		{
 			name:       "delete",
 			sql:        `delete from t where id =1`,
 			isValidate: false,
+			errMsg:     errMsg1,
 		},
 		{
 			name:       "update",
 			sql:        "update t set id =2 where id =1",
 			isValidate: false,
+			errMsg:     errMsg1,
 		},
 		{
 			name:       "point get",
@@ -660,41 +668,49 @@ func (s *testStaleTxnSerialSuite) TestValidateReadOnlyInStalenessTransaction(c *
 			name:       "select for update",
 			sql:        "select * from t where id = 1 for update",
 			isValidate: false,
+			errMsg:     errMsg2,
 		},
 		{
 			name:       "select lock in share mode",
 			sql:        "select * from t where id = 1 lock in share mode",
-			isValidate: true,
+			isValidate: false,
+			errMsg:     errMsg2,
 		},
 		{
 			name:       "select for update union statement",
 			sql:        "select * from t for update union select * from t;",
 			isValidate: false,
+			errMsg:     errMsg1,
 		},
 		{
 			name:       "replace statement",
 			sql:        "replace into t(id) values (1)",
 			isValidate: false,
+			errMsg:     errMsg1,
 		},
 		{
 			name:       "load data statement",
 			sql:        "LOAD DATA LOCAL INFILE '/mn/asa.csv' INTO TABLE t FIELDS TERMINATED BY x'2c' ENCLOSED BY b'100010' LINES TERMINATED BY '\r\n' IGNORE 1 LINES (id);",
 			isValidate: false,
+			errMsg:     errMsg1,
 		},
 		{
 			name:       "update multi tables",
 			sql:        "update t,t1 set t.id = 1,t1.id = 2 where t.1 = 2 and t1.id = 3;",
 			isValidate: false,
+			errMsg:     errMsg1,
 		},
 		{
 			name:       "delete multi tables",
 			sql:        "delete t from t1 where t.id = t1.id",
 			isValidate: false,
+			errMsg:     errMsg1,
 		},
 		{
 			name:       "insert select",
 			sql:        "insert into t select * from t1;",
 			isValidate: false,
+			errMsg:     errMsg1,
 		},
 	}
 	tk := testkit.NewTestKit(c, s.store)
@@ -720,7 +736,7 @@ func (s *testStaleTxnSerialSuite) TestValidateReadOnlyInStalenessTransaction(c *
 		} else {
 			err := tk.ExecToErr(testcase.sql)
 			c.Assert(err, NotNil)
-			c.Assert(err.Error(), Matches, `.*only support read-only statement during read-only staleness transactions.*`)
+			c.Assert(err.Error(), Matches, testcase.errMsg)
 		}
 		tk.MustExec("commit")
 		tk.MustExec("set transaction read only as of timestamp NOW(3);")
@@ -729,8 +745,9 @@ func (s *testStaleTxnSerialSuite) TestValidateReadOnlyInStalenessTransaction(c *
 		} else {
 			err := tk.ExecToErr(testcase.sql)
 			c.Assert(err, NotNil)
-			c.Assert(err.Error(), Matches, `.*only support read-only statement during read-only staleness transactions.*`)
+			c.Assert(err.Error(), Matches, testcase.errMsg)
 		}
+		// clean the status
 		tk.MustExec("set transaction read only as of timestamp ''")
 	}
 }
@@ -1181,3 +1198,50 @@ func (s *testStaleTxnSerialSuite) TestStaleReadCompatibility(c *C) {
 	c.Assert(tk.MustQuery("select * from t;").Rows(), HasLen, 3)
 	c.Assert(failpoint.Disable("github.com/pingcap/tidb/expression/injectNow"), IsNil)
 }
+
+func (s *testStaleTxnSerialSuite) TestStaleReadNoExtraTSORequest(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	// For mocktikv, safe point is not initialized, we manually insert it for snapshot to use.
+	safePointName := "tikv_gc_safe_point"
+	safePointValue := "20160102-15:04:05 -0700"
+	safePointComment := "All versions after safe point can be accessed. (DO NOT EDIT)"
+	updateSafePoint := fmt.Sprintf(`INSERT INTO mysql.tidb VALUES ('%[1]s', '%[2]s', '%[3]s')
+	ON DUPLICATE KEY
+	UPDATE variable_value = '%[2]s', comment = '%[3]s'`, safePointName, safePointValue, safePointComment)
+	tk.MustExec(updateSafePoint)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t (id int);")
+	time.Sleep(3 * time.Second)
+	// statement stale read
+	c.Assert(failpoint.Enable("github.com/pingcap/tidb/session/assertTSONotRequest", `return(true)`), IsNil)
+	tk.MustQuery("select * from t as of timestamp NOW() - INTERVAL 2 SECOND")
+	failpoint.Disable("github.com/pingcap/tidb/session/assertTSONotRequest")
+
+	// set and statement stale read
+	tk.MustExec("set transaction read only as of timestamp NOW() - INTERVAL 2 SECOND")
+	c.Assert(failpoint.Enable("github.com/pingcap/tidb/session/assertTSONotRequest", `return(true)`), IsNil)
+	tk.MustQuery("select * from t")
+	failpoint.Disable("github.com/pingcap/tidb/session/assertTSONotRequest")
+
+	// stale read transaction
+	c.Assert(failpoint.Enable("github.com/pingcap/tidb/session/assertTSONotRequest", `return(true)`), IsNil)
+	tk.MustExec("start transaction read only as of timestamp NOW() - INTERVAL 2 SECOND")
+	tk.MustQuery("select * from t")
+	tk.MustExec("commit")
+	failpoint.Disable("github.com/pingcap/tidb/session/assertTSONotRequest")
+
+	// set and stale read transaction
+	tk.MustExec("set transaction read only as of timestamp NOW() - INTERVAL 2 SECOND")
+	c.Assert(failpoint.Enable("github.com/pingcap/tidb/session/assertTSONotRequest", `return(true)`), IsNil)
+	tk.MustExec("begin")
+	tk.MustQuery("select * from t")
+	tk.MustExec("commit")
+	failpoint.Disable("github.com/pingcap/tidb/session/assertTSONotRequest")
+
+	// use tidb_read_staleness
+	tk.MustExec(`set @@tidb_read_staleness='-1'`)
+	c.Assert(failpoint.Enable("github.com/pingcap/tidb/session/assertTSONotRequest", `return(true)`), IsNil)
+	tk.MustQuery("select * from t")
+	failpoint.Disable("github.com/pingcap/tidb/session/assertTSONotRequest")
+}

planner/core/preprocess.go

@@ -299,6 +299,21 @@ func (p *preprocessor) Enter(in ast.Node) (out ast.Node, skipChildren bool) {
 		for _, cte := range node.CTEs {
 			p.withName[cte.Name.L] = struct{}{}
 		}
+	case *ast.BeginStmt:
+		// If the begin statement was like following:
+		// start transaction read only as of timestamp ....
+		// then we need set StmtCtx.IsStaleness as true in order to avoid take tso in PrepareTSFuture.
+		if node.AsOf != nil {
+			p.ctx.GetSessionVars().StmtCtx.IsStaleness = true
+			p.IsStaleness = true
+		} else if p.ctx.GetSessionVars().TxnReadTS.PeakTxnReadTS() > 0 {
+			// If the begin statement was like following:
+			// set transaction read only as of timestamp ...
+			// begin
+			// then we need set StmtCtx.IsStaleness as true in order to avoid take tso in PrepareTSFuture.
+			p.ctx.GetSessionVars().StmtCtx.IsStaleness = true
+			p.IsStaleness = true
+		}
 	default:
 		p.flag &= ^parentIsJoin
 	}
@@ -1531,6 +1546,17 @@ func (p *preprocessor) checkFuncCastExpr(node *ast.FuncCastExpr) {
 
 // handleAsOfAndReadTS tries to handle as of closure, or possibly read_ts.
 func (p *preprocessor) handleAsOfAndReadTS(node *ast.AsOfClause) {
+	if p.stmtTp != TypeSelect {
+		return
+	}
+	defer func() {
+		// If the select statement was like 'select * from t as of timestamp ...' or in a stale read transaction
+		// or is affected by the tidb_read_staleness session variable, then the statement will be makred as isStaleness
+		// in stmtCtx
+		if p.flag&inPrepare == 0 && p.IsStaleness {
+			p.ctx.GetSessionVars().StmtCtx.IsStaleness = true
+		}
+	}()
 	// When statement is during the Txn, we check whether there exists AsOfClause. If exists, we will return error,
 	// otherwise we should directly set the return param from TxnCtx.
 	p.ReadReplicaScope = kv.GlobalReplicaScope
@@ -1597,6 +1623,10 @@ func (p *preprocessor) handleAsOfAndReadTS(node *ast.AsOfClause) {
 			p.IsStaleness = true
 		}
 	case readTS == 0 && node == nil && readStaleness != 0:
+		// If both readTS and node is empty while the readStaleness isn't, it means we meet following situation:
+		// set @@tidb_read_staleness='-5';
+		// select * from t;
+		// Then the following select statement should be affected by the tidb_read_staleness in session.
 		ts, p.err = calculateTsWithReadStaleness(p.ctx, readStaleness)
 		if p.err != nil {
 			return
@@ -1613,6 +1643,8 @@ func (p *preprocessor) handleAsOfAndReadTS(node *ast.AsOfClause) {
 			p.IsStaleness = true
 		}
 	}
+
+	// If the select statement is related to multi tables, we should grantee that all tables use the same timestamp
 	if p.LastSnapshotTS != ts {
 		p.err = ErrAsOf.GenWithStack("can not set different time in the as of")
 		return
@@ -1631,9 +1663,6 @@ func (p *preprocessor) handleAsOfAndReadTS(node *ast.AsOfClause) {
 		}
 		p.InfoSchema = temptable.AttachLocalTemporaryTableInfoSchema(p.ctx, is)
 	}
-	if p.flag&inPrepare == 0 {
-		p.ctx.GetSessionVars().StmtCtx.IsStaleness = p.IsStaleness
-	}
 	p.initedLastSnapshotTS = true
 }
 

session/session.go

@@ -1608,10 +1608,20 @@ func (s *session) validateStatementReadOnlyInStaleness(stmtNode ast.StmtNode) er
 	}
 	errMsg := "only support read-only statement during read-only staleness transactions"
 	node := stmtNode.(ast.Node)
-	switch node.(type) {
+	switch v := node.(type) {
 	case *ast.SplitRegionStmt:
 		return nil
-	case *ast.SelectStmt, *ast.ExplainStmt, *ast.DoStmt, *ast.ShowStmt, *ast.SetOprStmt, *ast.ExecuteStmt, *ast.SetOprSelectList:
+	case *ast.SelectStmt:
+		// select lock statement needs start a transaction which will be conflict to stale read,
+		// we forbid select lock statement in stale read for now.
+		if v.LockInfo != nil {
+			return errors.New("select lock hasn't been supported in stale read yet")
+		}
+		if !planner.IsReadOnly(stmtNode, vars) {
+			return errors.New(errMsg)
+		}
+		return nil
+	case *ast.ExplainStmt, *ast.DoStmt, *ast.ShowStmt, *ast.SetOprStmt, *ast.ExecuteStmt, *ast.SetOprSelectList:
 		if !planner.IsReadOnly(stmtNode, vars) {
 			return errors.New(errMsg)
 		}
@@ -2778,6 +2788,14 @@ func (s *session) PrepareTSFuture(ctx context.Context) {
 		return
 	}
 	if !s.txn.validOrPending() {
+		if s.GetSessionVars().StmtCtx.IsStaleness {
+			// Do nothing when StmtCtx.IsStaleness is true
+			// we don't need to request tso for stale read
+			return
+		}
+		failpoint.Inject("assertTSONotRequest", func() {
+			panic("tso shouldn't be requested")
+		})
 		// Prepare the transaction future if the transaction is invalid (at the beginning of the transaction).
 		txnFuture := s.getTxnFuture(ctx)
 		s.txn.changeInvalidToPending(txnFuture)

sessionctx/stmtctx/stmtctx.go

@@ -87,7 +87,11 @@ type StatementContext struct {
 	IgnoreNoPartition            bool
 	MaybeOverOptimized4PlanCache bool
 	IgnoreExplainIDSuffix        bool
-	IsStaleness                  bool
+
+	// If the select statement was like 'select * from t as of timestamp ...' or in a stale read transaction
+	// or is affected by the tidb_read_staleness session variable, then the statement will be makred as isStaleness
+	// in stmtCtx
+	IsStaleness bool
 
 	// mu struct holds variables that change during execution.
 	mu struct {