executor: add check data visibility for point get (#14459)

executor/point_get_test.go

@@ -14,9 +14,12 @@
 package executor_test
 
 import (
+	"context"
 	"fmt"
+	"time"
 
 	. "github.com/pingcap/check"
+	"github.com/pingcap/parser/terror"
 	"github.com/pingcap/tidb/domain"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/session"
@@ -445,3 +448,35 @@ func (s *testPointGetSuite) TestPointGetByRowID(c *C) {
 		"Point_Get_1 1.00 root table:t, handle:1"))
 	tk.MustQuery("select * from t where t._tidb_rowid = 1").Check(testkit.Rows("aaa 12"))
 }
+
+func (s *testPointGetSuite) TestSelectCheckVisibility(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t (a varchar(10) key, b int,index idx(b))")
+	tk.MustExec("insert into t values('1',1)")
+	tk.MustExec("begin")
+	txn, err := tk.Se.Txn(false)
+	c.Assert(err, IsNil)
+	ts := txn.StartTS()
+	store := tk.Se.GetStore().(tikv.Storage)
+	// Update gc safe time for check data visibility.
+	store.UpdateSPCache(ts+1, time.Now())
+	checkSelectResultError := func(sql string, expectErr *terror.Error) {
+		re, err := tk.Exec(sql)
+		c.Assert(err, IsNil)
+		_, err = session.ResultSetToStringSlice(context.Background(), tk.Se, re)
+		c.Assert(err, NotNil)
+		c.Assert(expectErr.Equal(err), IsTrue)
+	}
+	// Test point get.
+	checkSelectResultError("select * from t where a='1'", tikv.ErrGCTooEarly)
+	// Test batch point get.
+	checkSelectResultError("select * from t where a in ('1','2')", tikv.ErrGCTooEarly)
+	// Test Index look up read.
+	checkSelectResultError("select * from t where b > 0 ", tikv.ErrGCTooEarly)
+	// Test Index read.
+	checkSelectResultError("select b from t where b > 0 ", tikv.ErrGCTooEarly)
+	// Test table read.
+	checkSelectResultError("select * from t", tikv.ErrGCTooEarly)
+}

store/tikv/snapshot.go

@@ -266,6 +266,11 @@ func (s *tikvSnapshot) Get(k kv.Key) ([]byte, error) {
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
+	err = s.store.CheckVisibility(s.version.Ver)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
 	if len(val) == 0 {
 		return nil, kv.ErrNotExist
 	}

store/tikv/txn.go

@@ -140,11 +140,6 @@ func (txn *tikvTxn) Get(k kv.Key) ([]byte, error) {
 		return nil, errors.Trace(err)
 	}
 
-	err = txn.store.CheckVisibility(txn.startTS)
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
-
 	return ret, nil
 }