unistore: Adjust some behaviors to be consistent with TiKV (#43397)

go.mod

@@ -53,10 +53,16 @@ require (
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c
 	github.com/pingcap/failpoint v0.0.0-20210316064728-7acb0f0a3dfd
 	github.com/pingcap/fn v0.0.0-20200306044125-d5540d389059
+<<<<<<< HEAD
 	github.com/pingcap/kvproto v0.0.0-20220428033740-e4924274acd8
 	github.com/pingcap/log v1.1.1-0.20221116035753-734d527bc87c
 	github.com/pingcap/sysutil v0.0.0-20211208032423-041a72e5860d
 	github.com/pingcap/tidb-tools v5.2.2-0.20211019062242-37a8bef2fa17+incompatible
+=======
+	github.com/pingcap/kvproto v0.0.0-20230424092600-14ac513b9eff
+	github.com/pingcap/log v1.1.1-0.20230317032135-a0d097d16e22
+	github.com/pingcap/sysutil v1.0.1-0.20230407040306-fb007c5aff21
+>>>>>>> 9b9796fc5c8 (unistore: Adjust some behaviors to be consistent with TiKV (#43397))
 	github.com/pingcap/tidb/parser v0.0.0-20211011031125-9b13dc409c5e
 	github.com/pingcap/tipb v0.0.0-20220704075611-01674e683e38
 	github.com/prometheus/client_golang v1.11.1

go.sum

@@ -640,12 +640,17 @@ github.com/pingcap/fn v0.0.0-20200306044125-d5540d389059/go.mod h1:fMRU1BA1y+r89
 github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989 h1:surzm05a8C9dN8dIUmo4Be2+pMRb6f55i+UIYrluu2E=
 github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989/go.mod h1:O17XtbryoCJhkKGbT62+L2OlrniwqiGLSqrmdHCMzZw=
 github.com/pingcap/kvproto v0.0.0-20191211054548-3c6b38ea5107/go.mod h1:WWLmULLO7l8IOcQG+t+ItJ3fEcrL5FxF0Wu+HrMy26w=
+<<<<<<< HEAD
 github.com/pingcap/kvproto v0.0.0-20200411081810-b85805c9476c/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/pingcap/kvproto v0.0.0-20210819164333-bd5706b9d9f2/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/pingcap/kvproto v0.0.0-20211109071446-a8b4d34474bc/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/pingcap/kvproto v0.0.0-20211122024046-03abd340988f/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
 github.com/pingcap/kvproto v0.0.0-20220428033740-e4924274acd8 h1:b/nHmr8qwWHg0H+h52ts4GkcnoOltVNBYGc8fAbcSuw=
 github.com/pingcap/kvproto v0.0.0-20220428033740-e4924274acd8/go.mod h1:IOdRDPLyda8GX2hE/jO7gqaCV/PNFh8BZQCQZXfIOqI=
+=======
+github.com/pingcap/kvproto v0.0.0-20230424092600-14ac513b9eff h1:58k95xMDOJkpwKs2ULr/KbEZmU2+UprcguoR/pYy6MA=
+github.com/pingcap/kvproto v0.0.0-20230424092600-14ac513b9eff/go.mod h1:guCyM5N+o+ru0TsoZ1hi9lDjUMs2sIBjW3ARTEpVbnk=
+>>>>>>> 9b9796fc5c8 (unistore: Adjust some behaviors to be consistent with TiKV (#43397))
 github.com/pingcap/log v0.0.0-20191012051959-b742a5d432e9/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
 github.com/pingcap/log v0.0.0-20200511115504-543df19646ad/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
 github.com/pingcap/log v0.0.0-20210317133921-96f4fcab92a4/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=

store/mockstore/unistore/tikv/errors.go

@@ -42,8 +42,8 @@ func BuildLockErr(key []byte, lock *mvcc.Lock) *ErrLocked {
 func (e *ErrLocked) Error() string {
 	lock := e.Lock
 	return fmt.Sprintf(
-		"key is locked, key: %q, Type: %v, primary: %q, startTS: %v, forUpdateTS: %v, useAsyncCommit: %v",
-		e.Key, lock.Op, lock.Primary, lock.StartTS, lock.ForUpdateTS, lock.UseAsyncCommit,
+		"key is locked, key: %v, lock: %v",
+		hex.EncodeToString(e.Key), lock.String(),
 	)
 }
 
@@ -132,3 +132,30 @@ type ErrTxnNotFound struct {
 func (e *ErrTxnNotFound) Error() string {
 	return "txn not found"
 }
+<<<<<<< HEAD:store/mockstore/unistore/tikv/errors.go
+=======
+
+// ErrAssertionFailed is returned if any assertion fails on a transaction request.
+type ErrAssertionFailed struct {
+	StartTS          uint64
+	Key              []byte
+	Assertion        kvrpcpb.Assertion
+	ExistingStartTS  uint64
+	ExistingCommitTS uint64
+}
+
+func (e *ErrAssertionFailed) Error() string {
+	return fmt.Sprintf("AssertionFailed { StartTS: %v, Key: %v, Assertion: %v, ExistingStartTS: %v, ExistingCommitTS: %v }",
+		e.StartTS, hex.EncodeToString(e.Key), e.Assertion.String(), e.ExistingStartTS, e.ExistingCommitTS)
+}
+
+// ErrPrimaryMismatch is returned if CheckTxnStatus request is sent to a secondary lock.
+type ErrPrimaryMismatch struct {
+	Key  []byte
+	Lock *mvcc.Lock
+}
+
+func (e *ErrPrimaryMismatch) Error() string {
+	return fmt.Sprintf("primary mismatch, key: %v, lock: %v", hex.EncodeToString(e.Key), e.Lock.String())
+}
+>>>>>>> 9b9796fc5c8 (unistore: Adjust some behaviors to be consistent with TiKV (#43397)):store/mockstore/unistore/tikv/kverrors/errors.go

store/mockstore/unistore/tikv/mvcc.go

@@ -405,6 +405,13 @@ func (store *MVCCStore) CheckTxnStatus(reqCtx *requestCtx,
 	lock := store.getLock(reqCtx, req.PrimaryKey)
 	batch := store.dbWriter.NewWriteBatch(req.LockTs, 0, reqCtx.rpcCtx)
 	if lock != nil && lock.StartTS == req.LockTs {
+		if !bytes.Equal(req.PrimaryKey, lock.Primary) {
+			return TxnStatus{}, &kverrors.ErrPrimaryMismatch{
+				Key:  req.PrimaryKey,
+				Lock: lock,
+			}
+		}
+
 		// For an async-commit lock, never roll it back or push forward it MinCommitTS.
 		if lock.UseAsyncCommit && !req.ForceSyncCommit {
 			log.S().Debugf("async commit startTS=%v secondaries=%v minCommitTS=%v", lock.StartTS, lock.Secondaries, lock.MinCommitTS)
@@ -952,10 +959,6 @@ func (store *MVCCStore) Commit(req *requestCtx, keys [][]byte, startTS, commitTS
 				Key:         key,
 			}
 		}
-		if lock.Op == uint8(kvrpcpb.Op_PessimisticLock) {
-			log.Warn("commit a pessimistic lock with Lock type", zap.Binary("key", key), zap.Uint64("start ts", startTS), zap.Uint64("commit ts", commitTS))
-			lock.Op = uint8(kvrpcpb.Op_Lock)
-		}
 		isPessimisticTxn = lock.ForUpdateTS > 0
 		tmpDiff += len(key) + len(lock.Value)
 		batch.Commit(key, &lock)
@@ -1197,12 +1200,7 @@ func (store *MVCCStore) Cleanup(reqCtx *requestCtx, key []byte, startTS, current
 func (store *MVCCStore) appendScannedLock(locks []*kvrpcpb.LockInfo, it *lockstore.Iterator, maxTS uint64) []*kvrpcpb.LockInfo {
 	lock := mvcc.DecodeLock(it.Value())
 	if lock.StartTS < maxTS {
-		locks = append(locks, &kvrpcpb.LockInfo{
-			PrimaryLock: lock.Primary,
-			LockVersion: lock.StartTS,
-			Key:         safeCopy(it.Key()),
-			LockTtl:     uint64(lock.TTL),
-		})
+		locks = append(locks, lock.ToLockInfo(append([]byte{}, it.Key()...)))
 	}
 	return locks
 }

store/mockstore/unistore/tikv/mvcc/mvcc.go

@@ -16,6 +16,8 @@ package mvcc
 
 import (
 	"encoding/binary"
+	"encoding/hex"
+	"fmt"
 	"unsafe"
 
 	"github.com/pingcap/kvproto/pkg/kvrpcpb"
@@ -114,6 +116,18 @@ func (l *Lock) ToLockInfo(key []byte) *kvrpcpb.LockInfo {
 	}
 }
 
+// String implements fmt.Stringer for Lock.
+func (l *Lock) String() string {
+	return fmt.Sprintf(
+		"Lock { Type: %v, StartTS: %v,  ForUpdateTS: %v, Primary: %v, UseAsyncCommit: %v }",
+		kvrpcpb.Op(l.Op).String(),
+		l.StartTS,
+		l.ForUpdateTS,
+		hex.EncodeToString(l.Primary),
+		l.UseAsyncCommit,
+	)
+}
+
 // UserMeta value for lock.
 const (
 	LockUserMetaNoneByte   = 0

store/mockstore/unistore/tikv/mvcc_test.go

@@ -690,6 +690,15 @@ func TestCheckTxnStatus(t *testing.T) {
 	require.Equal(t, uint64(41), resCommitTs)
 	require.NoError(t, err)
 	require.Equal(t, kvrpcpb.Action_NoAction, action)
+
+	// check on mismatching primary
+	startTs = 43
+	callerStartTs = 44
+	currentTs = 44
+	MustAcquirePessimisticLock([]byte("another_key"), pk, startTs, startTs, store)
+	_, _, _, err = CheckTxnStatus(pk, startTs, callerStartTs, currentTs, true, store)
+	require.IsType(t, &kverrors.ErrPrimaryMismatch{}, errors.Cause(err))
+	MustPessimisticRollback(pk, startTs, startTs, store)
 }
 
 func TestCheckSecondaryLocksStatus(t *testing.T) {

store/mockstore/unistore/tikv/server.go

@@ -1035,6 +1035,25 @@ func convertToKeyError(err error) *kvrpcpb.KeyError {
 				PrimaryKey: x.PrimaryKey,
 			},
 		}
+<<<<<<< HEAD
+=======
+	case *kverrors.ErrAssertionFailed:
+		return &kvrpcpb.KeyError{
+			AssertionFailed: &kvrpcpb.AssertionFailed{
+				StartTs:          x.StartTS,
+				Key:              x.Key,
+				Assertion:        x.Assertion,
+				ExistingStartTs:  x.ExistingStartTS,
+				ExistingCommitTs: x.ExistingCommitTS,
+			},
+		}
+	case *kverrors.ErrPrimaryMismatch:
+		return &kvrpcpb.KeyError{
+			PrimaryMismatch: &kvrpcpb.PrimaryMismatch{
+				LockInfo: x.Lock.ToLockInfo(x.Key),
+			},
+		}
+>>>>>>> 9b9796fc5c8 (unistore: Adjust some behaviors to be consistent with TiKV (#43397))
 	default:
 		return &kvrpcpb.KeyError{
 			Abort: err.Error(),

store/mockstore/unistore/tikv/write.go

@@ -24,9 +24,15 @@ import (
 	"github.com/pingcap/badger"
 	"github.com/pingcap/badger/y"
 	"github.com/pingcap/kvproto/pkg/kvrpcpb"
+	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/store/mockstore/unistore/lockstore"
 	"github.com/pingcap/tidb/store/mockstore/unistore/tikv/dbreader"
 	"github.com/pingcap/tidb/store/mockstore/unistore/tikv/mvcc"
+<<<<<<< HEAD
+=======
+	"github.com/pingcap/tidb/util/mathutil"
+	"go.uber.org/zap"
+>>>>>>> 9b9796fc5c8 (unistore: Adjust some behaviors to be consistent with TiKV (#43397))
 )
 
 const (
@@ -249,7 +255,10 @@ func (wb *writeBatch) Prewrite(key []byte, lock *mvcc.Lock) {
 func (wb *writeBatch) Commit(key []byte, lock *mvcc.Lock) {
 	userMeta := mvcc.NewDBUserMeta(wb.startTS, wb.commitTS)
 	k := y.KeyWithTs(key, wb.commitTS)
-	if lock.Op != uint8(kvrpcpb.Op_Lock) {
+	if lock.Op == uint8(kvrpcpb.Op_PessimisticLock) {
+		log.Info("removing a pessimistic lock when committing", zap.Binary("key", key), zap.Uint64("startTS", wb.startTS), zap.Uint64("commitTS", wb.commitTS))
+		// Write nothing as if PessimisticRollback is called.
+	} else if lock.Op != uint8(kvrpcpb.Op_Lock) {
 		wb.dbBatch.set(k, lock.Value, userMeta)
 	} else if bytes.Equal(key, lock.Primary) {
 		opLockKey := y.KeyWithTs(mvcc.EncodeExtraTxnStatusKey(key, wb.startTS), wb.startTS)