executor: do not read from lock cache when snapshot read (#21529)

executor/batch_point_get.go

@@ -157,7 +157,11 @@ func (e *BatchPointGetExec) initialize(ctx context.Context) error {
 	snapshot.SetOption(kv.TaskID, e.ctx.GetSessionVars().StmtCtx.TaskID)
 	var batchGetter kv.BatchGetter = snapshot
 	if sessVars.InTxn() {
-		batchGetter = kv.NewBufferBatchGetter(txn.GetMemBuffer(), &PessimisticLockCacheGetter{txnCtx: txnCtx}, snapshot)
+		if e.lock {
+			batchGetter = kv.NewBufferBatchGetter(txn.GetMemBuffer(), &PessimisticLockCacheGetter{txnCtx: txnCtx}, snapshot)
+		} else {
+			batchGetter = kv.NewBufferBatchGetter(txn.GetMemBuffer(), nil, snapshot)
+		}
 	}
 
 	var handleVals map[string][]byte

executor/point_get.go

@@ -304,10 +304,12 @@ func (e *PointGetExecutor) get(ctx context.Context, key kv.Key) ([]byte, error)
 			return nil, err
 		}
 		// key does not exist in mem buffer, check the lock cache
-		var ok bool
-		val, ok = e.ctx.GetSessionVars().TxnCtx.GetKeyInPessimisticLockCache(key)
-		if ok {
-			return val, nil
+		if e.lock {
+			var ok bool
+			val, ok = e.ctx.GetSessionVars().TxnCtx.GetKeyInPessimisticLockCache(key)
+			if ok {
+				return val, nil
+			}
 		}
 		// fallthrough to snapshot get.
 	}

executor/update_test.go

@@ -235,3 +235,27 @@ func (s *testUpdateSuite) TestUpdateMultiDatabaseTable(c *C) {
 	tk.MustExec("create table test2.t(a int, b int generated always  as (a+1) virtual)")
 	tk.MustExec("update t, test2.t set test.t.a=1")
 }
+
+func (s *testPointGetSuite) TestIssue21447(c *C) {
+	tk1, tk2 := testkit.NewTestKit(c, s.store), testkit.NewTestKit(c, s.store)
+	tk1.MustExec("use test")
+	tk2.MustExec("use test")
+
+	tk1.MustExec("drop table if exists t1")
+	tk1.MustExec("create table t1(id int primary key, name varchar(40))")
+	tk1.MustExec("insert into t1 values(1, 'abc')")
+
+	tk1.MustExec("begin pessimistic")
+	tk2.MustExec("begin pessimistic")
+	tk2.MustExec("update t1 set name='xyz' where id=1")
+	tk2.CheckExecResult(1, 0)
+	tk2.MustQuery("select * from t1 where id = 1").Check(testkit.Rows("1 xyz"))
+	tk2.MustExec("commit")
+	tk1.MustExec("update t1 set name='xyz' where id=1")
+	tk1.CheckExecResult(0, 0)
+	tk1.MustQuery("select * from t1 where id = 1").Check(testkit.Rows("1 abc"))
+	tk1.MustQuery("select * from t1 where id = 1 for update").Check(testkit.Rows("1 xyz"))
+	tk1.MustQuery("select * from t1 where id in (1, 2)").Check(testkit.Rows("1 abc"))
+	tk1.MustQuery("select * from t1 where id in (1, 2) for update").Check(testkit.Rows("1 xyz"))
+	tk1.MustExec("commit")
+}