tikv: eliminate unnecessary pessimistic rollback (#12561)

store/tikv/txn.go

@@ -23,6 +23,7 @@ import (
 	"github.com/opentracing/opentracing-go"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
+	"github.com/pingcap/parser/terror"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/metrics"
 	"github.com/pingcap/tidb/sessionctx"
@@ -428,13 +429,17 @@ func (txn *tikvTxn) LockKeys(ctx context.Context, forUpdateTS uint64, keysInput
 			for _, key := range keys {
 				txn.us.DeleteKeyExistErrInfo(key)
 			}
-			wg := txn.asyncPessimisticRollback(ctx, keys)
-			if dl, ok := errors.Cause(err).(*ErrDeadlock); ok && hashInKeys(dl.DeadlockKeyHash, keys) {
-				dl.IsRetryable = true
-				// Wait for the pessimistic rollback to finish before we retry the statement.
-				wg.Wait()
-				// Sleep a little, wait for the other transaction that blocked by this transaction to acquire the lock.
-				time.Sleep(time.Millisecond * 5)
+			keyMayBeLocked := terror.ErrorNotEqual(kv.ErrWriteConflict, err) && terror.ErrorNotEqual(kv.ErrKeyExists, err)
+			// If there is only 1 key and lock fails, no need to do pessimistic rollback.
+			if len(keys) > 1 || keyMayBeLocked {
+				wg := txn.asyncPessimisticRollback(ctx, keys)
+				if dl, ok := errors.Cause(err).(*ErrDeadlock); ok && hashInKeys(dl.DeadlockKeyHash, keys) {
+					dl.IsRetryable = true
+					// Wait for the pessimistic rollback to finish before we retry the statement.
+					wg.Wait()
+					// Sleep a little, wait for the other transaction that blocked by this transaction to acquire the lock.
+					time.Sleep(time.Millisecond * 5)
+				}
 			}
 			if assignedPrimaryKey {
 				// unset the primary key if we assigned primary key when failed to lock it.