diff --git a/internal/client/client_collapse.go b/internal/client/client_collapse.go index 4e753e92a0..9875389ffa 100644 --- a/internal/client/client_collapse.go +++ b/internal/client/client_collapse.go @@ -129,6 +129,7 @@ func (r reqCollapse) tryCollapseRequest(ctx context.Context, addr string, req *t } func resolveLockCollapseKey(req *tikvrpc.Request) string { + // IsTxnFile is implied by StartVersion, so it does not need to be part of the collapse key. resolveLock := req.ResolveLock() return strconv.FormatUint(req.RegionId, 10) + "-" + strconv.FormatUint(resolveLock.StartVersion, 10) + "-" + diff --git a/txnkv/txnlock/lock_resolver.go b/txnkv/txnlock/lock_resolver.go index 15c3d74e46..92321ad634 100644 --- a/txnkv/txnlock/lock_resolver.go +++ b/txnkv/txnlock/lock_resolver.go @@ -208,6 +208,7 @@ type Lock struct { UseAsyncCommit bool LockForUpdateTS uint64 MinCommitTS uint64 + IsTxnFile bool } func (l *Lock) IsPessimistic() bool { @@ -226,8 +227,8 @@ func (l *Lock) String() string { buf.WriteString(redact.Key(l.Key)) buf.WriteString(", primary: ") buf.WriteString(redact.Key(l.Primary)) - return fmt.Sprintf("%s, txnStartTS: %d, lockForUpdateTS:%d, minCommitTs:%d, ttl: %d, type: %s, UseAsyncCommit: %t, txnSize: %d", - buf.String(), l.TxnID, l.LockForUpdateTS, l.MinCommitTS, l.TTL, l.LockType, l.UseAsyncCommit, l.TxnSize) + return fmt.Sprintf("%s, txnStartTS: %d, lockForUpdateTS:%d, minCommitTs:%d, ttl: %d, type: %s, UseAsyncCommit: %t, txnSize: %d, isTxnFile: %t", + buf.String(), l.TxnID, l.LockForUpdateTS, l.MinCommitTS, l.TTL, l.LockType, l.UseAsyncCommit, l.TxnSize, l.IsTxnFile) } // NewLock creates a new *Lock. @@ -242,6 +243,7 @@ func NewLock(l *kvrpcpb.LockInfo) *Lock { UseAsyncCommit: l.UseAsyncCommit, LockForUpdateTS: l.LockForUpdateTs, MinCommitTS: l.MinCommitTs, + IsTxnFile: l.IsTxnFile, } } @@ -296,6 +298,7 @@ func (lr *LockResolver) BatchResolveLocks(bo *retry.Backoffer, locks []*Lock, lo expiredLocks := locks txnInfos := make(map[uint64]uint64) + txnFileIDs := make(map[uint64]bool) startTime := time.Now() for _, l := range expiredLocks { logutil.Logger(bo.GetCtx()).Debug("BatchResolveLocks handling lock", zap.Stringer("lock", l)) @@ -339,6 +342,9 @@ func (lr *LockResolver) BatchResolveLocks(bo *retry.Backoffer, locks []*Lock, lo resolveData, err := lr.checkAllSecondaries(bo, l, &status) if err == nil { txnInfos[l.TxnID] = resolveData.commitTs + if l.IsTxnFile { + txnFileIDs[l.TxnID] = true + } continue } if _, ok := errors.Cause(err).(*nonAsyncCommitLock); ok { @@ -357,6 +363,9 @@ func (lr *LockResolver) BatchResolveLocks(bo *retry.Backoffer, locks []*Lock, lo } txnInfos[l.TxnID] = status.commitTS + if l.IsTxnFile { + txnFileIDs[l.TxnID] = true + } } logutil.BgLogger().Info("BatchResolveLocks: lookup txn status", zap.Duration("cost time", time.Since(startTime)), @@ -365,8 +374,9 @@ func (lr *LockResolver) BatchResolveLocks(bo *retry.Backoffer, locks []*Lock, lo listTxnInfos := make([]*kvrpcpb.TxnInfo, 0, len(txnInfos)) for txnID, status := range txnInfos { listTxnInfos = append(listTxnInfos, &kvrpcpb.TxnInfo{ - Txn: txnID, - Status: status, + Txn: txnID, + Status: status, + IsTxnFile: txnFileIDs[txnID], }) } @@ -1008,6 +1018,7 @@ func (lr *LockResolver) getTxnStatus(bo *retry.Backoffer, txnID uint64, primary var status TxnStatus resolvingPessimisticLock := lockInfo != nil && lockInfo.IsPessimistic() + isTxnFile := lockInfo != nil && lockInfo.IsTxnFile req := tikvrpc.NewRequest(tikvrpc.CmdCheckTxnStatus, &kvrpcpb.CheckTxnStatusRequest{ PrimaryKey: primary, LockTs: txnID, @@ -1017,6 +1028,7 @@ func (lr *LockResolver) getTxnStatus(bo *retry.Backoffer, txnID uint64, primary ForceSyncCommit: forceSyncCommit, ResolvingPessimisticLock: resolvingPessimisticLock, VerifyIsPrimary: true, + IsTxnFile: isTxnFile, }, kvrpcpb.Context{ RequestSource: util.RequestSourceFromCtx(bo.GetCtx()), ResourceControlContext: &kvrpcpb.ResourceControlContext{ @@ -1460,6 +1472,7 @@ func (lr *LockResolver) batchLiteResolveLocks(bo *retry.Backoffer, l *Lock, keys func (lr *LockResolver) resolveRegionLocks(bo *retry.Backoffer, l *Lock, region locate.RegionVerID, keys [][]byte, status TxnStatus) error { lreq := &kvrpcpb.ResolveLockRequest{ StartVersion: l.TxnID, + IsTxnFile: l.IsTxnFile, } if status.IsCommitted() { lreq.CommitVersion = status.CommitTS() @@ -1560,6 +1573,7 @@ func (lr *LockResolver) resolveLock(bo *retry.Backoffer, l *Lock, status TxnStat } lreq := &kvrpcpb.ResolveLockRequest{ StartVersion: l.TxnID, + IsTxnFile: l.IsTxnFile, } if status.IsCommitted() { lreq.CommitVersion = status.CommitTS()