Merge branch 'master' into global-stats-tests

pingcap · Mar 8, 2021 · ea9344a · ea9344a
2 parents 56f8f0d + 9a75e50
commit ea9344a
Show file tree

Hide file tree

Showing 3 changed files with 51 additions and 2 deletions.
diff --git a/store/tikv/2pc.go b/store/tikv/2pc.go
@@ -1126,7 +1126,7 @@ func (c *twoPhaseCommitter) execute(ctx context.Context) (err error) {
 			}
 		}
 	}
-	c.commitTS = commitTS
+	atomic.StoreUint64(&c.commitTS, commitTS)
 
 	if c.store.oracle.IsExpired(c.startTS, kv.MaxTxnTimeUse, &oracle.Option{TxnScope: oracle.GlobalTxnScope}) {
 		err = errors.Errorf("session %d txn takes too much time, txnStartTS: %d, comm: %d",

diff --git a/store/tikv/snapshot.go b/store/tikv/snapshot.go
@@ -481,6 +481,7 @@ func (s *tikvSnapshot) get(ctx context.Context, bo *Backoffer, k kv.Key) ([]byte
 				return nil, errors.Trace(err)
 			}
 
+			snapVer := s.version.Ver
 			if s.version == kv.MaxVersion {
 				newTS, err := tsFuture.Wait()
 				if err != nil {
@@ -494,7 +495,10 @@ func (s *tikvSnapshot) get(ctx context.Context, bo *Backoffer, k kv.Key) ([]byte
 				}
 			}
 
-			msBeforeExpired, err := cli.ResolveLocks(bo, s.version.Ver, []*Lock{lock})
+			// Use the original snapshot version to resolve locks so we can use MaxUint64
+			// as the callerStartTS if it's an auto-commit point get. This could save us
+			// one write at TiKV by not pushing forward the minCommitTS.
+			msBeforeExpired, err := cli.ResolveLocks(bo, snapVer, []*Lock{lock})
 			if err != nil {
 				return nil, errors.Trace(err)
 			}

diff --git a/store/tikv/snapshot_fail_test.go b/store/tikv/snapshot_fail_test.go
@@ -15,6 +15,8 @@ package tikv
 
 import (
 	"context"
+	"sync/atomic"
+	"time"
 
 	. "github.com/pingcap/check"
 	"github.com/pingcap/failpoint"
@@ -153,6 +155,7 @@ func (s *testSnapshotFailSuite) TestRetryPointGetWithTS(c *C) {
 	err = txn.Set([]byte("k4"), []byte("v4"))
 	c.Assert(err, IsNil)
 	txn.SetOption(kv.EnableAsyncCommit, true)
+	txn.SetOption(kv.Enable1PC, false)
 	txn.SetOption(kv.GuaranteeLinearizability, false)
 	// Prewrite an async-commit lock and do not commit it.
 	c.Assert(failpoint.Enable("github.com/pingcap/tidb/store/tikv/asyncCommitDoNothing", `return`), IsNil)
@@ -169,3 +172,45 @@ func (s *testSnapshotFailSuite) TestRetryPointGetWithTS(c *C) {
 
 	c.Assert(failpoint.Disable("github.com/pingcap/tidb/store/tikv/asyncCommitDoNothing"), IsNil)
 }
+
+func (s *testSnapshotFailSuite) TestRetryPointGetResolveTS(c *C) {
+	defer s.cleanup(c)
+
+	txn, err := s.store.Begin()
+	c.Assert(err, IsNil)
+	err = txn.Set([]byte("k1"), []byte("v1"))
+	err = txn.Set([]byte("k2"), []byte("v2"))
+	c.Assert(err, IsNil)
+	txn.SetOption(kv.EnableAsyncCommit, false)
+	txn.SetOption(kv.Enable1PC, false)
+	txn.SetOption(kv.GuaranteeLinearizability, false)
+
+	// Prewrite the lock without committing it
+	c.Assert(failpoint.Enable("github.com/pingcap/tidb/store/tikv/beforeCommit", `pause`), IsNil)
+	ch := make(chan struct{})
+	committer, err := newTwoPhaseCommitterWithInit(txn, 1)
+	c.Assert(committer.primary(), DeepEquals, []byte("k1"))
+	go func() {
+		c.Assert(err, IsNil)
+		err = committer.execute(context.Background())
+		c.Assert(err, IsNil)
+		ch <- struct{}{}
+	}()
+
+	// Wait until prewrite finishes
+	time.Sleep(200 * time.Millisecond)
+	// Should get nothing with max version, and **not pushing forward minCommitTS** of the primary lock
+	snapshot := s.store.GetSnapshot(kv.MaxVersion)
+	_, err = snapshot.Get(context.Background(), []byte("k2"))
+	c.Assert(err, ErrorMatches, ".*key not exist")
+
+	initialCommitTS := atomic.LoadUint64(&committer.commitTS)
+	c.Assert(failpoint.Disable("github.com/pingcap/tidb/store/tikv/beforeCommit"), IsNil)
+
+	<-ch
+	// check the minCommitTS is not pushed forward
+	snapshot = s.store.GetSnapshot(kv.Version{Ver: initialCommitTS})
+	v, err := snapshot.Get(context.Background(), []byte("k2"))
+	c.Assert(err, IsNil)
+	c.Assert(v, DeepEquals, []byte("v2"))
+}