lock_resolver: avoid pessimistic transactions using resolveLocksForWr…

…ite (#25973)

store/tikv/lock_resolver.go

@@ -433,8 +433,10 @@ func (lr *LockResolver) resolveLocks(bo *Backoffer, callerStartTS uint64, locks
 	return msBeforeTxnExpired.value(), pushed, nil
 }
 
-func (lr *LockResolver) resolveLocksForWrite(bo *Backoffer, callerStartTS uint64, locks []*Lock) (int64, error) {
-	msBeforeTxnExpired, _, err := lr.resolveLocks(bo, callerStartTS, locks, true, false)
+func (lr *LockResolver) resolveLocksForWrite(bo *Backoffer, callerStartTS, callerForUpdateTS uint64, locks []*Lock) (int64, error) {
+	// The forWrite parameter is only useful for optimistic transactions which can avoid deadlock between large transactions,
+	// so only use forWrite if the callerForUpdateTS is zero.
+	msBeforeTxnExpired, _, err := lr.resolveLocks(bo, callerStartTS, locks, callerForUpdateTS == 0, false)
 	return msBeforeTxnExpired, err
 }
 

store/tikv/prewrite.go

@@ -293,7 +293,7 @@ func (action actionPrewrite) handleSingleBatch(c *twoPhaseCommitter, bo *Backoff
 			locks = append(locks, lock)
 		}
 		start := time.Now()
-		msBeforeExpired, err := c.store.lockResolver.resolveLocksForWrite(bo, c.startTS, locks)
+		msBeforeExpired, err := c.store.lockResolver.resolveLocksForWrite(bo, c.startTS, c.forUpdateTS, locks)
 		if err != nil {
 			return errors.Trace(err)
 		}

store/tikv/tests/async_commit_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/pingcap/tidb/store/mockstore/unistore"
 	"github.com/pingcap/tidb/store/tikv"
 	tikverr "github.com/pingcap/tidb/store/tikv/error"
+	"github.com/pingcap/tidb/store/tikv/kv"
 	"github.com/pingcap/tidb/store/tikv/mockstore/cluster"
 	"github.com/pingcap/tidb/store/tikv/oracle"
 	"github.com/pingcap/tidb/store/tikv/tikvrpc"
@@ -178,6 +179,7 @@ func (s *testAsyncCommitSuite) lockKeysWithAsyncCommit(c *C, keys, values [][]by
 	tpc, err := txnProbe.NewCommitter(0)
 	c.Assert(err, IsNil)
 	tpc.SetPrimaryKey(primaryKey)
+	tpc.SetUseAsyncCommit()
 
 	ctx := context.Background()
 	err = tpc.PrewriteAllMutations(ctx)
@@ -539,3 +541,24 @@ func (m *mockResolveClient) SendRequest(ctx context.Context, addr string, req *t
 func (m *mockResolveClient) Close() error {
 	return m.inner.Close()
 }
+
+// TestPessimisticTxnResolveAsyncCommitLock tests that pessimistic transactions resolve non-expired async-commit locks during the prewrite phase.
+// Pessimistic transactions will resolve locks immediately during the prewrite phase because of the special logic for handling non-pessimistic lock conflict.
+// However, async-commit locks can't be resolved until they expire. This test covers it.
+func (s *testAsyncCommitSuite) TestPessimisticTxnResolveAsyncCommitLock(c *C) {
+	ctx := context.Background()
+	k := []byte("k")
+
+	txn, err := s.store.Begin()
+	c.Assert(err, IsNil)
+	txn.SetPessimistic(true)
+	err = txn.LockKeys(ctx, &kv.LockCtx{ForUpdateTS: txn.StartTS()}, []byte("k1"))
+	c.Assert(err, IsNil)
+
+	// Lock the key with a async-commit lock.
+	s.lockKeysWithAsyncCommit(c, [][]byte{}, [][]byte{}, k, k, false)
+
+	txn.Set(k, k)
+	err = txn.Commit(context.Background())
+	c.Assert(err, IsNil)
+}