From eda2f7240b5405613b36afb2582f87bdd4b70464 Mon Sep 17 00:00:00 2001
From: ti-srebot <66930949+ti-srebot@users.noreply.github.com>
Date: Fri, 25 Mar 2022 19:38:34 +0800
Subject: [PATCH] session: forbid plan cache in stale read (#33273) (#33301)

close pingcap/tidb#31550
---
 executor/stale_txn_test.go     | 30 ++++++++++++++++++++++++++++++
 session/session.go             | 16 +++++-----------
 sessiontxn/txn_context_test.go | 14 --------------
 3 files changed, 35 insertions(+), 25 deletions(-)

diff --git a/executor/stale_txn_test.go b/executor/stale_txn_test.go
index 8ce913abdee1b..cf444ccf75443 100644
--- a/executor/stale_txn_test.go
+++ b/executor/stale_txn_test.go
@@ -15,6 +15,7 @@
 package executor_test
 
 import (
+	"context"
 	"fmt"
 	"testing"
 	"time"
@@ -1282,3 +1283,32 @@ func TestStaleReadNoExtraTSORequest(t *testing.T) {
 	tk.MustQuery("select * from t")
 	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/session/assertTSONotRequest"))
 }
+
+func TestPlanCacheWithStaleReadByBinaryProto(t *testing.T) {
+	store, _, clean := testkit.CreateMockStoreAndDomain(t)
+	defer clean()
+
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t1 (id int primary key, v int)")
+	tk.MustExec("insert into t1 values(1, 10)")
+	se := tk.Session()
+	tk.MustExec("set @a=now(6)")
+	time.Sleep(time.Millisecond * 5)
+	tk.MustExec("update t1 set v=100 where id=1")
+
+	stmtID1, _, _, err := se.PrepareStmt("select * from t1 as of timestamp @a where id=1")
+	require.NoError(t, err)
+
+	rs, err := se.ExecutePreparedStmt(context.TODO(), stmtID1, nil)
+	require.NoError(t, err)
+	tk.ResultSetToResult(rs, fmt.Sprintf("%v", rs)).Check(testkit.Rows("1 10"))
+
+	rs, err = se.ExecutePreparedStmt(context.TODO(), stmtID1, nil)
+	require.NoError(t, err)
+	tk.ResultSetToResult(rs, fmt.Sprintf("%v", rs)).Check(testkit.Rows("1 10"))
+
+	rs, err = se.ExecutePreparedStmt(context.TODO(), stmtID1, nil)
+	require.NoError(t, err)
+	tk.ResultSetToResult(rs, fmt.Sprintf("%v", rs)).Check(testkit.Rows("1 10"))
+}
diff --git a/session/session.go b/session/session.go
index 2e8449182ca93..51efe43531ee4 100644
--- a/session/session.go
+++ b/session/session.go
@@ -2246,23 +2246,17 @@ func (s *session) cachedPlanExec(ctx context.Context,
 // IsPointGetWithPKOrUniqueKeyByAutoCommit
 func (s *session) IsCachedExecOk(ctx context.Context, preparedStmt *plannercore.CachedPrepareStmt) (bool, error) {
 	prepared := preparedStmt.PreparedAst
-	if prepared.CachedPlan == nil {
+	if prepared.CachedPlan == nil || preparedStmt.SnapshotTSEvaluator != nil {
 		return false, nil
 	}
 	// check auto commit
 	if !plannercore.IsAutoCommitTxn(s) {
 		return false, nil
 	}
-	// SnapshotTSEvaluator != nil, it is stale read
-	// stale read expect a stale infoschema
-	// so skip infoschema check
-	if preparedStmt.SnapshotTSEvaluator == nil {
-		// check schema version
-		is := s.GetInfoSchema().(infoschema.InfoSchema)
-		if prepared.SchemaVersion != is.SchemaMetaVersion() {
-			prepared.CachedPlan = nil
-			return false, nil
-		}
+	is := s.GetInfoSchema().(infoschema.InfoSchema)
+	if prepared.SchemaVersion != is.SchemaMetaVersion() {
+		prepared.CachedPlan = nil
+		return false, nil
 	}
 	// maybe we'd better check cached plan type here, current
 	// only point select/update will be cached, see "getPhysicalPlan" func
diff --git a/sessiontxn/txn_context_test.go b/sessiontxn/txn_context_test.go
index 290569a8b8537..a0d7816e6b595 100644
--- a/sessiontxn/txn_context_test.go
+++ b/sessiontxn/txn_context_test.go
@@ -639,13 +639,6 @@ func TestTxnContextForStaleReadInPrepare(t *testing.T) {
 		tk.MustExec("execute s2")
 	})
 
-	// plan cache for stmtID2
-	doWithCheckPath(t, se, []string{"assertTxnManagerInCachedPlanExec", "assertTxnManagerInShortPointGetPlan"}, func() {
-		rs, err := se.ExecutePreparedStmt(context.TODO(), stmtID2, nil)
-		require.NoError(t, err)
-		tk.ResultSetToResult(rs, fmt.Sprintf("%v", rs)).Check(testkit.Rows("1 10"))
-	})
-
 	// tx_read_ts in prepare
 	se.SetValue(sessiontxn.AssertTxnInfoSchemaKey, is1)
 	doWithCheckPath(t, se, path, func() {
@@ -656,13 +649,6 @@ func TestTxnContextForStaleReadInPrepare(t *testing.T) {
 	doWithCheckPath(t, se, normalPathRecords, func() {
 		tk.MustExec("execute s3")
 	})
-
-	// plan cache for stmtID3
-	doWithCheckPath(t, se, []string{"assertTxnManagerInCachedPlanExec", "assertTxnManagerInShortPointGetPlan"}, func() {
-		rs, err := se.ExecutePreparedStmt(context.TODO(), stmtID3, nil)
-		require.NoError(t, err)
-		tk.ResultSetToResult(rs, fmt.Sprintf("%v", rs)).Check(testkit.Rows("1 10"))
-	})
 }
 
 func TestTxnContextPreparedStmtWithForUpdate(t *testing.T) {