Repository: tikv/client-go
Pull request #939 (2023-08-11 09:47:46)

merge master into cse region client


Patch:
diff --git a/.github/workflows/compatibility.yml b/.github/workflows/compatibility.yml
index 0f09dec..ec54405 100644
--- a/.github/workflows/compatibility.yml
+++ b/.github/workflows/compatibility.yml
@@ -12,9 +12,9 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: Set up Go
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v4
       with:
-        go-version: "1.20.x"
+        go-version: "1.21.0"
 
     - name: Checkout Client-Go
       uses: actions/checkout@v2
diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml
index 1656293..045f067 100644
--- a/.github/workflows/integration.yml
+++ b/.github/workflows/integration.yml
@@ -15,9 +15,9 @@ jobs:
         uses: actions/checkout@v2
 
       - name: Set up Go
-        uses: actions/setup-go@v2
+        uses: actions/setup-go@v4
         with:
-          go-version: 1.20.2
+          go-version: 1.21.0
 
       - name: Test
         run: go test ./...
@@ -30,9 +30,9 @@ jobs:
         uses: actions/checkout@v2
 
       - name: Set up Go
-        uses: actions/setup-go@v2
+        uses: actions/setup-go@v4
         with:
-          go-version: 1.20.2
+          go-version: 1.21.0
 
       - name: Test
         run: go test ./... -race
@@ -45,9 +45,9 @@ jobs:
         uses: actions/checkout@v2
 
       - name: Set up Go
-        uses: actions/setup-go@v2
+        uses: actions/setup-go@v4
         with:
-          go-version: 1.20.2
+          go-version: 1.21.0
 
       - name: Fetch PD
         uses: shrink/actions-docker-extract@v1
@@ -87,9 +87,9 @@ jobs:
         uses: actions/checkout@v2
 
       - name: Set up Go
-        uses: actions/setup-go@v2
+        uses: actions/setup-go@v4
         with:
-          go-version: 1.20.2
+          go-version: 1.21.0
 
       - name: Fetch PD
         uses: shrink/actions-docker-extract@v1
diff --git a/config/client.go b/config/client.go
index a57fd0c..b29c7b9 100644
--- a/config/client.go
+++ b/config/client.go
@@ -81,6 +81,9 @@ type TiKVClient struct {
 	// StoreLivenessTimeout is the timeout for store liveness check request.
 	StoreLivenessTimeout string           `toml:"store-liveness-timeout" json:"store-liveness-timeout"`
 	CoprCache            CoprocessorCache `toml:"copr-cache" json:"copr-cache"`
+	// CoprReqTimeout is the timeout for a single coprocessor request
+	// Note: this is a transitional modification, and it will be removed if it's dynamic configurable version is ready.
+	CoprReqTimeout time.Duration `toml:"copr-req-timeout" json:"copr-req-timeout"`
 	// TTLRefreshedTxnSize controls whether a transaction should update its TTL or not.
 	TTLRefreshedTxnSize      int64  `toml:"ttl-refreshed-txn-size" json:"ttl-refreshed-txn-size"`
 	ResolveLockLiteThreshold uint64 `toml:"resolve-lock-lite-threshold" json:"resolve-lock-lite-threshold"`
@@ -153,6 +156,7 @@ func DefaultTiKVClient() TiKVClient {
 			AdmissionMaxResultMB:  10,
 			AdmissionMinProcessMs: 5,
 		},
+		CoprReqTimeout: 60 * time.Second,
 
 		ResolveLockLiteThreshold: 16,
 	}
diff --git a/error/error.go b/error/error.go
index 761ebef..62a75ff 100644
--- a/error/error.go
+++ b/error/error.go
@@ -246,7 +246,7 @@ type ErrAssertionFailed struct {
 	*kvrpcpb.AssertionFailed
 }
 
-// ErrLockOnlyIfExistsNoReturnValue is used when the flag `LockOnlyIfExists` of `LockCtx` is set, but `ReturnValues“ is not.
+// ErrLockOnlyIfExistsNoReturnValue is used when the flag `LockOnlyIfExists` of `LockCtx` is set, but `ReturnValues` is not.
 type ErrLockOnlyIfExistsNoReturnValue struct {
 	StartTS     uint64
 	ForUpdateTs uint64
diff --git a/examples/gcworker/gcworker.go b/examples/gcworker/gcworker.go
index 39da00d..1deafdd 100644
--- a/examples/gcworker/gcworker.go
+++ b/examples/gcworker/gcworker.go
@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"github.com/tikv/client-go/v2/oracle"
+	"github.com/tikv/client-go/v2/tikv"
 	"github.com/tikv/client-go/v2/txnkv"
 )
 
@@ -36,10 +37,14 @@ func main() {
 		panic(err)
 	}
 
-	sysSafepoint, err := client.GC(context.Background(), *safepoint)
+	sysSafepoint, err := client.GC(context.Background(), *safepoint, tikv.WithConcurrency(10))
 	if err != nil {
 		panic(err)
 	}
 	fmt.Printf("Finished GC, expect safepoint:%d(%+v),real safepoint:%d(+%v)\n", *safepoint, oracle.GetTimeFromTS(*safepoint), sysSafepoint, oracle.GetTimeFromTS(sysSafepoint))
 
+	err = client.Close()
+	if err != nil {
+		panic(err)
+	}
 }
diff --git a/go.mod b/go.mod
index 18a28ab..f8695bf 100644
--- a/go.mod
+++ b/go.mod
@@ -14,14 +14,14 @@ require (
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c
 	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c
 	github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989
-	github.com/pingcap/kvproto v0.0.0-20230713060620-89756bd21be1
+	github.com/pingcap/kvproto v0.0.0-20230720094213-a3b4a77b4333
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.15.1
 	github.com/prometheus/client_model v0.4.0
 	github.com/stretchr/testify v1.8.2
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a
-	github.com/tikv/pd/client v0.0.0-20230717031845-58eb48b840e7
+	github.com/tikv/pd/client v0.0.0-20230724080549-de985b8e0afc
 	github.com/twmb/murmur3 v1.1.3
 	go.etcd.io/etcd/api/v3 v3.5.2
 	go.etcd.io/etcd/client/v3 v3.5.2
@@ -51,6 +51,7 @@ require (
 	github.com/rogpeppe/go-internal v1.10.0 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.2 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
+	golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 // indirect
 	golang.org/x/net v0.10.0 // indirect
 	golang.org/x/sys v0.8.0 // indirect
 	golang.org/x/text v0.9.0 // indirect
diff --git a/go.sum b/go.sum
index 3a515b6..1feda02 100644
--- a/go.sum
+++ b/go.sum
@@ -135,8 +135,8 @@ github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c h1:CgbKAHto5CQgW
 github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c/go.mod h1:4qGtCB0QK0wBzKtFEGDhxXnSnbQApw1gc9siScUl8ew=
 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-20230713060620-89756bd21be1 h1:sC3XRNNBQNjFJGRtSzJRvqi2aDLFOsQoCHItr9rbbY8=
-github.com/pingcap/kvproto v0.0.0-20230713060620-89756bd21be1/go.mod h1:r0q/CFcwvyeRhKtoqzmWMBebrtpIziQQ9vR+JKh1knc=
+github.com/pingcap/kvproto v0.0.0-20230720094213-a3b4a77b4333 h1:A6Wqgq0uMw51UiRAH27TVN0QlzVR5CVtV6fTQSAmvKM=
+github.com/pingcap/kvproto v0.0.0-20230720094213-a3b4a77b4333/go.mod h1:r0q/CFcwvyeRhKtoqzmWMBebrtpIziQQ9vR+JKh1knc=
 github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 h1:HR/ylkkLmGdSSDaD8IDP+SZrdhV1Kibl9KrHxJ9eciw=
 github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -192,8 +192,8 @@ github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ
 github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a h1:J/YdBZ46WKpXsxsW93SG+q0F8KI+yFrcIDT4c/RNoc4=
 github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a/go.mod h1:h4xBhSNtOeEosLJ4P7JyKXX7Cabg7AVkWCK5gV2vOrM=
-github.com/tikv/pd/client v0.0.0-20230717031845-58eb48b840e7 h1:N1m2YFtQAZEkk5ilmsMqauELh29TJQrX2JbcNhdnm7k=
-github.com/tikv/pd/client v0.0.0-20230717031845-58eb48b840e7/go.mod h1:YAB/lbBQtURR2A1tyqGQYEkBVCKHVqyZFefN7CPIZi4=
+github.com/tikv/pd/client v0.0.0-20230724080549-de985b8e0afc h1:IUg0j2nWoGYj3FQ3vA3vg97fPSpJEZQrDpgF8RkMLEU=
+github.com/tikv/pd/client v0.0.0-20230724080549-de985b8e0afc/go.mod h1:wfHRc4iYaqJiOQZCHcrF+r4hYnkGDaYWDfcicee//pc=
 github.com/twmb/murmur3 v1.1.3 h1:D83U0XYKcHRYwYIpBKf3Pks91Z0Byda/9SJ8B6EMRcA=
 github.com/twmb/murmur3 v1.1.3/go.mod h1:Qq/R7NUyOfr65zD+6Q5IHKsJLwP7exErjN6lyyq3OSQ=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
@@ -229,6 +229,8 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
+golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 h1:QLureRX3moex6NVu/Lr4MGakp9FdA7sBHGBmvRW7NaM=
+golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4/go.mod h1:FXUEEKJgO7OQYeo8N01OfiKP8RXMtf6e8aTskBGqWdc=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
diff --git a/internal/client/client_batch.go b/internal/client/client_batch.go
index 6952bfc..6a270bc 100644
--- a/internal/client/client_batch.go
+++ b/internal/client/client_batch.go
@@ -347,6 +347,12 @@ func (a *batchConn) batchSendLoop(cfg config.TiKVClient) {
 }
 
 func (a *batchConn) getClientAndSend() {
+	if val, err := util.EvalFailpoint("mockBatchClientSendDelay"); err == nil {
+		if timeout, ok := val.(int); ok && timeout > 0 {
+			time.Sleep(time.Duration(timeout * int(time.Millisecond)))
+		}
+	}
+
 	// Choose a connection by round-robbin.
 	var (
 		cli    *batchCommandsClient
@@ -781,7 +787,7 @@ func sendBatchRequest(
 	select {
 	case batchConn.batchCommandsCh <- entry:
 	case <-ctx.Done():
-		logutil.BgLogger().Debug("send request is cancelled",
+		logutil.Logger(ctx).Debug("send request is cancelled",
 			zap.String("to", addr), zap.String("cause", ctx.Err().Error()))
 		return nil, errors.WithStack(ctx.Err())
 	case <-timer.C:
@@ -797,7 +803,7 @@ func sendBatchRequest(
 		return tikvrpc.FromBatchCommandsResponse(res)
 	case <-ctx.Done():
 		atomic.StoreInt32(&entry.canceled, 1)
-		logutil.BgLogger().Debug("wait response is cancelled",
+		logutil.Logger(ctx).Debug("wait response is cancelled",
 			zap.String("to", addr), zap.String("cause", ctx.Err().Error()))
 		return nil, errors.WithStack(ctx.Err())
 	case <-timer.C:
diff --git a/internal/client/client_interceptor.go b/internal/client/client_interceptor.go
index e490d18..64ae333 100644
--- a/internal/client/client_interceptor.go
+++ b/internal/client/client_interceptor.go
@@ -110,9 +110,12 @@ func buildResourceControlInterceptor(
 			// bypass some internal requests and it's may influence user experience. For example, the
 			// request of `alter user password`, totally bypasses the resource control. it's not cost
 			// many resources, but it's may influence the user experience.
-			if reqInfo.Bypass() {
+			// If the resource group has background jobs, we should not record consumption and wait for it.
+			// Background jobs will record and report in tikv side.
+			if reqInfo.Bypass() || resourceControlInterceptor.IsBackgroundRequest(ctx, resourceGroupName, req.RequestSource) {
 				return next(target, req)
 			}
+
 			consumption, penalty, err := resourceControlInterceptor.OnRequestWait(ctx, resourceGroupName, reqInfo)
 			if err != nil {
 				return nil, err
diff --git a/internal/locate/region_cache.go b/internal/locate/region_cache.go
index 59c413e..87cd067 100644
--- a/internal/locate/region_cache.go
+++ b/internal/locate/region_cache.go
@@ -153,6 +153,7 @@ type Region struct {
 	syncFlag      int32          // region need be sync in next turn
 	lastAccess    int64          // last region access time, see checkRegionCacheTTL
 	invalidReason InvalidReason  // the reason why the region is invalidated
+	asyncReload   atomic.Bool    // the region need to be reloaded in async mode
 }
 
 // AccessIndex represent the index for accessIndex array
@@ -177,6 +178,9 @@ type regionStore struct {
 	// buckets is not accurate and it can change even if the region is not changed.
 	// It can be stale and buckets keys can be out of the region range.
 	buckets *metapb.Buckets
+	// record all storeIDs on which pending peers reside.
+	// key is storeID, val is peerID.
+	pendingTiFlashPeerStores map[uint64]uint64
 }
 
 func (r *regionStore) accessStore(mode accessMode, idx AccessIndex) (int, *Store) {
@@ -269,11 +273,12 @@ func newRegion(bo *retry.Backoffer, c *RegionCache, pdRegion *pd.Region) (*Regio
 	// regionStore pull used store from global store map
 	// to avoid acquire storeMu in later access.
 	rs := &regionStore{
-		workTiKVIdx:  0,
-		proxyTiKVIdx: -1,
-		stores:       make([]*Store, 0, len(r.meta.Peers)),
-		storeEpochs:  make([]uint32, 0, len(r.meta.Peers)),
-		buckets:      pdRegion.Buckets,
+		workTiKVIdx:              0,
+		proxyTiKVIdx:             -1,
+		stores:                   make([]*Store, 0, len(r.meta.Peers)),
+		pendingTiFlashPeerStores: map[uint64]uint64{},
+		storeEpochs:              make([]uint32, 0, len(r.meta.Peers)),
+		buckets:                  pdRegion.Buckets,
 	}
 
 	leader := pdRegion.Leader
@@ -314,6 +319,11 @@ func newRegion(bo *retry.Backoffer, c *RegionCache, pdRegion *pd.Region) (*Regio
 		}
 		rs.stores = append(rs.stores, store)
 		rs.storeEpochs = append(rs.storeEpochs, atomic.LoadUint32(&store.epoch))
+		for _, pendingPeer := range pdRegion.PendingPeers {
+			if pendingPeer.Id == p.Id {
+				rs.pendingTiFlashPeerStores[store.storeID] = p.Id
+			}
+		}
 	}
 	// TODO(youjiali1995): It's possible the region info in PD is stale for now but it can recover.
 	// Maybe we need backoff here.
@@ -411,7 +421,7 @@ func newRegionIndexMu(rs []*Region) *regionIndexMu {
 	r.latestVersions = make(map[uint64]RegionVerID)
 	r.sorted = NewSortedRegions(btreeDegree)
 	for _, region := range rs {
-		r.insertRegionToCache(region)
+		r.insertRegionToCache(region, true)
 	}
 	return r
 }
@@ -457,6 +467,11 @@ type RegionCache struct {
 		// requestLiveness always returns unreachable.
 		mockRequestLiveness atomic.Pointer[livenessFunc]
 	}
+
+	regionsNeedReload struct {
+		sync.Mutex
+		regions []uint64
+	}
 }
 
 // NewRegionCache creates a RegionCache.
@@ -510,8 +525,8 @@ func (c *RegionCache) clear() {
 }
 
 // thread unsafe, should use with lock
-func (c *RegionCache) insertRegionToCache(cachedRegion *Region) {
-	c.mu.insertRegionToCache(cachedRegion)
+func (c *RegionCache) insertRegionToCache(cachedRegion *Region, invalidateOldRegion bool) {
+	c.mu.insertRegionToCache(cachedRegion, invalidateOldRegion)
 }
 
 // Close releases region cache's resource.
@@ -522,8 +537,13 @@ func (c *RegionCache) Close() {
 // asyncCheckAndResolveLoop with
 func (c *RegionCache) asyncCheckAndResolveLoop(interval time.Duration) {
 	ticker := time.NewTicker(interval)
-	defer ticker.Stop()
+	reloadRegionTicker := time.NewTicker(10 * time.Second)
+	defer func() {
+		ticker.Stop()
+		reloadRegionTicker.Stop()
+	}()
 	var needCheckStores []*Store
+	reloadNextLoop := make(map[uint64]struct{})
 	for {
 		needCheckStores = needCheckStores[:0]
 		select {
@@ -541,6 +561,21 @@ func (c *RegionCache) asyncCheckAndResolveLoop(interval time.Duration) {
 				// there's a deleted store in the stores map which guaranteed by reReslve().
 				return state != unresolved && state != tombstone && state != deleted
 			})
+		case <-reloadRegionTicker.C:
+			for regionID := range reloadNextLoop {
+				c.reloadRegion(regionID)
+				delete(reloadNextLoop, regionID)
+			}
+			c.regionsNeedReload.Lock()
+			for _, regionID := range c.regionsNeedReload.regions {
+				// will reload in next tick, wait a while for two reasons:
+				// 1. there may an unavailable duration while recreating the connection.
+				// 2. the store may just be started, and wait safe ts synced to avoid the
+				// possible dataIsNotReady error.
+				reloadNextLoop[regionID] = struct{}{}
+			}
+			c.regionsNeedReload.regions = c.regionsNeedReload.regions[:0]
+			c.regionsNeedReload.Unlock()
 		}
 	}
 }
@@ -741,7 +776,7 @@ func (c *RegionCache) GetTiKVRPCContext(bo *retry.Backoffer, id RegionVerID, rep
 	storeFailEpoch := atomic.LoadUint32(&store.epoch)
 	if storeFailEpoch != regionStore.storeEpochs[storeIdx] {
 		cachedRegion.invalidate(Other)
-		logutil.BgLogger().Info("invalidate current region, because others failed on same store",
+		logutil.Logger(bo.GetCtx()).Info("invalidate current region, because others failed on same store",
 			zap.Uint64("region", id.GetID()),
 			zap.String("store", store.addr))
 		return nil, nil
@@ -780,23 +815,26 @@ func (c *RegionCache) GetTiKVRPCContext(bo *retry.Backoffer, id RegionVerID, rep
 }
 
 // GetAllValidTiFlashStores returns the store ids of all valid TiFlash stores, the store id of currentStore is always the first one
-func (c *RegionCache) GetAllValidTiFlashStores(id RegionVerID, currentStore *Store, labelFilter LabelFilter) []uint64 {
+// Caller may use `nonPendingStores` first, this can avoid task need to wait tiflash replica syncing from tikv.
+// But if all tiflash peers are pending(len(nonPendingStores) == 0), use `allStores` is also ok.
+func (c *RegionCache) GetAllValidTiFlashStores(id RegionVerID, currentStore *Store, labelFilter LabelFilter) ([]uint64, []uint64) {
 	// set the cap to 2 because usually, TiFlash table will have 2 replicas
 	allStores := make([]uint64, 0, 2)
+	nonPendingStores := make([]uint64, 0, 2)
 	// make sure currentStore id is always the first in allStores
 	allStores = append(allStores, currentStore.storeID)
 	ts := time.Now().Unix()
 	cachedRegion := c.GetCachedRegionWithRLock(id)
 	if cachedRegion == nil {
-		return allStores
+		return allStores, nonPendingStores
 	}
 	if !cachedRegion.checkRegionCacheTTL(ts) {
-		return allStores
+		return allStores, nonPendingStores
 	}
 	regionStore := cachedRegion.getStore()
 	currentIndex := regionStore.getAccessIndex(tiFlashOnly, currentStore)
 	if currentIndex == -1 {
-		return allStores
+		return allStores, nonPendingStores
 	}
 	for startOffset := 1; startOffset < regionStore.accessStoreNum(tiFlashOnly); startOffset++ {
 		accessIdx := AccessIndex((int(currentIndex) + startOffset) % regionStore.accessStoreNum(tiFlashOnly))
@@ -813,7 +851,12 @@ func (c *RegionCache) GetAllValidTiFlashStores(id RegionVerID, currentStore *Sto
 		}
 		allStores = append(allStores, store.storeID)
 	}
-	return allStores
+	for _, storeID := range allStores {
+		if _, ok := regionStore.pendingTiFlashPeerStores[storeID]; !ok {
+			nonPendingStores = append(nonPendingStores, storeID)
+		}
+	}
+	return allStores, nonPendingStores
 }
 
 // GetTiFlashRPCContext returns RPCContext for a region must access flash store. If it returns nil, the region
@@ -862,7 +905,7 @@ func (c *RegionCache) GetTiFlashRPCContext(bo *retry.Backoffer, id RegionVerID,
 		storeFailEpoch := atomic.LoadUint32(&store.epoch)
 		if storeFailEpoch != regionStore.storeEpochs[storeIdx] {
 			cachedRegion.invalidate(Other)
-			logutil.BgLogger().Info("invalidate current region, because others failed on same store",
+			logutil.Logger(bo.GetCtx()).Info("invalidate current region, because others failed on same store",
 				zap.Uint64("region", id.GetID()),
 				zap.String("store", store.addr))
 			// TiFlash will always try to find out a valid peer, avoiding to retry too many times.
@@ -1043,7 +1086,7 @@ func (c *RegionCache) findRegionByKey(bo *retry.Backoffer, key []byte, isEndKey
 		logutil.Eventf(bo.GetCtx(), "load region %d from pd, due to cache-miss", lr.GetID())
 		r = lr
 		c.mu.Lock()
-		c.insertRegionToCache(r)
+		c.insertRegionToCache(r, true)
 		c.mu.Unlock()
 	} else if r.checkNeedReloadAndMarkUpdated() {
 		// load region when it be marked as need reload.
@@ -1056,7 +1099,7 @@ func (c *RegionCache) findRegionByKey(bo *retry.Backoffer, key []byte, isEndKey
 			logutil.Eventf(bo.GetCtx(), "load region %d from pd, due to need-reload", lr.GetID())
 			r = lr
 			c.mu.Lock()
-			c.insertRegionToCache(r)
+			c.insertRegionToCache(r, true)
 			c.mu.Unlock()
 		}
 	}
@@ -1197,7 +1240,7 @@ func (c *RegionCache) LocateRegionByID(bo *retry.Backoffer, regionID uint64) (*K
 			} else {
 				r = lr
 				c.mu.Lock()
-				c.insertRegionToCache(r)
+				c.insertRegionToCache(r, true)
 				c.mu.Unlock()
 			}
 		}
@@ -1216,7 +1259,7 @@ func (c *RegionCache) LocateRegionByID(bo *retry.Backoffer, regionID uint64) (*K
 	}
 
 	c.mu.Lock()
-	c.insertRegionToCache(r)
+	c.insertRegionToCache(r, true)
 	c.mu.Unlock()
 	return &KeyLocation{
 		Region:   r.VerID(),
@@ -1226,6 +1269,36 @@ func (c *RegionCache) LocateRegionByID(bo *retry.Backoffer, regionID uint64) (*K
 	}, nil
 }
 
+func (c *RegionCache) scheduleReloadRegion(region *Region) {
+	if region == nil || !region.asyncReload.CompareAndSwap(false, true) {
+		// async reload scheduled by other thread.
+		return
+	}
+	regionID := region.GetID()
+	if regionID > 0 {
+		c.regionsNeedReload.Lock()
+		c.regionsNeedReload.regions = append(c.regionsNeedReload.regions, regionID)
+		c.regionsNeedReload.Unlock()
+	}
+}
+
+func (c *RegionCache) reloadRegion(regionID uint64) {
+	bo := retry.NewNoopBackoff(context.Background())
+	lr, err := c.loadRegionByID(bo, regionID)
+	if err != nil {
+		// ignore error and use old region info.
+		logutil.Logger(bo.GetCtx()).Error("load region failure",
+			zap.Uint64("regionID", regionID), zap.Error(err))
+		if oldRegion := c.getRegionByIDFromCache(regionID); oldRegion != nil {
+			oldRegion.asyncReload.Store(false)
+		}
+		return
+	}
+	c.mu.Lock()
+	c.insertRegionToCache(lr, false)
+	c.mu.Unlock()
+}
+
 // GroupKeysByRegion separates keys into groups by their belonging Regions.
 // Specially it also returns the first key's region which may be used as the
 // 'PrimaryLockKey' and should be committed ahead of others.
@@ -1300,7 +1373,7 @@ func (c *RegionCache) BatchLoadRegionsWithKeyRange(bo *retry.Backoffer, startKey
 		return
 	}
 	if len(regions) == 0 {
-		err = errors.New("PD returned no region")
+		err = errors.Errorf("PD returned no region, startKey: %q, endKey: %q", util.HexRegionKeyStr(startKey), util.HexRegionKeyStr(endKey))
 		return
 	}
 
@@ -1310,7 +1383,7 @@ func (c *RegionCache) BatchLoadRegionsWithKeyRange(bo *retry.Backoffer, startKey
 	// TODO(youjiali1995): scanRegions always fetch regions from PD and these regions don't contain buckets information
 	// for less traffic, so newly inserted regions in region cache don't have buckets information. We should improve it.
 	for _, region := range regions {
-		c.insertRegionToCache(region)
+		c.insertRegionToCache(region, true)
 	}
 
 	return
@@ -1384,7 +1457,9 @@ func (mu *regionIndexMu) removeVersionFromCache(oldVer RegionVerID, regionID uin
 
 // insertRegionToCache tries to insert the Region to cache.
 // It should be protected by c.mu.l.Lock().
-func (mu *regionIndexMu) insertRegionToCache(cachedRegion *Region) {
+// if `invalidateOldRegion` is false, the old region cache should be still valid,
+// and it may still be used by some kv requests.
+func (mu *regionIndexMu) insertRegionToCache(cachedRegion *Region, invalidateOldRegion bool) {
 	oldRegion := mu.sorted.ReplaceOrInsert(cachedRegion)
 	if oldRegion != nil {
 		store := cachedRegion.getStore()
@@ -1399,8 +1474,11 @@ func (mu *regionIndexMu) insertRegionToCache(cachedRegion *Region) {
 		if InvalidReason(atomic.LoadInt32((*int32)(&oldRegion.invalidReason))) == NoLeader {
 			store.workTiKVIdx = (oldRegionStore.workTiKVIdx + 1) % AccessIndex(store.accessStoreNum(tiKVOnly))
 		}
-		// Invalidate the old region in case it's not invalidated and some requests try with the stale region information.
-		oldRegion.invalidate(Other)
+		// If the old region is still valid, do not invalidate it to avoid unnecessary backoff.
+		if invalidateOldRegion {
+			// Invalidate the old region in case it's not invalidated and some requests try with the stale region information.
+			oldRegion.invalidate(Other)
+		}
 		// Don't refresh TiFlash work idx for region. Otherwise, it will always goto a invalid store which
 		// is under transferring regions.
 		store.workTiFlashIdx.Store(oldRegionStore.workTiFlashIdx.Load())
@@ -1705,12 +1783,12 @@ func (c *RegionCache) scanRegions(bo *retry.Backoffer, startKey, endKey []byte,
 		metrics.LoadRegionCacheHistogramWithRegions.Observe(time.Since(start).Seconds())
 		if err != nil {
 			if apicodec.IsDecodeError(err) {
-				return nil, errors.Errorf("failed to decode region range key, startKey: %q, limit: %q, err: %v", util.HexRegionKeyStr(startKey), limit, err)
+				return nil, errors.Errorf("failed to decode region range key, startKey: %q, limit: %d, err: %v", util.HexRegionKeyStr(startKey), limit, err)
 			}
 			metrics.RegionCacheCounterWithScanRegionsError.Inc()
 			backoffErr = errors.Errorf(
-				"scanRegion from PD failed, startKey: %q, limit: %q, err: %v",
-				startKey,
+				"scanRegion from PD failed, startKey: %q, limit: %d, err: %v",
+				util.HexRegionKeyStr(startKey),
 				limit,
 				err)
 			continue
@@ -1719,7 +1797,10 @@ func (c *RegionCache) scanRegions(bo *retry.Backoffer, startKey, endKey []byte,
 		metrics.RegionCacheCounterWithScanRegionsOK.Inc()
 
 		if len(regionsInfo) == 0 {
-			return nil, errors.New("PD returned no region")
+			return nil, errors.Errorf(
+				"PD returned no region, startKey: %q, endKey: %q, limit: %d",
+				util.HexRegionKeyStr(startKey), util.HexRegionKeyStr(endKey), limit,
+			)
 		}
 		regions := make([]*Region, 0, len(regionsInfo))
 		for _, r := range regionsInfo {
@@ -1878,7 +1959,7 @@ func (c *RegionCache) OnRegionEpochNotMatch(bo *retry.Backoffer, ctx *RPCContext
 			(meta.GetRegionEpoch().GetConfVer() < ctx.Region.confVer ||
 				meta.GetRegionEpoch().GetVersion() < ctx.Region.ver) {
 			err := errors.Errorf("region epoch is ahead of tikv. rpc ctx: %+v, currentRegions: %+v", ctx, currentRegions)
-			logutil.BgLogger().Info("region epoch is ahead of tikv", zap.Error(err))
+			logutil.Logger(bo.GetCtx()).Info("region epoch is ahead of tikv", zap.Error(err))
 			return true, bo.Backoff(retry.BoRegionMiss, err)
 		}
 	}
@@ -1919,7 +2000,7 @@ func (c *RegionCache) OnRegionEpochNotMatch(bo *retry.Backoffer, ctx *RPCContext
 
 	c.mu.Lock()
 	for _, region := range newRegions {
-		c.insertRegionToCache(region)
+		c.insertRegionToCache(region, true)
 	}
 	c.mu.Unlock()
 
@@ -2037,7 +2118,7 @@ func (c *RegionCache) UpdateBucketsIfNeeded(regionID RegionVerID, latestBucketsV
 				return
 			}
 			c.mu.Lock()
-			c.insertRegionToCache(new)
+			c.insertRegionToCache(new, true)
 			c.mu.Unlock()
 		}()
 	}
@@ -2392,6 +2473,24 @@ const (
 	tombstone
 )
 
+// String implements fmt.Stringer interface.
+func (s resolveState) String() string {
+	switch s {
+	case unresolved:
+		return "unresolved"
+	case resolved:
+		return "resolved"
+	case needCheck:
+		return "needCheck"
+	case deleted:
+		return "deleted"
+	case tombstone:
+		return "tombstone"
+	default:
+		return fmt.Sprintf("unknown-%v", uint64(s))
+	}
+}
+
 // IsTiFlash returns true if the storeType is TiFlash
 func (s *Store) IsTiFlash() bool {
 	return s.storeType == tikvrpc.TiFlash
@@ -2507,8 +2606,8 @@ func (s *Store) reResolve(c *RegionCache) (bool, error) {
 }
 
 func (s *Store) getResolveState() resolveState {
-	var state resolveState
 	if s == nil {
+		var state resolveState
 		return state
 	}
 	return resolveState(atomic.LoadUint64(&s.state))
@@ -2531,6 +2630,12 @@ func (s *Store) changeResolveStateTo(from, to resolveState) bool {
 			return false
 		}
 		if atomic.CompareAndSwapUint64(&s.state, uint64(from), uint64(to)) {
+			logutil.BgLogger().Info("change store resolve state",
+				zap.Uint64("store", s.storeID),
+				zap.String("addr", s.addr),
+				zap.String("from", from.String()),
+				zap.String("to", to.String()),
+				zap.String("liveness-state", s.getLivenessState().String()))
 			return true
 		}
 	}
@@ -2631,6 +2736,20 @@ const (
 	unknown
 )
 
+// String implements fmt.Stringer interface.
+func (s livenessState) String() string {
+	switch s {
+	case unreachable:
+		return "unreachable"
+	case reachable:
+		return "reachable"
+	case unknown:
+		return "unknown"
+	default:
+		return fmt.Sprintf("unknown-%v", uint32(s))
+	}
+}
+
 func (s *Store) startHealthCheckLoopIfNeeded(c *RegionCache, liveness livenessState) {
 	// This mechanism doesn't support non-TiKV stores currently.
 	if s.storeType != tikvrpc.TiKV {
diff --git a/internal/locate/region_request.go b/internal/locate/region_request.go
index 299dedc..287fc17 100644
--- a/internal/locate/region_request.go
+++ b/internal/locate/region_request.go
@@ -35,6 +35,7 @@
 package locate
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"math"
@@ -245,6 +246,8 @@ type replica struct {
 	peer     *metapb.Peer
 	epoch    uint32
 	attempts int
+	// deadlineErrUsingConfTimeout indicates the replica is already tried, but the received deadline exceeded error.
+	deadlineErrUsingConfTimeout bool
 }
 
 func (r *replica) isEpochStale() bool {
@@ -260,6 +263,7 @@ type replicaSelector struct {
 	region      *Region
 	regionStore *regionStore
 	replicas    []*replica
+	labels      []*metapb.StoreLabel
 	state       selectorState
 	// replicas[targetIdx] is the replica handling the request this time
 	targetIdx AccessIndex
@@ -376,7 +380,7 @@ func (state *accessKnownLeader) onSendFailure(bo *retry.Backoffer, selector *rep
 }
 
 func (state *accessKnownLeader) onNoLeader(selector *replicaSelector) {
-	selector.state = &tryFollower{leaderIdx: state.leaderIdx, lastIdx: state.leaderIdx}
+	selector.state = &tryFollower{leaderIdx: state.leaderIdx, lastIdx: state.leaderIdx, fromOnNotLeader: true}
 }
 
 // tryFollower is the state where we cannot access the known leader
@@ -390,19 +394,23 @@ type tryFollower struct {
 	stateBase
 	leaderIdx AccessIndex
 	lastIdx   AccessIndex
+	// fromOnNotLeader indicates whether the state is changed from onNotLeader.
+	fromOnNotLeader bool
 }
 
 func (state *tryFollower) next(bo *retry.Backoffer, selector *replicaSelector) (*RPCContext, error) {
 	var targetReplica *replica
+	hasDeadlineExceededErr := false
 	// Search replica that is not attempted from the last accessed replica
 	for i := 1; i < len(selector.replicas); i++ {
 		idx := AccessIndex((int(state.lastIdx) + i) % len(selector.replicas))
+		targetReplica = selector.replicas[idx]
+		hasDeadlineExceededErr = hasDeadlineExceededErr || targetReplica.deadlineErrUsingConfTimeout
 		if idx == state.leaderIdx {
 			continue
 		}
-		targetReplica = selector.replicas[idx]
 		// Each follower is only tried once
-		if !targetReplica.isExhausted(1) {
+		if !targetReplica.isExhausted(1) && targetReplica.store.getLivenessState() != unreachable && !targetReplica.deadlineErrUsingConfTimeout {
 			state.lastIdx = idx
 			selector.targetIdx = idx
 			break
@@ -410,16 +418,33 @@ func (state *tryFollower) next(bo *retry.Backoffer, selector *replicaSelector) (
 	}
 	// If all followers are tried and fail, backoff and retry.
 	if selector.targetIdx < 0 {
+		if hasDeadlineExceededErr {
+			// when meet deadline exceeded error, do fast retry without invalidate region cache.
+			return nil, nil
+		}
 		metrics.TiKVReplicaSelectorFailureCounter.WithLabelValues("exhausted").Inc()
 		selector.invalidateRegion()
 		return nil, nil
 	}
-	return selector.buildRPCContext(bo)
+	rpcCtx, err := selector.buildRPCContext(bo)
+	if err != nil || rpcCtx == nil {
+		return nil, err
+	}
+	// If the state is changed from onNotLeader, the `replicaRead` flag should not be set as leader read would still be used.
+	if !state.fromOnNotLeader {
+		replicaRead := selector.targetIdx != state.leaderIdx
+		rpcCtx.contextPatcher.replicaRead = &replicaRead
+	}
+	disableStaleRead := false
+	rpcCtx.contextPatcher.staleRead = &disableStaleRead
+	return rpcCtx, nil
 }
 
 func (state *tryFollower) onSendSuccess(selector *replicaSelector) {
-	if !selector.region.switchWorkLeaderToPeer(selector.targetReplica().peer) {
-		panic("the store must exist")
+	if state.fromOnNotLeader {
+		if !selector.region.switchWorkLeaderToPeer(selector.targetReplica().peer) {
+			panic("the store must exist")
+		}
 	}
 }
 
@@ -533,12 +558,12 @@ func (state *tryNewProxy) onNoLeader(selector *replicaSelector) {
 type accessFollower struct {
 	stateBase
 	// If tryLeader is true, the request can also be sent to the leader when !leader.isSlow()
-	tryLeader         bool
-	isGlobalStaleRead bool
-	option            storeSelectorOp
-	leaderIdx         AccessIndex
-	lastIdx           AccessIndex
-	learnerOnly       bool
+	tryLeader   bool
+	isStaleRead bool
+	option      storeSelectorOp
+	leaderIdx   AccessIndex
+	lastIdx     AccessIndex
+	learnerOnly bool
 }
 
 func (state *accessFollower) next(bo *retry.Backoffer, selector *replicaSelector) (*RPCContext, error) {
@@ -559,12 +584,10 @@ func (state *accessFollower) next(bo *retry.Backoffer, selector *replicaSelector
 			}
 		}
 	} else {
-		// Stale Read request will retry the leader or next peer on error,
-		// if txnScope is global, we will only retry the leader by using the WithLeaderOnly option,
-		// if txnScope is local, we will retry both other peers and the leader by the strategy of replicaSelector.
-		if state.isGlobalStaleRead {
+		// Stale Read request will retry the leader only by using the WithLeaderOnly option.
+		if state.isStaleRead {
 			WithLeaderOnly()(&state.option)
-			// retry on the leader should not use stale read flag to avoid possible DataIsNotReady error as it always can serve any read
+			// retry on the leader should not use stale read flag to avoid possible DataIsNotReady error as it always can serve any read.
 			resetStaleRead = true
 		}
 		state.lastIdx++
@@ -574,29 +597,55 @@ func (state *accessFollower) next(bo *retry.Backoffer, selector *replicaSelector
 	if state.option.preferLeader {
 		state.lastIdx = state.leaderIdx
 	}
+	var offset int
+	if state.lastIdx >= 0 {
+		offset = rand.Intn(replicaSize)
+	}
+	reloadRegion := false
 	for i := 0; i < replicaSize && !state.option.leaderOnly; i++ {
-		idx := AccessIndex((int(state.lastIdx) + i) % replicaSize)
-		// If the given store is abnormal to be accessed under `ReplicaReadMixed` mode, we should choose other followers or leader
-		// as candidates to serve the Read request. Meanwhile, we should make the choice of next() meet Uniform Distribution.
-		for cnt := 0; cnt < replicaSize && !state.isCandidate(idx, selector.replicas[idx]); cnt++ {
-			idx = AccessIndex((int(idx) + rand.Intn(replicaSize)) % replicaSize)
+		var idx AccessIndex
+		if state.option.preferLeader {
+			if i == 0 {
+				idx = state.lastIdx
+			} else {
+				// randomly select next replica, but skip state.lastIdx
+				if (i+offset)%replicaSize == 0 {
+					offset++
+				}
+			}
+		} else {
+			idx = AccessIndex((int(state.lastIdx) + i) % replicaSize)
 		}
-		if state.isCandidate(idx, selector.replicas[idx]) {
+		selectReplica := selector.replicas[idx]
+		if state.isCandidate(idx, selectReplica) {
 			state.lastIdx = idx
 			selector.targetIdx = idx
 			break
 		}
+		if selectReplica.isEpochStale() &&
+			selectReplica.store.getResolveState() == resolved &&
+			selectReplica.store.getLivenessState() == reachable {
+			reloadRegion = true
+		}
+	}
+	if reloadRegion {
+		selector.regionCache.scheduleReloadRegion(selector.region)
 	}
 	// If there is no candidate, fallback to the leader.
 	if selector.targetIdx < 0 {
+		leader := selector.replicas[state.leaderIdx]
+		leaderInvalid := leader.isEpochStale() || state.IsLeaderExhausted(leader)
 		if len(state.option.labels) > 0 {
-			logutil.BgLogger().Warn(
-				"unable to find stores with given labels",
-				zap.Any("labels", state.option.labels),
-			)
+			logutil.Logger(bo.GetCtx()).Warn("unable to find stores with given labels",
+				zap.Uint64("region", selector.region.GetID()),
+				zap.Bool("leader-invalid", leaderInvalid),
+				zap.Any("labels", state.option.labels))
 		}
-		leader := selector.replicas[state.leaderIdx]
-		if leader.isEpochStale() || (!state.option.leaderOnly && leader.isExhausted(1)) {
+		// If leader tried and received deadline exceeded error, return nil to upper layer to retry with default timeout.
+		if leader.deadlineErrUsingConfTimeout {
+			return nil, nil
+		}
+		if leaderInvalid {
 			metrics.TiKVReplicaSelectorFailureCounter.WithLabelValues("exhausted").Inc()
 			selector.invalidateRegion()
 			return nil, nil
@@ -623,6 +672,19 @@ func (state *accessFollower) next(bo *retry.Backoffer, selector *replicaSelector
 	return rpcCtx, nil
 }
 
+func (state *accessFollower) IsLeaderExhausted(leader *replica) bool {
+	// Allow another extra retry for the following case:
+	// 1. The stale read is enabled and leader peer is selected as the target peer at first.
+	// 2. Data is not ready is returned from the leader peer.
+	// 3. Stale read flag is removed and processing falls back to snapshot read on the leader peer.
+	// 4. The leader peer should be retried again using snapshot read.
+	if state.isStaleRead && state.option.leaderOnly {
+		return leader.isExhausted(2)
+	} else {
+		return leader.isExhausted(1)
+	}
+}
+
 func (state *accessFollower) onSendFailure(bo *retry.Backoffer, selector *replicaSelector, cause error) {
 	if selector.checkLiveness(bo, selector.targetReplica()) != reachable {
 		selector.invalidateReplicaStore(selector.targetReplica(), cause)
@@ -630,8 +692,8 @@ func (state *accessFollower) onSendFailure(bo *retry.Backoffer, selector *replic
 }
 
 func (state *accessFollower) isCandidate(idx AccessIndex, replica *replica) bool {
-	// the epoch is staled or retry exhausted.
-	if replica.isEpochStale() || replica.isExhausted(1) {
+	// the epoch is staled or retry exhausted, or the store is unreachable.
+	if replica.isEpochStale() || replica.isExhausted(1) || replica.store.getLivenessState() == unreachable || replica.deadlineErrUsingConfTimeout {
 		return false
 	}
 	// The request can only be sent to the leader.
@@ -750,6 +812,10 @@ func newReplicaSelector(
 		)
 	}
 
+	option := storeSelectorOp{}
+	for _, op := range opts {
+		op(&option)
+	}
 	var state selectorState
 	if !req.ReplicaReadType.IsFollowerRead() {
 		if regionCache.enableForwarding && regionStore.proxyTiKVIdx >= 0 {
@@ -758,21 +824,17 @@ func newReplicaSelector(
 			state = &accessKnownLeader{leaderIdx: regionStore.workTiKVIdx}
 		}
 	} else {
-		option := storeSelectorOp{}
-		for _, op := range opts {
-			op(&option)
-		}
 		if req.ReplicaReadType == kv.ReplicaReadPreferLeader {
 			WithPerferLeader()(&option)
 		}
 		tryLeader := req.ReplicaReadType == kv.ReplicaReadMixed || req.ReplicaReadType == kv.ReplicaReadPreferLeader
 		state = &accessFollower{
-			tryLeader:         tryLeader,
-			isGlobalStaleRead: req.IsGlobalStaleRead(),
-			option:            option,
-			leaderIdx:         regionStore.workTiKVIdx,
-			lastIdx:           -1,
-			learnerOnly:       req.ReplicaReadType == kv.ReplicaReadLearner,
+			tryLeader:   tryLeader,
+			isStaleRead: req.StaleRead,
+			option:      option,
+			leaderIdx:   regionStore.workTiKVIdx,
+			lastIdx:     -1,
+			learnerOnly: req.ReplicaReadType == kv.ReplicaReadLearner,
 		}
 	}
 
@@ -781,6 +843,7 @@ func newReplicaSelector(
 		cachedRegion,
 		regionStore,
 		replicas,
+		option.labels,
 		state,
 		-1,
 		-1,
@@ -912,6 +975,16 @@ func (s *replicaSelector) onSendFailure(bo *retry.Backoffer, err error) {
 	s.state.onSendFailure(bo, s, err)
 }
 
+func (s *replicaSelector) onDeadlineExceeded() {
+	if target := s.targetReplica(); target != nil {
+		target.deadlineErrUsingConfTimeout = true
+	}
+	if accessLeader, ok := s.state.(*accessKnownLeader); ok {
+		// If leader return deadline exceeded error, we should try to access follower next time.
+		s.state = &tryFollower{leaderIdx: accessLeader.leaderIdx, lastIdx: accessLeader.leaderIdx}
+	}
+}
+
 func (s *replicaSelector) checkLiveness(bo *retry.Backoffer, accessReplica *replica) livenessState {
 	store := accessReplica.store
 	liveness := store.requestLiveness(bo, s.regionCache)
@@ -1159,11 +1232,17 @@ func (s *RegionRequestSender) SendReqCtx(
 
 	var staleReadCollector *staleReadMetricsCollector
 	if req.StaleRead {
-		staleReadCollector = &staleReadMetricsCollector{hit: true}
-		staleReadCollector.onReq(req)
-		defer staleReadCollector.collect()
+		staleReadCollector = &staleReadMetricsCollector{}
+		defer func() {
+			if retryTimes == 0 {
+				metrics.StaleReadHitCounter.Add(1)
+			} else {
+				metrics.StaleReadMissCounter.Add(1)
+			}
+		}()
 	}
 
+	totalErrors := make(map[string]int)
 	for {
 		if retryTimes > 0 {
 			req.IsRetryRequest = true
@@ -1174,9 +1253,6 @@ func (s *RegionRequestSender) SendReqCtx(
 					zap.Int("times", retryTimes),
 				)
 			}
-			if req.StaleRead && staleReadCollector != nil {
-				staleReadCollector.hit = false
-			}
 		}
 
 		rpcCtx, err = s.getRPCContext(bo, req, regionID, et, opts...)
@@ -1199,14 +1275,19 @@ func (s *RegionRequestSender) SendReqCtx(
 
 			// TODO: Change the returned error to something like "region missing in cache",
 			// and handle this error like EpochNotMatch, which means to re-split the request and retry.
-			logutil.Logger(bo.GetCtx()).Debug(
-				"throwing pseudo region error due to region not found in cache",
-				zap.Stringer("region", &regionID),
-			)
+			s.logSendReqError(bo, "throwing pseudo region error due to no replica available", regionID, retryTimes, req, totalErrors)
 			resp, err = tikvrpc.GenRegionErrorResp(req, &errorpb.Error{EpochNotMatch: &errorpb.EpochNotMatch{}})
 			return resp, nil, retryTimes, err
 		}
 
+		var isLocalTraffic bool
+		if staleReadCollector != nil && s.replicaSelector != nil {
+			if target := s.replicaSelector.targetReplica(); target != nil {
+				isLocalTraffic = target.store.IsLabelsMatch(s.replicaSelector.labels)
+				staleReadCollector.onReq(req, isLocalTraffic)
+			}
+		}
+
 		logutil.Eventf(bo.GetCtx(), "send %s request to region %d at %s", req.Type, regionID.id, rpcCtx.Addr)
 		s.storeAddr = rpcCtx.Addr
 
@@ -1220,6 +1301,8 @@ func (s *RegionRequestSender) SendReqCtx(
 		var retry bool
 		resp, retry, err = s.sendReqToRegion(bo, rpcCtx, req, timeout)
 		if err != nil {
+			msg := fmt.Sprintf("send request failed, err: %v", err.Error())
+			s.logSendReqError(bo, msg, regionID, retryTimes, req, totalErrors)
 			return nil, nil, retryTimes, err
 		}
 
@@ -1251,26 +1334,100 @@ func (s *RegionRequestSender) SendReqCtx(
 			return nil, nil, retryTimes, err
 		}
 		if regionErr != nil {
+			regionErrLabel := regionErrorToLabel(regionErr)
+			totalErrors[regionErrLabel]++
 			retry, err = s.onRegionError(bo, rpcCtx, req, regionErr)
 			if err != nil {
+				msg := fmt.Sprintf("send request on region error failed, err: %v", err.Error())
+				s.logSendReqError(bo, msg, regionID, retryTimes, req, totalErrors)
 				return nil, nil, retryTimes, err
 			}
 			if retry {
 				retryTimes++
 				continue
 			}
+			s.logSendReqError(bo, "send request meet region error without retry", regionID, retryTimes, req, totalErrors)
 		} else {
 			if s.replicaSelector != nil {
 				s.replicaSelector.onSendSuccess()
 			}
 		}
 		if staleReadCollector != nil {
-			staleReadCollector.onResp(resp)
+			staleReadCollector.onResp(req.Type, resp, isLocalTraffic)
 		}
 		return resp, rpcCtx, retryTimes, nil
 	}
 }
 
+func (s *RegionRequestSender) logSendReqError(bo *retry.Backoffer, msg string, regionID RegionVerID, retryTimes int, req *tikvrpc.Request, totalErrors map[string]int) {
+	var replicaStatus []string
+	replicaSelectorState := "nil"
+	cacheRegionIsValid := "unknown"
+	if s.replicaSelector != nil {
+		switch s.replicaSelector.state.(type) {
+		case *accessKnownLeader:
+			replicaSelectorState = "accessKnownLeader"
+		case *accessFollower:
+			replicaSelectorState = "accessFollower"
+		case *accessByKnownProxy:
+			replicaSelectorState = "accessByKnownProxy"
+		case *tryFollower:
+			replicaSelectorState = "tryFollower"
+		case *tryNewProxy:
+			replicaSelectorState = "tryNewProxy"
+		case *invalidLeader:
+			replicaSelectorState = "invalidLeader"
+		case *invalidStore:
+			replicaSelectorState = "invalidStore"
+		case *stateBase:
+			replicaSelectorState = "stateBase"
+		case nil:
+			replicaSelectorState = "nil"
+		}
+		if s.replicaSelector.region != nil {
+			if s.replicaSelector.region.isValid() {
+				cacheRegionIsValid = "true"
+			} else {
+				cacheRegionIsValid = "false"
+			}
+		}
+		for _, replica := range s.replicaSelector.replicas {
+			replicaStatus = append(replicaStatus, fmt.Sprintf("peer: %v, store: %v, isEpochStale: %v, attempts: %v, replica-epoch: %v, store-epoch: %v, store-state: %v, store-liveness-state: %v",
+				replica.peer.GetId(),
+				replica.store.storeID,
+				replica.isEpochStale(),
+				replica.attempts,
+				replica.epoch,
+				atomic.LoadUint32(&replica.store.epoch),
+				replica.store.getResolveState(),
+				replica.store.getLivenessState(),
+			))
+		}
+	}
+	var totalErrorStr bytes.Buffer
+	for err, cnt := range totalErrors {
+		if totalErrorStr.Len() > 0 {
+			totalErrorStr.WriteString(", ")
+		}
+		totalErrorStr.WriteString(err)
+		totalErrorStr.WriteString(":")
+		totalErrorStr.WriteString(strconv.Itoa(cnt))
+	}
+	logutil.Logger(bo.GetCtx()).Info(msg,
+		zap.Uint64("req-ts", req.GetStartTS()),
+		zap.String("req-type", req.Type.String()),
+		zap.String("region", regionID.String()),
+		zap.String("region-is-valid", cacheRegionIsValid),
+		zap.Int("retry-times", retryTimes),
+		zap.String("replica-read-type", req.ReplicaReadType.String()),
+		zap.String("replica-selector-state", replicaSelectorState),
+		zap.Bool("stale-read", req.StaleRead),
+		zap.String("replica-status", strings.Join(replicaStatus, "; ")),
+		zap.Int("total-backoff-ms", bo.GetTotalSleep()),
+		zap.Int("total-backoff-times", bo.GetTotalBackoffTimes()),
+		zap.String("total-region-errors", totalErrorStr.String()))
+}
+
 // RPCCancellerCtxKey is context key attach rpc send cancelFunc collector to ctx.
 type RPCCancellerCtxKey struct{}
 
@@ -1512,7 +1669,7 @@ func (s *RegionRequestSender) sendReqToRegion(
 				return nil, false, err
 			}
 		}
-		if e := s.onSendFail(bo, rpcCtx, err); e != nil {
+		if e := s.onSendFail(bo, rpcCtx, req, err); e != nil {
 			return nil, false, err
 		}
 		return nil, true, nil
@@ -1542,7 +1699,7 @@ func (s *RegionRequestSender) releaseStoreToken(st *Store) {
 	logutil.BgLogger().Warn("release store token failed, count equals to 0")
 }
 
-func (s *RegionRequestSender) onSendFail(bo *retry.Backoffer, ctx *RPCContext, err error) error {
+func (s *RegionRequestSender) onSendFail(bo *retry.Backoffer, ctx *RPCContext, req *tikvrpc.Request, err error) error {
 	if span := opentracing.SpanFromContext(bo.GetCtx()); span != nil && span.Tracer() != nil {
 		span1 := span.Tracer().StartSpan("regionRequest.onSendFail", opentracing.ChildOf(span.Context()))
 		defer span1.Finish()
@@ -1553,6 +1710,11 @@ func (s *RegionRequestSender) onSendFail(bo *retry.Backoffer, ctx *RPCContext, e
 		return errors.WithStack(err)
 	} else if LoadShuttingDown() > 0 {
 		return errors.WithStack(tikverr.ErrTiDBShuttingDown)
+	} else if errors.Cause(err) == context.DeadlineExceeded && req.MaxExecutionDurationMs < uint64(client.ReadTimeoutShort.Milliseconds()) {
+		if s.replicaSelector != nil {
+			s.replicaSelector.onDeadlineExceeded()
+			return nil
+		}
 	}
 	if status.Code(errors.Cause(err)) == codes.Canceled {
 		select {
@@ -1562,7 +1724,7 @@ func (s *RegionRequestSender) onSendFail(bo *retry.Backoffer, ctx *RPCContext, e
 			// If we don't cancel, but the error code is Canceled, it must be from grpc remote.
 			// This may happen when tikv is killed and exiting.
 			// Backoff and retry in this case.
-			logutil.BgLogger().Warn("receive a grpc cancel signal from remote", zap.Error(err))
+			logutil.Logger(bo.GetCtx()).Warn("receive a grpc cancel signal from remote", zap.Error(err))
 		}
 	}
 
@@ -1644,6 +1806,9 @@ func regionErrorToLabel(e *errorpb.Error) string {
 	} else if e.GetEpochNotMatch() != nil {
 		return "epoch_not_match"
 	} else if e.GetServerIsBusy() != nil {
+		if strings.Contains(e.GetServerIsBusy().GetReason(), "deadline is exceeded") {
+			return "deadline_exceeded"
+		}
 		return "server_is_busy"
 	} else if e.GetStaleCommand() != nil {
 		return "stale_command"
@@ -1671,10 +1836,16 @@ func regionErrorToLabel(e *errorpb.Error) string {
 		return "flashback_not_prepared"
 	} else if e.GetIsWitness() != nil {
 		return "peer_is_witness"
+	} else if isDeadlineExceeded(e) {
+		return "deadline_exceeded"
 	}
 	return "unknown"
 }
 
+func isDeadlineExceeded(e *errorpb.Error) bool {
+	return strings.Contains(e.GetMessage(), "Deadline is exceeded")
+}
+
 func (s *RegionRequestSender) onRegionError(
 	bo *retry.Backoffer, ctx *RPCContext, req *tikvrpc.Request, regionErr *errorpb.Error,
 ) (shouldRetry bool, err error) {
@@ -1693,7 +1864,7 @@ func (s *RegionRequestSender) onRegionError(
 
 	if notLeader := regionErr.GetNotLeader(); notLeader != nil {
 		// Retry if error is `NotLeader`.
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `NotLeader` retry later",
 			zap.String("notLeader", notLeader.String()),
 			zap.String("ctx", ctx.String()),
@@ -1734,7 +1905,7 @@ func (s *RegionRequestSender) onRegionError(
 
 	if regionErr.GetRecoveryInProgress() != nil {
 		s.regionCache.InvalidateCachedRegion(ctx.Region)
-		logutil.BgLogger().Debug("tikv reports `RecoveryInProgress`", zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Debug("tikv reports `RecoveryInProgress`", zap.Stringer("ctx", ctx))
 		err = bo.Backoff(retry.BoRegionRecoveryInProgress, errors.Errorf("region recovery in progress, ctx: %v", ctx))
 		if err != nil {
 			return false, err
@@ -1744,7 +1915,7 @@ func (s *RegionRequestSender) onRegionError(
 
 	if regionErr.GetIsWitness() != nil {
 		s.regionCache.InvalidateCachedRegion(ctx.Region)
-		logutil.BgLogger().Debug("tikv reports `IsWitness`", zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Debug("tikv reports `IsWitness`", zap.Stringer("ctx", ctx))
 		err = bo.Backoff(retry.BoIsWitness, errors.Errorf("is witness, ctx: %v", ctx))
 		if err != nil {
 			return false, err
@@ -1756,7 +1927,7 @@ func (s *RegionRequestSender) onRegionError(
 	// if a request meets the FlashbackInProgress error, it should stop retrying immediately
 	// to avoid unnecessary backoff and potential unexpected data status to the user.
 	if flashbackInProgress := regionErr.GetFlashbackInProgress(); flashbackInProgress != nil {
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `FlashbackInProgress`",
 			zap.Stringer("req", req),
 			zap.Stringer("ctx", ctx),
@@ -1785,7 +1956,7 @@ func (s *RegionRequestSender) onRegionError(
 	// prepared for the flashback before, it should stop retrying immediately to avoid
 	// unnecessary backoff.
 	if regionErr.GetFlashbackNotPrepared() != nil {
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `FlashbackNotPrepared`",
 			zap.Stringer("req", req),
 			zap.Stringer("ctx", ctx),
@@ -1803,13 +1974,13 @@ func (s *RegionRequestSender) onRegionError(
 	}
 
 	if regionErr.GetKeyNotInRegion() != nil {
-		logutil.BgLogger().Error("tikv reports `KeyNotInRegion`", zap.Stringer("req", req), zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Error("tikv reports `KeyNotInRegion`", zap.Stringer("req", req), zap.Stringer("ctx", ctx))
 		s.regionCache.InvalidateCachedRegion(ctx.Region)
 		return false, nil
 	}
 
 	if epochNotMatch := regionErr.GetEpochNotMatch(); epochNotMatch != nil {
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `EpochNotMatch` retry later",
 			zap.Stringer("EpochNotMatch", epochNotMatch),
 			zap.Stringer("ctx", ctx),
@@ -1822,10 +1993,14 @@ func (s *RegionRequestSender) onRegionError(
 	}
 
 	if serverIsBusy := regionErr.GetServerIsBusy(); serverIsBusy != nil {
+		if s.replicaSelector != nil && strings.Contains(serverIsBusy.GetReason(), "deadline is exceeded") {
+			s.replicaSelector.onDeadlineExceeded()
+			return true, nil
+		}
 		if s.replicaSelector != nil {
 			return s.replicaSelector.onServerIsBusy(bo, ctx, req, serverIsBusy)
 		}
-		logutil.BgLogger().Warn(
+		logutil.Logger(bo.GetCtx()).Warn(
 			"tikv reports `ServerIsBusy` retry later",
 			zap.String("reason", regionErr.GetServerIsBusy().GetReason()),
 			zap.Stringer("ctx", ctx),
@@ -1845,7 +2020,7 @@ func (s *RegionRequestSender) onRegionError(
 	// We can't know whether the request is committed or not, so it's an undetermined error too,
 	// but we don't handle it now.
 	if regionErr.GetStaleCommand() != nil {
-		logutil.BgLogger().Debug("tikv reports `StaleCommand`", zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Debug("tikv reports `StaleCommand`", zap.Stringer("ctx", ctx))
 		if s.replicaSelector != nil {
 			// Needn't backoff because the new leader should be elected soon
 			// and the replicaSelector will try the next peer.
@@ -1860,7 +2035,7 @@ func (s *RegionRequestSender) onRegionError(
 
 	if storeNotMatch := regionErr.GetStoreNotMatch(); storeNotMatch != nil {
 		// store not match
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `StoreNotMatch` retry later",
 			zap.Stringer("storeNotMatch", storeNotMatch),
 			zap.Stringer("ctx", ctx),
@@ -1874,12 +2049,12 @@ func (s *RegionRequestSender) onRegionError(
 	}
 
 	if regionErr.GetRaftEntryTooLarge() != nil {
-		logutil.BgLogger().Warn("tikv reports `RaftEntryTooLarge`", zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Warn("tikv reports `RaftEntryTooLarge`", zap.Stringer("ctx", ctx))
 		return false, errors.New(regionErr.String())
 	}
 
 	if regionErr.GetMaxTimestampNotSynced() != nil {
-		logutil.BgLogger().Debug("tikv reports `MaxTimestampNotSynced`", zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Debug("tikv reports `MaxTimestampNotSynced`", zap.Stringer("ctx", ctx))
 		err = bo.Backoff(retry.BoMaxTsNotSynced, errors.Errorf("max timestamp not synced, ctx: %v", ctx))
 		if err != nil {
 			return false, err
@@ -1889,7 +2064,7 @@ func (s *RegionRequestSender) onRegionError(
 
 	// A read request may be sent to a peer which has not been initialized yet, we should retry in this case.
 	if regionErr.GetRegionNotInitialized() != nil {
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `RegionNotInitialized` retry later",
 			zap.Uint64("store-id", ctx.Store.storeID),
 			zap.Uint64("region-id", regionErr.GetRegionNotInitialized().GetRegionId()),
@@ -1904,7 +2079,7 @@ func (s *RegionRequestSender) onRegionError(
 
 	// The read-index can't be handled timely because the region is splitting or merging.
 	if regionErr.GetReadIndexNotReady() != nil {
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `ReadIndexNotReady` retry later",
 			zap.Uint64("store-id", ctx.Store.storeID),
 			zap.Uint64("region-id", regionErr.GetRegionNotInitialized().GetRegionId()),
@@ -1919,7 +2094,7 @@ func (s *RegionRequestSender) onRegionError(
 	}
 
 	if regionErr.GetProposalInMergingMode() != nil {
-		logutil.BgLogger().Debug("tikv reports `ProposalInMergingMode`", zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Debug("tikv reports `ProposalInMergingMode`", zap.Stringer("ctx", ctx))
 		// The region is merging and it can't provide service until merge finished, so backoff.
 		err = bo.Backoff(retry.BoRegionScheduling, errors.Errorf("region is merging, ctx: %v", ctx))
 		if err != nil {
@@ -1932,7 +2107,7 @@ func (s *RegionRequestSender) onRegionError(
 	// This error is specific to stale read and the target replica is randomly selected. If the request is sent
 	// to the leader, the data must be ready, so we don't backoff here.
 	if regionErr.GetDataIsNotReady() != nil {
-		logutil.BgLogger().Warn(
+		logutil.Logger(bo.GetCtx()).Warn(
 			"tikv reports `DataIsNotReady` retry later",
 			zap.Uint64("store-id", ctx.Store.storeID),
 			zap.Uint64("peer-id", regionErr.GetDataIsNotReady().GetPeerId()),
@@ -1950,7 +2125,11 @@ func (s *RegionRequestSender) onRegionError(
 		return true, nil
 	}
 
-	logutil.BgLogger().Debug(
+	if isDeadlineExceeded(regionErr) && s.replicaSelector != nil {
+		s.replicaSelector.onDeadlineExceeded()
+	}
+
+	logutil.Logger(bo.GetCtx()).Debug(
 		"tikv reports region failed",
 		zap.Stringer("regionErr", regionErr),
 		zap.Stringer("ctx", ctx),
@@ -1972,35 +2151,36 @@ func (s *RegionRequestSender) onRegionError(
 }
 
 type staleReadMetricsCollector struct {
-	tp  tikvrpc.CmdType
-	hit bool
-	out int
-	in  int
 }
 
-func (s *staleReadMetricsCollector) onReq(req *tikvrpc.Request) {
+func (s *staleReadMetricsCollector) onReq(req *tikvrpc.Request, isLocalTraffic bool) {
 	size := 0
 	switch req.Type {
 	case tikvrpc.CmdGet:
-		size += req.Get().Size()
+		size = req.Get().Size()
 	case tikvrpc.CmdBatchGet:
-		size += req.BatchGet().Size()
+		size = req.BatchGet().Size()
 	case tikvrpc.CmdScan:
-		size += req.Scan().Size()
+		size = req.Scan().Size()
 	case tikvrpc.CmdCop:
-		size += req.Cop().Size()
+		size = req.Cop().Size()
 	default:
 		// ignore non-read requests
 		return
 	}
-	s.tp = req.Type
 	size += req.Context.Size()
-	s.out = size
+	if isLocalTraffic {
+		metrics.StaleReadLocalOutBytes.Add(float64(size))
+		metrics.StaleReadReqLocalCounter.Add(1)
+	} else {
+		metrics.StaleReadRemoteOutBytes.Add(float64(size))
+		metrics.StaleReadReqCrossZoneCounter.Add(1)
+	}
 }
 
-func (s *staleReadMetricsCollector) onResp(resp *tikvrpc.Response) {
+func (s *staleReadMetricsCollector) onResp(tp tikvrpc.CmdType, resp *tikvrpc.Response, isLocalTraffic bool) {
 	size := 0
-	switch s.tp {
+	switch tp {
 	case tikvrpc.CmdGet:
 		size += resp.Resp.(*kvrpcpb.GetResponse).Size()
 	case tikvrpc.CmdBatchGet:
@@ -2010,19 +2190,12 @@ func (s *staleReadMetricsCollector) onResp(resp *tikvrpc.Response) {
 	case tikvrpc.CmdCop:
 		size += resp.Resp.(*coprocessor.Response).Size()
 	default:
-		// unreachable
+		// ignore non-read requests
 		return
 	}
-	s.in = size
-}
-
-func (s *staleReadMetricsCollector) collect() {
-	in, out := metrics.StaleReadHitInTraffic, metrics.StaleReadHitOutTraffic
-	if !s.hit {
-		in, out = metrics.StaleReadMissInTraffic, metrics.StaleReadMissOutTraffic
-	}
-	if s.in > 0 && s.out > 0 {
-		in.Observe(float64(s.in))
-		out.Observe(float64(s.out))
+	if isLocalTraffic {
+		metrics.StaleReadLocalInBytes.Add(float64(size))
+	} else {
+		metrics.StaleReadRemoteInBytes.Add(float64(size))
 	}
 }
diff --git a/internal/mockstore/mocktikv/pd.go b/internal/mockstore/mocktikv/pd.go
index a5dd62d..4450cec 100644
--- a/internal/mockstore/mocktikv/pd.go
+++ b/internal/mockstore/mocktikv/pd.go
@@ -290,6 +290,10 @@ func (c *pdClient) UpdateServiceGCSafePoint(ctx context.Context, serviceID strin
 	return minSafePoint, nil
 }
 
+func (c *pdClient) GetAllKeyspaces(ctx context.Context, startID uint32, limit uint32) ([]*keyspacepb.KeyspaceMeta, error) {
+	return nil, nil
+}
+
 func (c *pdClient) Close() {
 }
 
@@ -335,10 +339,6 @@ func (c *pdClient) UpdateKeyspaceState(ctx context.Context, id uint32, state key
 	return nil, nil
 }
 
-func (c *pdClient) GetAllKeyspaces(ctx context.Context, startID uint32, limit uint32) ([]*keyspacepb.KeyspaceMeta, error) {
-	return nil, nil
-}
-
 func (c *pdClient) ListResourceGroups(ctx context.Context) ([]*rmpb.ResourceGroup, error) {
 	return nil, nil
 }
diff --git a/internal/retry/backoff.go b/internal/retry/backoff.go
index 6a27d05..19f6fca 100644
--- a/internal/retry/backoff.go
+++ b/internal/retry/backoff.go
@@ -35,9 +35,11 @@
 package retry
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"math"
+	"strconv"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -131,7 +133,7 @@ func (b *Backoffer) BackoffWithMaxSleepTxnLockFast(maxSleepMs int, err error) er
 // and never sleep more than maxSleepMs for each sleep.
 func (b *Backoffer) BackoffWithCfgAndMaxSleep(cfg *Config, maxSleepMs int, err error) error {
 	if strings.Contains(err.Error(), tikverr.MismatchClusterID) {
-		logutil.BgLogger().Fatal("critical error", zap.Error(err))
+		logutil.Logger(b.ctx).Fatal("critical error", zap.Error(err))
 	}
 	select {
 	case <-b.ctx.Done():
@@ -150,12 +152,24 @@ func (b *Backoffer) BackoffWithCfgAndMaxSleep(cfg *Config, maxSleepMs int, err e
 				errMsg += "\n" + err.Error()
 			}
 		}
+		var backoffDetail bytes.Buffer
+		totalTimes := 0
+		for name, times := range b.backoffTimes {
+			totalTimes += times
+			if backoffDetail.Len() > 0 {
+				backoffDetail.WriteString(", ")
+			}
+			backoffDetail.WriteString(name)
+			backoffDetail.WriteString(":")
+			backoffDetail.WriteString(strconv.Itoa(times))
+		}
+		errMsg += fmt.Sprintf("\ntotal-backoff-times: %v, backoff-detail: %v", totalTimes, backoffDetail.String())
 		returnedErr := err
 		if longestSleepCfg != nil {
 			errMsg += fmt.Sprintf("\nlongest sleep type: %s, time: %dms", longestSleepCfg.String(), longestSleepTime)
 			returnedErr = longestSleepCfg.err
 		}
-		logutil.BgLogger().Warn(errMsg)
+		logutil.Logger(b.ctx).Warn(errMsg)
 		// Use the backoff type that contributes most to the timeout to generate a MySQL error.
 		return errors.WithStack(returnedErr)
 	}
diff --git a/internal/unionstore/memdb_iterator.go b/internal/unionstore/memdb_iterator.go
index 3e37d86..3b4bdfd 100644
--- a/internal/unionstore/memdb_iterator.go
+++ b/internal/unionstore/memdb_iterator.go
@@ -67,10 +67,11 @@ func (db *MemDB) Iter(k []byte, upperBound []byte) (Iterator, error) {
 // IterReverse creates a reversed Iterator positioned on the first entry which key is less than k.
 // The returned iterator will iterate from greater key to smaller key.
 // If k is nil, the returned iterator will be positioned at the last key.
-// TODO: Add lower bound limit
-func (db *MemDB) IterReverse(k []byte) (Iterator, error) {
+// It yields only keys that >= lowerBound. If lowerBound is nil, it means the lowerBound is unbounded.
+func (db *MemDB) IterReverse(k []byte, lowerBound []byte) (Iterator, error) {
 	i := &MemdbIterator{
 		db:      db,
+		start:   lowerBound,
 		end:     k,
 		reverse: true,
 	}
@@ -128,7 +129,7 @@ func (i *MemdbIterator) Valid() bool {
 	if !i.reverse {
 		return !i.curr.isNull() && (i.end == nil || bytes.Compare(i.Key(), i.end) < 0)
 	}
-	return !i.curr.isNull()
+	return !i.curr.isNull() && (i.start == nil || bytes.Compare(i.Key(), i.start) >= 0)
 }
 
 // Flags returns flags belong to current iterator.
diff --git a/internal/unionstore/memdb_snapshot.go b/internal/unionstore/memdb_snapshot.go
index 2adedfb..ad6b87d 100644
--- a/internal/unionstore/memdb_snapshot.go
+++ b/internal/unionstore/memdb_snapshot.go
@@ -60,6 +60,21 @@ func (db *MemDB) SnapshotIter(start, end []byte) Iterator {
 	return it
 }
 
+// SnapshotIterReverse returns a reverse Iterator for a snapshot of MemBuffer.
+func (db *MemDB) SnapshotIterReverse(k, lowerBound []byte) Iterator {
+	it := &memdbSnapIter{
+		MemdbIterator: &MemdbIterator{
+			db:      db,
+			start:   lowerBound,
+			end:     k,
+			reverse: true,
+		},
+		cp: db.getSnapshot(),
+	}
+	it.init()
+	return it
+}
+
 func (db *MemDB) getSnapshot() MemDBCheckpoint {
 	if len(db.stages) > 0 {
 		return db.stages[0]
@@ -123,10 +138,18 @@ func (i *memdbSnapIter) setValue() bool {
 }
 
 func (i *memdbSnapIter) init() {
-	if len(i.start) == 0 {
-		i.seekToFirst()
+	if i.reverse {
+		if len(i.end) == 0 {
+			i.seekToLast()
+		} else {
+			i.seek(i.end)
+		}
 	} else {
-		i.seek(i.start)
+		if len(i.start) == 0 {
+			i.seekToFirst()
+		} else {
+			i.seek(i.start)
+		}
 	}
 
 	if !i.setValue() {
diff --git a/internal/unionstore/mock.go b/internal/unionstore/mock.go
index 66c5f6c..6586975 100644
--- a/internal/unionstore/mock.go
+++ b/internal/unionstore/mock.go
@@ -71,8 +71,8 @@ func (s *mockSnapshot) Iter(k []byte, upperBound []byte) (Iterator, error) {
 	return s.store.Iter(k, upperBound)
 }
 
-func (s *mockSnapshot) IterReverse(k []byte) (Iterator, error) {
-	return s.store.IterReverse(k)
+func (s *mockSnapshot) IterReverse(k, lowerBound []byte) (Iterator, error) {
+	return s.store.IterReverse(k, lowerBound)
 }
 
 func (s *mockSnapshot) SetOption(opt int, val interface{}) {}
diff --git a/internal/unionstore/union_store.go b/internal/unionstore/union_store.go
index e87190f..fbf6290 100644
--- a/internal/unionstore/union_store.go
+++ b/internal/unionstore/union_store.go
@@ -71,8 +71,8 @@ type uSnapshot interface {
 	// IterReverse creates a reversed Iterator positioned on the first entry which key is less than k.
 	// The returned iterator will iterate from greater key to smaller key.
 	// If k is nil, the returned iterator will be positioned at the last key.
-	// TODO: Add lower bound limit
-	IterReverse(k []byte) (Iterator, error)
+	// It yields only keys that >= lowerBound. If lowerBound is nil, it means the lowerBound is unbounded.
+	IterReverse(k, lowerBound []byte) (Iterator, error)
 }
 
 // KVUnionStore is an in-memory Store which contains a buffer for write and a
@@ -124,12 +124,12 @@ func (us *KVUnionStore) Iter(k, upperBound []byte) (Iterator, error) {
 }
 
 // IterReverse implements the Retriever interface.
-func (us *KVUnionStore) IterReverse(k []byte) (Iterator, error) {
-	bufferIt, err := us.memBuffer.IterReverse(k)
+func (us *KVUnionStore) IterReverse(k, lowerBound []byte) (Iterator, error) {
+	bufferIt, err := us.memBuffer.IterReverse(k, lowerBound)
 	if err != nil {
 		return nil, err
 	}
-	retrieverIt, err := us.snapshot.IterReverse(k)
+	retrieverIt, err := us.snapshot.IterReverse(k, lowerBound)
 	if err != nil {
 		return nil, err
 	}
diff --git a/kv/store_vars.go b/kv/store_vars.go
index cce3e14..184975c 100644
--- a/kv/store_vars.go
+++ b/kv/store_vars.go
@@ -35,6 +35,8 @@
 package kv
 
 import (
+	"fmt"
+
 	"go.uber.org/atomic"
 )
 
@@ -72,3 +74,21 @@ const (
 func (r ReplicaReadType) IsFollowerRead() bool {
 	return r != ReplicaReadLeader
 }
+
+// String implements fmt.Stringer interface.
+func (r ReplicaReadType) String() string {
+	switch r {
+	case ReplicaReadLeader:
+		return "leader"
+	case ReplicaReadFollower:
+		return "follower"
+	case ReplicaReadMixed:
+		return "mixed"
+	case ReplicaReadLearner:
+		return "learner"
+	case ReplicaReadPreferLeader:
+		return "prefer-leader"
+	default:
+		return fmt.Sprintf("unknown-%v", byte(r))
+	}
+}
diff --git a/metrics/metrics.go b/metrics/metrics.go
index 4f5abfd..5c72f05 100644
--- a/metrics/metrics.go
+++ b/metrics/metrics.go
@@ -101,7 +101,9 @@ var (
 	TiKVAggressiveLockedKeysCounter          *prometheus.CounterVec
 	TiKVStoreSlowScoreGauge                  *prometheus.GaugeVec
 	TiKVPreferLeaderFlowsGauge               *prometheus.GaugeVec
-	TiKVStaleReadSizeSummary                 *prometheus.SummaryVec
+	TiKVStaleReadCounter                     *prometheus.CounterVec
+	TiKVStaleReadReqCounter                  *prometheus.CounterVec
+	TiKVStaleReadBytes                       *prometheus.CounterVec
 )
 
 // Label constants.
@@ -700,13 +702,28 @@ func initMetrics(namespace, subsystem string, constLabels prometheus.Labels) {
 			ConstLabels: constLabels,
 		}, []string{LblType, LblStore})
 
-	TiKVStaleReadSizeSummary = prometheus.NewSummaryVec(
-		prometheus.SummaryOpts{
-			Namespace:   namespace,
-			Subsystem:   subsystem,
-			Name:        "stale_read_bytes",
-			Help:        "Size of stale read.",
-			ConstLabels: constLabels,
+	TiKVStaleReadCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "stale_read_counter",
+			Help:      "Counter of stale read hit/miss",
+		}, []string{LblResult})
+
+	TiKVStaleReadReqCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "stale_read_req_counter",
+			Help:      "Counter of stale read requests",
+		}, []string{LblType})
+
+	TiKVStaleReadBytes = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "stale_read_bytes",
+			Help:      "Counter of stale read requests bytes",
 		}, []string{LblResult, LblDirection})
 
 	initShortcuts()
@@ -789,7 +806,9 @@ func RegisterMetrics() {
 	prometheus.MustRegister(TiKVAggressiveLockedKeysCounter)
 	prometheus.MustRegister(TiKVStoreSlowScoreGauge)
 	prometheus.MustRegister(TiKVPreferLeaderFlowsGauge)
-	prometheus.MustRegister(TiKVStaleReadSizeSummary)
+	prometheus.MustRegister(TiKVStaleReadCounter)
+	prometheus.MustRegister(TiKVStaleReadReqCounter)
+	prometheus.MustRegister(TiKVStaleReadBytes)
 }
 
 // readCounter reads the value of a prometheus.Counter.
diff --git a/metrics/shortcuts.go b/metrics/shortcuts.go
index 0acd7ab..c791cb9 100644
--- a/metrics/shortcuts.go
+++ b/metrics/shortcuts.go
@@ -160,10 +160,16 @@ var (
 	AggressiveLockedKeysLockedWithConflict prometheus.Counter
 	AggressiveLockedKeysNonForceLock       prometheus.Counter
 
-	StaleReadHitInTraffic   prometheus.Observer
-	StaleReadHitOutTraffic  prometheus.Observer
-	StaleReadMissInTraffic  prometheus.Observer
-	StaleReadMissOutTraffic prometheus.Observer
+	StaleReadHitCounter  prometheus.Counter
+	StaleReadMissCounter prometheus.Counter
+
+	StaleReadReqLocalCounter     prometheus.Counter
+	StaleReadReqCrossZoneCounter prometheus.Counter
+
+	StaleReadLocalInBytes   prometheus.Counter
+	StaleReadLocalOutBytes  prometheus.Counter
+	StaleReadRemoteInBytes  prometheus.Counter
+	StaleReadRemoteOutBytes prometheus.Counter
 )
 
 func initShortcuts() {
@@ -296,8 +302,14 @@ func initShortcuts() {
 	// TiKV).
 	AggressiveLockedKeysNonForceLock = TiKVAggressiveLockedKeysCounter.WithLabelValues("non_force_lock")
 
-	StaleReadHitInTraffic = TiKVStaleReadSizeSummary.WithLabelValues("hit", "in")
-	StaleReadHitOutTraffic = TiKVStaleReadSizeSummary.WithLabelValues("hit", "out")
-	StaleReadMissInTraffic = TiKVStaleReadSizeSummary.WithLabelValues("miss", "in")
-	StaleReadMissOutTraffic = TiKVStaleReadSizeSummary.WithLabelValues("miss", "out")
+	StaleReadHitCounter = TiKVStaleReadCounter.WithLabelValues("hit")
+	StaleReadMissCounter = TiKVStaleReadCounter.WithLabelValues("miss")
+
+	StaleReadReqLocalCounter = TiKVStaleReadReqCounter.WithLabelValues("local")
+	StaleReadReqCrossZoneCounter = TiKVStaleReadReqCounter.WithLabelValues("cross-zone")
+
+	StaleReadLocalInBytes = TiKVStaleReadBytes.WithLabelValues("local", "in")
+	StaleReadLocalOutBytes = TiKVStaleReadBytes.WithLabelValues("local", "out")
+	StaleReadRemoteInBytes = TiKVStaleReadBytes.WithLabelValues("cross-zone", "in")
+	StaleReadRemoteOutBytes = TiKVStaleReadBytes.WithLabelValues("cross-zone", "out")
 }
diff --git a/rawkv/rawkv.go b/rawkv/rawkv.go
index e1026e7..f851a3e 100644
--- a/rawkv/rawkv.go
+++ b/rawkv/rawkv.go
@@ -551,7 +551,8 @@ func (c *Client) Scan(ctx context.Context, startKey, endKey []byte, limit int, o
 	return
 }
 
-// ReverseScan queries continuous kv pairs in range [endKey, startKey), up to limit pairs.
+// ReverseScan queries continuous kv pairs in range [endKey, startKey),
+// from startKey(upperBound) to endKey(lowerBound), up to limit pairs.
 // The returned keys are in reversed lexicographical order.
 // If endKey is empty, it means unbounded.
 // If you want to include the startKey or exclude the endKey, push a '\0' to the key. For example, to scan
diff --git a/tikv/gc.go b/tikv/gc.go
index d448815..2b47e6b 100644
--- a/tikv/gc.go
+++ b/tikv/gc.go
@@ -50,8 +50,15 @@ import (
 //
 // GC is a simplified version of [GC in TiDB](https://docs.pingcap.com/tidb/stable/garbage-collection-overview).
 // We skip the second step "delete ranges" which is an optimization for TiDB.
-func (s *KVStore) GC(ctx context.Context, safepoint uint64) (newSafePoint uint64, err error) {
-	err = s.resolveLocks(ctx, safepoint, 8)
+func (s *KVStore) GC(ctx context.Context, safepoint uint64, opts ...GCOpt) (newSafePoint uint64, err error) {
+	// default concurrency 8
+	opt := &gcOption{concurrency: 8}
+	// Apply gc options.
+	for _, o := range opts {
+		o(opt)
+	}
+
+	err = s.resolveLocks(ctx, safepoint, opt.concurrency)
 	if err != nil {
 		return
 	}
@@ -59,6 +66,20 @@ func (s *KVStore) GC(ctx context.Context, safepoint uint64) (newSafePoint uint64
 	return s.pdClient.UpdateGCSafePoint(ctx, safepoint)
 }
 
+type gcOption struct {
+	concurrency int
+}
+
+// GCOpt gc options
+type GCOpt func(*gcOption)
+
+// WithConcurrency is used to set gc RangeTaskRunner concurrency.
+func WithConcurrency(concurrency int) GCOpt {
+	return func(opt *gcOption) {
+		opt.concurrency = concurrency
+	}
+}
+
 func (s *KVStore) resolveLocks(ctx context.Context, safePoint uint64, concurrency int) error {
 	handler := func(ctx context.Context, r kv.KeyRange) (rangetask.TaskStat, error) {
 		return s.resolveLocksForRange(ctx, safePoint, r.StartKey, r.EndKey)
diff --git a/txnkv/transaction/txn.go b/txnkv/transaction/txn.go
index 88552a2..6d90a16 100644
--- a/txnkv/transaction/txn.go
+++ b/txnkv/transaction/txn.go
@@ -247,8 +247,8 @@ func (txn *KVTxn) Iter(k []byte, upperBound []byte) (unionstore.Iterator, error)
 }
 
 // IterReverse creates a reversed Iterator positioned on the first entry which key is less than k.
-func (txn *KVTxn) IterReverse(k []byte) (unionstore.Iterator, error) {
-	return txn.us.IterReverse(k)
+func (txn *KVTxn) IterReverse(k, lowerBound []byte) (unionstore.Iterator, error) {
+	return txn.us.IterReverse(k, lowerBound)
 }
 
 // Delete removes the entry for key k from kv store.
@@ -556,10 +556,27 @@ func (txn *KVTxn) Rollback() error {
 		txn.CancelAggressiveLocking(context.Background())
 	}
 
+	// `skipPessimisticRollback` may be true only when set by failpoint in tests.
+	skipPessimisticRollback := false
+	if val, err := util.EvalFailpoint("onRollback"); err == nil {
+		if s, ok := val.(string); ok {
+			if s == "skipRollbackPessimisticLock" {
+				logutil.BgLogger().Info("[failpoint] injected skip pessimistic rollback on explicit rollback",
+					zap.Uint64("txnStartTS", txn.startTS))
+				skipPessimisticRollback = true
+			} else {
+				panic(fmt.Sprintf("unknown instruction %s for failpoint \"onRollback\"", s))
+			}
+		}
+	}
+
 	start := time.Now()
 	// Clean up pessimistic lock.
 	if txn.IsPessimistic() && txn.committer != nil {
-		err := txn.rollbackPessimisticLocks()
+		var err error
+		if !skipPessimisticRollback {
+			err = txn.rollbackPessimisticLocks()
+		}
 		txn.committer.ttlManager.close()
 		if err != nil {
 			logutil.BgLogger().Error(err.Error())
diff --git a/txnkv/txnlock/lock_resolver.go b/txnkv/txnlock/lock_resolver.go
index 184674b..b93b86e 100644
--- a/txnkv/txnlock/lock_resolver.go
+++ b/txnkv/txnlock/lock_resolver.go
@@ -245,6 +245,12 @@ func (lr *LockResolver) BatchResolveLocks(bo *retry.Backoffer, locks []*Lock, lo
 		}
 		metrics.LockResolverCountWithExpired.Inc()
 
+		// Use currentTS = math.MaxUint64 means rollback the txn, no matter the lock is expired or not!
+		status, err := lr.getTxnStatus(bo, l.TxnID, l.Primary, 0, math.MaxUint64, true, false, l)
+		if err != nil {
+			return false, err
+		}
+
 		if l.LockType == kvrpcpb.Op_PessimisticLock {
 			// BatchResolveLocks forces resolving the locks ignoring whether whey are expired.
 			// For pessimistic locks, committing them makes no sense, but it won't affect transaction
@@ -252,6 +258,9 @@ func (lr *LockResolver) BatchResolveLocks(bo *retry.Backoffer, locks []*Lock, lo
 			// Pessimistic locks needs special handling logic because their primary may not point
 			// to the real primary of that transaction, and their state cannot be put in `txnInfos`.
 			// (see: https://github.com/pingcap/tidb/issues/42937).
+			//
+			// `resolvePessimisticLock` should be called after calling `getTxnStatus`.
+			// See: https://github.com/pingcap/tidb/issues/45134
 			err := lr.resolvePessimisticLock(bo, l)
 			if err != nil {
 				return false, err
@@ -259,12 +268,6 @@ func (lr *LockResolver) BatchResolveLocks(bo *retry.Backoffer, locks []*Lock, lo
 			continue
 		}
 
-		// Use currentTS = math.MaxUint64 means rollback the txn, no matter the lock is expired or not!
-		status, err := lr.getTxnStatus(bo, l.TxnID, l.Primary, 0, math.MaxUint64, true, false, l)
-		if err != nil {
-			return false, err
-		}
-
 		// If the transaction uses async commit, CheckTxnStatus will reject rolling back the primary lock.
 		// Then we need to check the secondary locks to determine the final status of the transaction.
 		if status.primaryLock != nil && status.primaryLock.UseAsyncCommit {
@@ -1173,6 +1176,8 @@ func (lr *LockResolver) resolveLock(bo *retry.Backoffer, l *Lock, status TxnStat
 	}
 }
 
+// resolvePessimisticLock handles pessimistic locks after checking txn status.
+// Note that this function assumes `CheckTxnStatus` is done (or `getTxnStatusFromLock` has been called) on the lock.
 func (lr *LockResolver) resolvePessimisticLock(bo *retry.Backoffer, l *Lock) error {
 	metrics.LockResolverCountWithResolveLocks.Inc()
 	// The lock has been resolved by getTxnStatusFromLock.
diff --git a/txnkv/txnsnapshot/scan.go b/txnkv/txnsnapshot/scan.go
index 221cf59..01662f1 100644
--- a/txnkv/txnsnapshot/scan.go
+++ b/txnkv/txnsnapshot/scan.go
@@ -146,7 +146,7 @@ func (s *Scanner) Next() error {
 		}
 
 		current := s.cache[s.idx]
-		if (!s.reverse && (len(s.endKey) > 0 && kv.CmpKey(current.Key, s.endKey) >= 0)) ||
+		if (!s.reverse && len(s.endKey) > 0 && kv.CmpKey(current.Key, s.endKey) >= 0) ||
 			(s.reverse && len(s.nextStartKey) > 0 && kv.CmpKey(current.Key, s.nextStartKey) < 0) {
 			s.eof = true
 			s.Close()
diff --git a/txnkv/txnsnapshot/snapshot.go b/txnkv/txnsnapshot/snapshot.go
index ef293e6..ab29655 100644
--- a/txnkv/txnsnapshot/snapshot.go
+++ b/txnkv/txnsnapshot/snapshot.go
@@ -118,6 +118,7 @@ type KVSnapshot struct {
 	resolvedLocks   util.TSSet
 	committedLocks  util.TSSet
 	scanBatchSize   int
+	readTimeout     time.Duration
 
 	// Cache the result of BatchGet.
 	// The invariance is that calling BatchGet multiple times using the same start ts,
@@ -387,6 +388,7 @@ func (s *KVSnapshot) batchGetSingleRegion(bo *retry.Backoffer, batch batchKeys,
 
 	pending := batch.keys
 	var resolvingRecordToken *int
+	useConfigurableKVTimeout := true
 	for {
 		s.mu.RLock()
 		req := tikvrpc.NewReplicaReadRequest(tikvrpc.CmdBatchGet, &kvrpcpb.BatchGetRequest{
@@ -416,6 +418,12 @@ func (s *KVSnapshot) batchGetSingleRegion(bo *retry.Backoffer, batch batchKeys,
 		if isStaleness {
 			req.EnableStaleRead()
 		}
+		timeout := client.ReadTimeoutMedium
+		if useConfigurableKVTimeout && s.readTimeout > 0 {
+			useConfigurableKVTimeout = false
+			timeout = s.readTimeout
+		}
+		req.MaxExecutionDurationMs = uint64(timeout.Milliseconds())
 		ops := make([]locate.StoreSelectorOption, 0, 2)
 		if len(matchStoreLabels) > 0 {
 			ops = append(ops, locate.WithMatchLabels(matchStoreLabels))
@@ -427,7 +435,7 @@ func (s *KVSnapshot) batchGetSingleRegion(bo *retry.Backoffer, batch batchKeys,
 			}
 			req.ReplicaReadType = readType
 		}
-		resp, _, _, err := cli.SendReqCtx(bo, req, batch.region, client.ReadTimeoutMedium, tikvrpc.TiKV, "", ops...)
+		resp, _, _, err := cli.SendReqCtx(bo, req, batch.region, timeout, tikvrpc.TiKV, "", ops...)
 		if err != nil {
 			return err
 		}
@@ -651,13 +659,20 @@ func (s *KVSnapshot) get(ctx context.Context, bo *retry.Backoffer, k []byte) ([]
 
 	var firstLock *txnlock.Lock
 	var resolvingRecordToken *int
+	useConfigurableKVTimeout := true
 	for {
 		util.EvalFailpoint("beforeSendPointGet")
 		loc, err := s.store.GetRegionCache().LocateKey(bo, k)
 		if err != nil {
 			return nil, err
 		}
-		resp, _, _, err := cli.SendReqCtx(bo, req, loc.Region, client.ReadTimeoutShort, tikvrpc.TiKV, "", ops...)
+		timeout := client.ReadTimeoutShort
+		if useConfigurableKVTimeout && s.readTimeout > 0 {
+			useConfigurableKVTimeout = false
+			timeout = s.readTimeout
+		}
+		req.MaxExecutionDurationMs = uint64(timeout.Milliseconds())
+		resp, _, _, err := cli.SendReqCtx(bo, req, loc.Region, timeout, tikvrpc.TiKV, "", ops...)
 		if err != nil {
 			return nil, err
 		}
@@ -771,8 +786,8 @@ func (s *KVSnapshot) Iter(k []byte, upperBound []byte) (unionstore.Iterator, err
 }
 
 // IterReverse creates a reversed Iterator positioned on the first entry which key is less than k.
-func (s *KVSnapshot) IterReverse(k []byte) (unionstore.Iterator, error) {
-	scanner, err := newScanner(s, nil, k, s.scanBatchSize, true)
+func (s *KVSnapshot) IterReverse(k, lowerBound []byte) (unionstore.Iterator, error) {
+	scanner, err := newScanner(s, lowerBound, k, s.scanBatchSize, true)
 	return scanner, err
 }
 
@@ -984,6 +999,16 @@ func (s *KVSnapshot) mergeRegionRequestStats(stats map[tikvrpc.CmdType]*locate.R
 	}
 }
 
+// SetKVReadTimeout sets timeout for individual KV read operations under this snapshot
+func (s *KVSnapshot) SetKVReadTimeout(readTimeout time.Duration) {
+	s.readTimeout = readTimeout
+}
+
+// GetKVReadTimeout returns timeout for individual KV read operations under this snapshot or 0 if timeout is not set
+func (s *KVSnapshot) GetKVReadTimeout() time.Duration {
+	return s.readTimeout
+}
+
 func (s *KVSnapshot) getResolveLockDetail() *util.ResolveLockDetail {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
diff --git a/util/request_source.go b/util/request_source.go
index 60337e7..97d3b83 100644
--- a/util/request_source.go
+++ b/util/request_source.go
@@ -1,3 +1,17 @@
+// Copyright 2023 TiKV Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package util
 
 import (
@@ -30,15 +44,15 @@ const (
 // explicit source types.
 const (
 	ExplicitTypeEmpty      = ""
-	ExplicitTypeDefault    = "default"
 	ExplicitTypeLightning  = "lightning"
 	ExplicitTypeBR         = "br"
 	ExplicitTypeDumpling   = "dumpling"
 	ExplicitTypeBackground = "background"
+	ExplicitTypeDDL        = "ddl"
 )
 
 // ExplicitTypeList is the list of all explicit source types.
-var ExplicitTypeList = []string{ExplicitTypeEmpty, ExplicitTypeDefault, ExplicitTypeLightning, ExplicitTypeBR, ExplicitTypeDumpling, ExplicitTypeBackground}
+var ExplicitTypeList = []string{ExplicitTypeEmpty, ExplicitTypeLightning, ExplicitTypeBR, ExplicitTypeDumpling, ExplicitTypeBackground, ExplicitTypeDDL}
 
 const (
 	// InternalRequest is the scope of internal queries
@@ -83,6 +97,25 @@ func WithInternalSourceType(ctx context.Context, source string) context.Context
 	})
 }
 
+// WithInternalSourceAndTaskType create context with internal source and task name.
+func WithInternalSourceAndTaskType(ctx context.Context, source, taskName string) context.Context {
+	return context.WithValue(ctx, RequestSourceKey, RequestSource{
+		RequestSourceInternal:     true,
+		RequestSourceType:         source,
+		ExplicitRequestSourceType: taskName,
+	})
+}
+
+// BuildRequestSource builds a request_source from internal, source and explicitSource.
+func BuildRequestSource(internal bool, source, explicitSource string) string {
+	requestSource := RequestSource{
+		RequestSourceInternal:     internal,
+		RequestSourceType:         source,
+		ExplicitRequestSourceType: explicitSource,
+	}
+	return requestSource.GetRequestSource()
+}
+
 // IsRequestSourceInternal checks whether the input request source type is internal type.
 func IsRequestSourceInternal(reqSrc *RequestSource) bool {
 	isInternal := false
@@ -95,24 +128,26 @@ func IsRequestSourceInternal(reqSrc *RequestSource) bool {
 // GetRequestSource gets the request_source field of the request.
 func (r *RequestSource) GetRequestSource() string {
 	source := SourceUnknown
-	explicitSourceType := ExplicitTypeDefault
 	origin := ExternalRequest
 	if r == nil || (len(r.RequestSourceType) == 0 && len(r.ExplicitRequestSourceType) == 0) {
 		// if r.RequestSourceType and r.ExplicitRequestSourceType are not set, it's mostly possible that r.RequestSourceInternal is not set
 		// to avoid internal requests be marked as external(default value), return unknown source here.
-		return strings.Join([]string{source, explicitSourceType}, "_")
+		return source
 	}
-
+	if r.RequestSourceInternal {
+		origin = InternalRequest
+	}
+	labelList := make([]string, 0, 3)
+	labelList = append(labelList, origin)
 	if len(r.RequestSourceType) > 0 {
 		source = r.RequestSourceType
 	}
-	if len(r.ExplicitRequestSourceType) > 0 {
-		explicitSourceType = r.ExplicitRequestSourceType
+	labelList = append(labelList, source)
+	if len(r.ExplicitRequestSourceType) > 0 && r.ExplicitRequestSourceType != r.RequestSourceType {
+		labelList = append(labelList, r.ExplicitRequestSourceType)
 	}
-	if r.RequestSourceInternal {
-		origin = InternalRequest
-	}
-	return strings.Join([]string{origin, source, explicitSourceType}, "_")
+
+	return strings.Join(labelList, "_")
 }
 
 // RequestSourceFromCtx extract source from passed context.
@@ -135,9 +170,9 @@ type resourceGroupNameKeyType struct{}
 // ResourceGroupNameKey is used as the key of request source type in context.
 var resourceGroupNameKey = resourceGroupNameKeyType{}
 
-// WithResouceGroupName return a copy of the given context with a associated
-// reosurce group name.
-func WithResouceGroupName(ctx context.Context, groupName string) context.Context {
+// WithResourceGroupName return a copy of the given context with a associated
+// resource group name.
+func WithResourceGroupName(ctx context.Context, groupName string) context.Context {
 	return context.WithValue(ctx, resourceGroupNameKey, groupName)
 }
 


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 84a6c3b..72b64e0 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -13,9 +13,9 @@ jobs:
     - uses: actions/checkout@v2
 
     - name: Set up Go
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v4
       with:
-        go-version: 1.20.2
+        go-version: 1.21.0
 
     - name: Test
       run: go test ./...
@@ -26,9 +26,9 @@ jobs:
     - uses: actions/checkout@v2
 
     - name: Set up Go
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v4
       with:
-        go-version: 1.20.2
+        go-version: 1.21.0
 
     - name: Test with race
       run: go test -race ./...
@@ -40,9 +40,9 @@ jobs:
         uses: actions/checkout@v2
 
       - name: Set up Go
-        uses: actions/setup-go@v2
+        uses: actions/setup-go@v4
         with:
-          go-version: 1.20.2
+          go-version: 1.21.0
 
       - name: Lint
         uses: golangci/golangci-lint-action@v3
diff --git a/integration_tests/go.mod b/integration_tests/go.mod
index b72c524..97636a2 100644
--- a/integration_tests/go.mod
+++ b/integration_tests/go.mod
@@ -6,13 +6,13 @@ require (
 	github.com/ninedraft/israce v0.0.3
 	github.com/pingcap/errors v0.11.5-0.20221009092201-b66cddb77c32
 	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c
-	github.com/pingcap/kvproto v0.0.0-20230713060620-89756bd21be1
+	github.com/pingcap/kvproto v0.0.0-20230720094213-a3b4a77b4333
 	github.com/pingcap/tidb v1.1.0-beta.0.20230619015310-8b1006f1af04
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.8.4
 	github.com/tidwall/gjson v1.14.1
-	github.com/tikv/client-go/v2 v2.0.8-0.20230615161845-b32f340d0609
-	github.com/tikv/pd/client v0.0.0-20230717031845-58eb48b840e7
+	github.com/tikv/client-go/v2 v2.0.8-0.20230714052714-85fc8f337565
+	github.com/tikv/pd/client v0.0.0-20230724080549-de985b8e0afc
 	go.uber.org/goleak v1.2.1
 )
 
@@ -66,13 +66,13 @@ require (
 	github.com/pingcap/log v1.1.1-0.20230317032135-a0d097d16e22 // indirect
 	github.com/pingcap/sysutil v1.0.1-0.20230407040306-fb007c5aff21 // indirect
 	github.com/pingcap/tidb/parser v0.0.0-20230619015310-8b1006f1af04 // indirect
-	github.com/pingcap/tipb v0.0.0-20230602100112-acb7942db1ca // indirect
+	github.com/pingcap/tipb v0.0.0-20230607071926-bda24015c2d6 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b // indirect
-	github.com/prometheus/client_golang v1.15.1 // indirect
+	github.com/prometheus/client_golang v1.16.0 // indirect
 	github.com/prometheus/client_model v0.4.0 // indirect
 	github.com/prometheus/common v0.44.0 // indirect
-	github.com/prometheus/procfs v0.10.1 // indirect
+	github.com/prometheus/procfs v0.11.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	github.com/rogpeppe/go-internal v1.10.0 // indirect
 	github.com/sasha-s/go-deadlock v0.2.0 // indirect
@@ -94,14 +94,14 @@ require (
 	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.24.0 // indirect
-	golang.org/x/crypto v0.9.0 // indirect
-	golang.org/x/exp v0.0.0-20230519143937-03e91628a987 // indirect
-	golang.org/x/net v0.10.0 // indirect
-	golang.org/x/sync v0.2.0 // indirect
-	golang.org/x/sys v0.8.0 // indirect
-	golang.org/x/text v0.9.0 // indirect
+	golang.org/x/crypto v0.11.0 // indirect
+	golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 // indirect
+	golang.org/x/net v0.11.0 // indirect
+	golang.org/x/sync v0.3.0 // indirect
+	golang.org/x/sys v0.10.0 // indirect
+	golang.org/x/text v0.11.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
-	golang.org/x/tools v0.9.3 // indirect
+	golang.org/x/tools v0.10.0 // indirect
 	google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 // indirect
 	google.golang.org/grpc v1.54.0 // indirect
 	google.golang.org/protobuf v1.30.0 // indirect
@@ -112,5 +112,7 @@ require (
 
 replace (
 	github.com/go-ldap/ldap/v3 => github.com/YangKeao/ldap/v3 v3.4.5-0.20230421065457-369a3bab1117
+	github.com/pingcap/tidb => github.com/HuSharp/tidb v1.1.0-beta.0.20230726045237-a2b0085ad7c5
+	github.com/pingcap/tidb/parser => github.com/HuSharp/tidb/parser v0.0.0-20230726045237-a2b0085ad7c5
 	github.com/tikv/client-go/v2 => ../
 )
diff --git a/integration_tests/go.sum b/integration_tests/go.sum
index bc1b508..ac10fa0 100644
--- a/integration_tests/go.sum
+++ b/integration_tests/go.sum
@@ -20,6 +20,10 @@ github.com/CloudyKit/fastprinter v0.0.0-20170127035650-74b38d55f37a/go.mod h1:EF
 github.com/CloudyKit/jet v2.1.3-0.20180809161101-62edd43e4f88+incompatible/go.mod h1:HPYO+50pSWkPoj9Q/eq0aRGByCL6ScRlUmiEX5Zgm+w=
 github.com/DataDog/zstd v1.4.5 h1:EndNeuB0l9syBZhut0wns3gV1hL8zX8LIu6ZiVHWLIQ=
 github.com/HdrHistogram/hdrhistogram-go v1.1.2 h1:5IcZpTvzydCQeHzK4Ef/D5rrSqwxob0t8PQPMybUNFM=
+github.com/HuSharp/tidb v1.1.0-beta.0.20230726045237-a2b0085ad7c5 h1:wSOvDYbKkvHjlWWFBihIoeJ5yBc1jZe9Ehkku3Jn8cA=
+github.com/HuSharp/tidb v1.1.0-beta.0.20230726045237-a2b0085ad7c5/go.mod h1:C3tuWINS2/Vt/gxZ0OLdGI2x5crlN8E3/qNJJkIIkTI=
+github.com/HuSharp/tidb/parser v0.0.0-20230726045237-a2b0085ad7c5 h1:bxwmPI7ambmbOAaozdYz81HFpIeu6ctWo7TiXfOGE14=
+github.com/HuSharp/tidb/parser v0.0.0-20230726045237-a2b0085ad7c5/go.mod h1:ENXEsaVS6N3CTMpL4txc6m93y6XaztF9W4SFLjhPWJg=
 github.com/Joker/hpp v1.0.0/go.mod h1:8x5n+M1Hp5hC0g8okX3sR3vFQwynaX/UgSOM9MeBKzY=
 github.com/Joker/jade v1.0.1-0.20190614124447-d475f43051e7/go.mod h1:6E6s8o2AE4KhCrqr6GRJjdC/gNfTdxkIXvuGZZda2VM=
 github.com/Masterminds/semver v1.5.0 h1:H65muMkzWKEuNDnfl9d70GUjFniHKHRbFPGBuZ3QEww=
@@ -97,7 +101,7 @@ github.com/danjacques/gofslock v0.0.0-20220131014315-6e321f4509c8/go.mod h1:VT5E
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/decred/dcrd/dcrec/secp256k1/v4 v4.1.0 h1:HbphB4TFFXpv7MNrT52FGrrgVXF1owhMVTHFZIlnvd4=
+github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 h1:8UrgZ3GkP4i/CLijOJx79Yu+etlyjdBU4sfcs2WYQMs=
 github.com/dgraph-io/badger v1.6.0/go.mod h1:zwt7syl517jmP8s94KqSxTlM6IMsdhYy6psNgSztDR4=
 github.com/dgraph-io/ristretto v0.1.1 h1:6CWw5tJNgpegArSHpNHJKldNeq03FQCwYvfMVWajOK8=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
@@ -148,7 +152,7 @@ github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/me
 github.com/gobwas/httphead v0.0.0-20180130184737-2c6c146eadee/go.mod h1:L0fX3K22YWvt/FAX9NnzrNzcI4wNYi9Yku4O0LKYflo=
 github.com/gobwas/pool v0.2.0/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw=
 github.com/gobwas/ws v1.0.2/go.mod h1:szmBTxLgaFppYjEmNtny/v3w89xOydFnnZMcgRRu/EM=
-github.com/goccy/go-json v0.9.11 h1:/pAaQDLHEoCq/5FFmSKBswWmK6H0e8g4159Kc/X/nqk=
+github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/gogo/googleapis v0.0.0-20180223154316-0cd9801be74a/go.mod h1:gf4bu3Q80BeJ6H1S1vYPm8/ELATdvryBaNFGgqEef3s=
 github.com/gogo/protobuf v0.0.0-20171007142547-342cbe0a0415/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
@@ -286,8 +290,8 @@ github.com/lestrrat-go/blackmagic v1.0.1 h1:lS5Zts+5HIC/8og6cGHb0uCcNCa3OUt1ygh3
 github.com/lestrrat-go/httpcc v1.0.1 h1:ydWCStUeJLkpYyjLDHihupbn2tYmZ7m22BGkcvZZrIE=
 github.com/lestrrat-go/httprc v1.0.4 h1:bAZymwoZQb+Oq8MEbyipag7iSq6YIga8Wj6GOiJGdI8=
 github.com/lestrrat-go/iter v1.0.2 h1:gMXo1q4c2pHmC3dn8LzRhJfP1ceCbgSiT9lUydIzltI=
-github.com/lestrrat-go/jwx/v2 v2.0.6 h1:RlyYNLV892Ed7+FTfj1ROoF6x7WxL965PGTHso/60G0=
-github.com/lestrrat-go/option v1.0.0 h1:WqAWL8kh8VcSoD6xjSH34/1m8yxluXQbDeKNfvFeEO4=
+github.com/lestrrat-go/jwx/v2 v2.0.11 h1:ViHMnaMeaO0qV16RZWBHM7GTrAnX2aFLVKofc7FuKLQ=
+github.com/lestrrat-go/option v1.0.1 h1:oAzP2fvZGQKWkvHa1/SAcFolBEca1oN+mQ7eooNBEYU=
 github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=
 github.com/lufia/plan9stats v0.0.0-20230326075908-cb1d2100619a h1:N9zuLhTvBSRt0gWSiJswwQ2HqDmtX/ZCDJURnKUt1Ik=
 github.com/lufia/plan9stats v0.0.0-20230326075908-cb1d2100619a/go.mod h1:JKx41uQRwqlTZabZc+kILPrO/3jlKnQ2Z8b7YiVw5cE=
@@ -355,24 +359,20 @@ github.com/pingcap/errors v0.11.5-0.20221009092201-b66cddb77c32 h1:m5ZsBa5o/0Ckz
 github.com/pingcap/errors v0.11.5-0.20221009092201-b66cddb77c32/go.mod h1:X2r9ueLEUZgtx2cIogM0v4Zj5uvvzhuuiu7Pn8HzMPg=
 github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c h1:CgbKAHto5CQgWM9fSBIvaxsJHuGP0uM74HXtv3MyyGQ=
 github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c/go.mod h1:4qGtCB0QK0wBzKtFEGDhxXnSnbQApw1gc9siScUl8ew=
-github.com/pingcap/fn v0.0.0-20200306044125-d5540d389059 h1:Pe2LbxRmbTfAoKJ65bZLmhahmvHm7n9DUxGRQT00208=
+github.com/pingcap/fn v1.0.0 h1:CyA6AxcOZkQh52wIqYlAmaVmF6EvrcqFywP463pjA8g=
 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=
-github.com/pingcap/kvproto v0.0.0-20230713060620-89756bd21be1 h1:sC3XRNNBQNjFJGRtSzJRvqi2aDLFOsQoCHItr9rbbY8=
-github.com/pingcap/kvproto v0.0.0-20230713060620-89756bd21be1/go.mod h1:r0q/CFcwvyeRhKtoqzmWMBebrtpIziQQ9vR+JKh1knc=
+github.com/pingcap/kvproto v0.0.0-20230720094213-a3b4a77b4333 h1:A6Wqgq0uMw51UiRAH27TVN0QlzVR5CVtV6fTQSAmvKM=
+github.com/pingcap/kvproto v0.0.0-20230720094213-a3b4a77b4333/go.mod h1:r0q/CFcwvyeRhKtoqzmWMBebrtpIziQQ9vR+JKh1knc=
 github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuRurdGxZXBz0At+9avep+ub7U1AGYLIMM=
 github.com/pingcap/log v1.1.0/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
 github.com/pingcap/log v1.1.1-0.20230317032135-a0d097d16e22 h1:2SOzvGvE8beiC1Y4g9Onkvu6UmuBBOeWRGQEjJaT/JY=
 github.com/pingcap/log v1.1.1-0.20230317032135-a0d097d16e22/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
 github.com/pingcap/sysutil v1.0.1-0.20230407040306-fb007c5aff21 h1:QV6jqlfOkh8hqvEAgwBZa+4bSgO0EeKC7s5c6Luam2I=
 github.com/pingcap/sysutil v1.0.1-0.20230407040306-fb007c5aff21/go.mod h1:QYnjfA95ZaMefyl1NO8oPtKeb8pYUdnDVhQgf+qdpjM=
-github.com/pingcap/tidb v1.1.0-beta.0.20230619015310-8b1006f1af04 h1:rBlNvmdFTB+WPFCvCSBQwK8PtoZA7JVblygbn3X8mmg=
-github.com/pingcap/tidb v1.1.0-beta.0.20230619015310-8b1006f1af04/go.mod h1:Pi4ObsVr4eRNxCyFXsho0uzA/+ZDSjB0ASvaUI3ECUI=
-github.com/pingcap/tidb/parser v0.0.0-20230619015310-8b1006f1af04 h1:pNz5l+3UFAu040skaE92eyVSlnAPMaCQBRW3E746lMs=
-github.com/pingcap/tidb/parser v0.0.0-20230619015310-8b1006f1af04/go.mod h1:QKyCQPh1u6N7yXpPElNJzzWL7J70duuRo45GMH0FMNk=
-github.com/pingcap/tipb v0.0.0-20230602100112-acb7942db1ca h1:J2HQyR5v1AcoBzx5/AYJW9XFSIl6si6YoC6yGI1W89c=
-github.com/pingcap/tipb v0.0.0-20230602100112-acb7942db1ca/go.mod h1:A7mrd7WHBl1o63LE2bIBGEJMTNWXqhgmYiOvMLxozfs=
+github.com/pingcap/tipb v0.0.0-20230607071926-bda24015c2d6 h1:D79RE4RVhq2ic8sqDSv7QdL0tT5aZV3CaCXUAT41iWc=
+github.com/pingcap/tipb v0.0.0-20230607071926-bda24015c2d6/go.mod h1:A7mrd7WHBl1o63LE2bIBGEJMTNWXqhgmYiOvMLxozfs=
 github.com/pkg/browser v0.0.0-20210115035449-ce105d075bb4 h1:Qj1ukM4GlMWXNdMBuXcXfz/Kw9s1qm0CLY32QxuSImI=
 github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -389,8 +389,8 @@ github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXP
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
 github.com/prometheus/client_golang v1.7.1/go.mod h1:PY5Wy2awLA44sXw4AOSfFBetzPP4j5+D6mVACh+pe2M=
 github.com/prometheus/client_golang v1.11.0/go.mod h1:Z6t4BnS23TR94PD6BsDNk8yVqroYurpAkEiz0P2BEV0=
-github.com/prometheus/client_golang v1.15.1 h1:8tXpTmJbyH5lydzFPoxSIJ0J46jdh3tylbvM1xCv0LI=
-github.com/prometheus/client_golang v1.15.1/go.mod h1:e9yaBhRPU2pPNsZwE+JdQl0KEt1N9XgF6zxWmaC0xOk=
+github.com/prometheus/client_golang v1.16.0 h1:yk/hx9hDbrGHovbci4BY+pRMfSuuat626eFsHb7tmT8=
+github.com/prometheus/client_golang v1.16.0/go.mod h1:Zsulrv/L9oM40tJ7T815tM89lFEugiJ9HzIqaAx4LKc=
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
@@ -407,8 +407,8 @@ github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R
 github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
 github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
-github.com/prometheus/procfs v0.10.1 h1:kYK1Va/YMlutzCGazswoHKo//tZVlFpKYh+PymziUAg=
-github.com/prometheus/procfs v0.10.1/go.mod h1:nwNm2aOCAYw8uTR/9bWRREkZFxAUcWzPHWJq+XBB/FM=
+github.com/prometheus/procfs v0.11.0 h1:5EAgkfkMl659uZPbe9AS2N68a7Cc1TJbPEuGzFuRbyk=
+github.com/prometheus/procfs v0.11.0/go.mod h1:nwNm2aOCAYw8uTR/9bWRREkZFxAUcWzPHWJq+XBB/FM=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/rivo/uniseg v0.4.4 h1:8TfxU8dW6PdqD27gjM8MVNuicgxIjxpm4K7x4jp8sis=
@@ -422,6 +422,7 @@ github.com/ryanuber/columnize v2.1.0+incompatible/go.mod h1:sm1tb6uqfes/u+d4ooFo
 github.com/sasha-s/go-deadlock v0.2.0 h1:lMqc+fUb7RrFS3gQLtoQsJ7/6TV/pAIFvBsqX73DK8Y=
 github.com/sasha-s/go-deadlock v0.2.0/go.mod h1:StQn567HiB1fF2yJ44N9au7wOhrPS3iZqiDbRupzT10=
 github.com/sclevine/agouti v3.0.0+incompatible/go.mod h1:b4WX9W9L1sfQKXeJf1mUTLZKJ48R1S7H23Ji7oFO5Bw=
+github.com/segmentio/asm v1.2.0 h1:9BQrFxC+YOHJlTlHGkTrFWf59nbL3XnCoFLTwDCI7ys=
 github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
 github.com/shirou/gopsutil/v3 v3.21.12/go.mod h1:BToYZVTlSVlfazpDDYFnsVZLaoRG+g8ufT6fPQLdJzA=
 github.com/shirou/gopsutil/v3 v3.23.5 h1:5SgDCeQ0KW0S4N0znjeM/eFHXXOKyv2dVNgRq/c9P6Y=
@@ -474,8 +475,8 @@ github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
 github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
 github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
 github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
-github.com/tikv/pd/client v0.0.0-20230717031845-58eb48b840e7 h1:N1m2YFtQAZEkk5ilmsMqauELh29TJQrX2JbcNhdnm7k=
-github.com/tikv/pd/client v0.0.0-20230717031845-58eb48b840e7/go.mod h1:YAB/lbBQtURR2A1tyqGQYEkBVCKHVqyZFefN7CPIZi4=
+github.com/tikv/pd/client v0.0.0-20230724080549-de985b8e0afc h1:IUg0j2nWoGYj3FQ3vA3vg97fPSpJEZQrDpgF8RkMLEU=
+github.com/tikv/pd/client v0.0.0-20230724080549-de985b8e0afc/go.mod h1:wfHRc4iYaqJiOQZCHcrF+r4hYnkGDaYWDfcicee//pc=
 github.com/tklauser/go-sysconf v0.3.9/go.mod h1:11DU/5sG7UexIrp/O6g35hrWzu0JxlwQ3LSFUzyeuhs=
 github.com/tklauser/go-sysconf v0.3.11 h1:89WgdJhk5SNwJfu+GKyYveZ4IaJ7xAkecBo+KdJV0CM=
 github.com/tklauser/go-sysconf v0.3.11/go.mod h1:GqXfhXY3kiPa0nAXPDIQIWzJbMCB7AmcWpGR8lSZfqI=
@@ -576,12 +577,12 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.7.0/go.mod h1:pYwdfH91IfpZVANVyUOhSIPZaFoJGxTFbZhFTx+dXZU=
-golang.org/x/crypto v0.9.0 h1:LF6fAI+IutBocDJ2OT0Q1g8plpYljMZ4+lty+dsqw3g=
-golang.org/x/crypto v0.9.0/go.mod h1:yrmDGqONDYtNj3tH8X9dzUun2m2lzPa9ngI6/RUPGR0=
+golang.org/x/crypto v0.11.0 h1:6Ewdq3tDic1mg5xRO4milcWCfMVQhI4NkqWWvqejpuA=
+golang.org/x/crypto v0.11.0/go.mod h1:xgJhtzW8F9jGdVFWZESrid1U1bjeNy4zgy5cRr/CIio=
 golang.org/x/exp v0.0.0-20180321215751-8460e604b9de/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
-golang.org/x/exp v0.0.0-20230519143937-03e91628a987 h1:3xJIFvzUFbu4ls0BTBYcgbCGhA63eAOEMxIHugyXJqA=
-golang.org/x/exp v0.0.0-20230519143937-03e91628a987/go.mod h1:V1LtkGg67GoY2N1AnLN78QLrzxkLyJw7RJb1gzOOz9w=
+golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 h1:QLureRX3moex6NVu/Lr4MGakp9FdA7sBHGBmvRW7NaM=
+golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4/go.mod h1:FXUEEKJgO7OQYeo8N01OfiKP8RXMtf6e8aTskBGqWdc=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
@@ -622,8 +623,8 @@ golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96b
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
-golang.org/x/net v0.10.0 h1:X2//UzNDwYmtCLn7To6G58Wr6f5ahEAQgKNzv9Y951M=
-golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
+golang.org/x/net v0.11.0 h1:Gi2tvZIJyBtO9SDr1q9h5hEQCp/4L2RQ+ar0qjx2oNU=
+golang.org/x/net v0.11.0/go.mod h1:2L/ixqYpgIVXmeoSA/4Lu7BzTG4KIyPIryS4IsOd1oQ=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -639,8 +640,8 @@ golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.2.0 h1:PUR+T4wwASmuSTYdKjYHI5TD22Wy5ogLU5qZCOLxBrI=
-golang.org/x/sync v0.2.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E=
+golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -678,13 +679,14 @@ golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.8.0 h1:EBmGv8NaZBZTWvrbjNoL6HVt+IVy3QDQpJs7VRIw3tU=
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.10.0 h1:SqMFp9UcQJZa+pmYuAKjd9xq1f0j5rLcDIk0mj4qAsA=
+golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
 golang.org/x/term v0.6.0/go.mod h1:m6U89DPEgQRMq3DNkDClhWw02AUbt2daBVO4cn4Hv9U=
-golang.org/x/term v0.8.0 h1:n5xxQn2i3PC0yLAbjTpNT85q/Kgzcr2gIoX9OrJUols=
+golang.org/x/term v0.10.0 h1:3R7pNqamzBraeqj/Tj8qt1aQ2HpmlC+Cx/qL/7hn4/c=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
@@ -692,8 +694,8 @@ golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
-golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
-golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/text v0.11.0 h1:LAntKIrcmeSKERyiOh0XMV39LXS8IE9UL2yP7+f5ij4=
+golang.org/x/text v0.11.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4=
 golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -721,8 +723,8 @@ golang.org/x/tools v0.1.2/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
-golang.org/x/tools v0.9.3 h1:Gn1I8+64MsuTb/HpH+LmQtNas23LhUVr3rYZ0eKuaMM=
-golang.org/x/tools v0.9.3/go.mod h1:owI94Op576fPu3cIGQeHs3joujW/2Oc6MtlxbF5dfNc=
+golang.org/x/tools v0.10.0 h1:tvDr/iQoUqNdohiYm0LmmKcBk+q86lb9EprIUFhHHGg=
+golang.org/x/tools v0.10.0/go.mod h1:UJwyiVBsOA2uwvK/e5OY3GTpDUJriEd+/YlqAwLPmyM=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
diff --git a/integration_tests/lock_test.go b/integration_tests/lock_test.go
index bafe216..89bead1 100644
--- a/integration_tests/lock_test.go
+++ b/integration_tests/lock_test.go
@@ -1472,52 +1472,81 @@ func (s *testLockWithTiKVSuite) TestBatchResolveLocks() {
 
 	s.NoError(failpoint.Enable("tikvclient/beforeAsyncPessimisticRollback", `return("skip")`))
 	s.NoError(failpoint.Enable("tikvclient/beforeCommitSecondaries", `return("skip")`))
-	s.NoError(failpoint.Enable("tikvclient/twoPCRequestBatchSizeLimit", `return("skip")`))
+	s.NoError(failpoint.Enable("tikvclient/twoPCRequestBatchSizeLimit", `return`))
+	s.NoError(failpoint.Enable("tikvclient/onRollback", `return("skipRollbackPessimisticLock")`))
 	defer func() {
 		s.NoError(failpoint.Disable("tikvclient/beforeAsyncPessimisticRollback"))
 		s.NoError(failpoint.Disable("tikvclient/beforeCommitSecondaries"))
 		s.NoError(failpoint.Disable("tikvclient/twoPCRequestBatchSizeLimit"))
+		s.NoError(failpoint.Disable("tikvclient/onRollback"))
 	}()
 
-	k1, k2, k3 := []byte("k1"), []byte("k2"), []byte("k3")
+	k1, k2, k3, k4 := []byte("k1"), []byte("k2"), []byte("k3"), []byte("k4")
 	v2, v3 := []byte("v2"), []byte("v3")
 
 	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
 
-	txn, err := s.store.Begin()
+	txn1, err := s.store.Begin()
 	s.NoError(err)
-	txn.SetPessimistic(true)
+	txn1.SetPessimistic(true)
 
 	{
 		// Produce write conflict on key k2
-		txn2, err := s.store.Begin()
+		helperTxn, err := s.store.Begin()
 		s.NoError(err)
-		s.NoError(txn2.Set(k2, []byte("v0")))
-		s.NoError(txn2.Commit(ctx))
+		s.NoError(helperTxn.Set(k2, []byte("v0")))
+		s.NoError(helperTxn.Commit(ctx))
 	}
 
-	lockCtx := kv.NewLockCtx(txn.StartTS(), 200, time.Now())
-	err = txn.LockKeys(ctx, lockCtx, k1, k2)
+	lockCtx := kv.NewLockCtx(txn1.StartTS(), 200, time.Now())
+	err = txn1.LockKeys(ctx, lockCtx, k1, k2)
 	s.IsType(&tikverr.ErrWriteConflict{}, errors.Cause(err))
 
-	// k1 has txn's stale pessimistic lock now.
+	// k1 has txn1's stale pessimistic lock now.
 
 	forUpdateTS, err := s.store.CurrentTimestamp(oracle.GlobalTxnScope)
 	s.NoError(err)
 	lockCtx = kv.NewLockCtx(forUpdateTS, 200, time.Now())
-	s.NoError(txn.LockKeys(ctx, lockCtx, k2, k3))
+	s.NoError(txn1.LockKeys(ctx, lockCtx, k2, k3))
 
-	s.NoError(txn.Set(k2, v2))
-	s.NoError(txn.Set(k3, v3))
-	s.NoError(txn.Commit(ctx))
+	s.NoError(txn1.Set(k2, v2))
+	s.NoError(txn1.Set(k3, v3))
+	s.NoError(txn1.Commit(ctx))
 
-	// k3 has txn's stale prewrite lock now.
+	// k3 has txn1's stale prewrite lock now.
 
-	// Perform ScanLock - BatchResolveLock.
+	txn2, err := s.store.Begin()
+	txn2.SetPessimistic(true)
+	s.NoError(err)
+	lockCtx = kv.NewLockCtx(txn1.StartTS(), 200, time.Now())
+	err = txn2.LockKeys(ctx, lockCtx, k4)
+	s.NoError(err)
+	s.NoError(txn2.Rollback())
+
+	// k4 has txn2's stale primary pessimistic lock now.
 	currentTS, err := s.store.CurrentTimestamp(oracle.GlobalTxnScope)
+
+	remainingLocks, err := s.store.ScanLocks(ctx, []byte("k"), []byte("l"), currentTS)
+	s.NoError(err)
+
+	s.Len(remainingLocks, 3)
+	s.Equal(remainingLocks[0].Key, k1)
+	s.Equal(remainingLocks[0].LockType, kvrpcpb.Op_PessimisticLock)
+	s.Equal(remainingLocks[1].Key, k3)
+	s.Equal(remainingLocks[1].LockType, kvrpcpb.Op_Put)
+	s.Equal(remainingLocks[2].Key, k4)
+	s.Equal(remainingLocks[2].LockType, kvrpcpb.Op_PessimisticLock)
+	s.Equal(remainingLocks[2].Primary, k4)
+
+	// Perform ScanLock - BatchResolveLock.
 	s.NoError(err)
 	s.NoError(s.store.GCResolveLockPhase(ctx, currentTS, 1))
 
+	// Do ScanLock again to make sure no locks are left.
+	remainingLocks, err = s.store.ScanLocks(ctx, []byte("k"), []byte("l"), currentTS)
+	s.NoError(err)
+	s.Empty(remainingLocks)
+
 	// Check data consistency
 	readTS, err := s.store.CurrentTimestamp(oracle.GlobalTxnScope)
 	snapshot := s.store.GetSnapshot(readTS)
diff --git a/integration_tests/split_test.go b/integration_tests/split_test.go
index b913e8f..5580882 100644
--- a/integration_tests/split_test.go
+++ b/integration_tests/split_test.go
@@ -291,6 +291,10 @@ func (c *mockPDClient) UpdateOption(option pd.DynamicOption, value interface{})
 	return nil
 }
 
+func (c *mockPDClient) GetAllKeyspaces(ctx context.Context, startID uint32, limit uint32) ([]*keyspacepb.KeyspaceMeta, error) {
+	return nil, nil
+}
+
 func (c *mockPDClient) LoadKeyspace(ctx context.Context, name string) (*keyspacepb.KeyspaceMeta, error) {
 	return nil, nil
 }
diff --git a/internal/client/main_test.go b/internal/client/main_test.go
index c4d2be2..6a22714 100644
--- a/internal/client/main_test.go
+++ b/internal/client/main_test.go
@@ -26,6 +26,7 @@ func TestMain(m *testing.M) {
 	opts := []goleak.Option{
 		goleak.IgnoreTopFunction("google.golang.org/grpc.(*ClientConn).WaitForStateChange"),
 		goleak.IgnoreTopFunction("github.com/tikv/client-go/v2/internal/retry.newBackoffFn.func1"),
+		goleak.IgnoreTopFunction("github.com/tikv/client-go/v2/internal/retry.(*Config).createBackoffFn.newBackoffFn.func2"),
 	}
 	goleak.VerifyTestMain(m, opts...)
 }
diff --git a/internal/locate/region_cache_test.go b/internal/locate/region_cache_test.go
index 03ef3e3..619da2d 100644
--- a/internal/locate/region_cache_test.go
+++ b/internal/locate/region_cache_test.go
@@ -966,7 +966,7 @@ func (s *testRegionCacheSuite) TestRegionEpochAheadOfTiKV() {
 	region := createSampleRegion([]byte("k1"), []byte("k2"))
 	region.meta.Id = 1
 	region.meta.RegionEpoch = &metapb.RegionEpoch{Version: 10, ConfVer: 10}
-	cache.insertRegionToCache(region)
+	cache.insertRegionToCache(region, true)
 
 	r1 := metapb.Region{Id: 1, RegionEpoch: &metapb.RegionEpoch{Version: 9, ConfVer: 10}}
 	r2 := metapb.Region{Id: 1, RegionEpoch: &metapb.RegionEpoch{Version: 10, ConfVer: 9}}
@@ -1257,7 +1257,7 @@ func (s *testRegionCacheSuite) TestPeersLenChange() {
 	filterUnavailablePeers(cpRegion)
 	region, err := newRegion(s.bo, s.cache, cpRegion)
 	s.Nil(err)
-	s.cache.insertRegionToCache(region)
+	s.cache.insertRegionToCache(region, true)
 
 	// OnSendFail should not panic
 	s.cache.OnSendFail(retry.NewNoopBackoff(context.Background()), ctx, false, errors.New("send fail"))
@@ -1293,7 +1293,7 @@ func (s *testRegionCacheSuite) TestPeersLenChangedByWitness() {
 	cpRegion := &pd.Region{Meta: cpMeta}
 	region, err := newRegion(s.bo, s.cache, cpRegion)
 	s.Nil(err)
-	s.cache.insertRegionToCache(region)
+	s.cache.insertRegionToCache(region, true)
 
 	// OnSendFail should not panic
 	s.cache.OnSendFail(retry.NewNoopBackoff(context.Background()), ctx, false, errors.New("send fail"))
@@ -1466,12 +1466,12 @@ func (s *testRegionCacheSuite) TestBuckets() {
 	fakeRegion.setStore(cachedRegion.getStore().clone())
 	// no buckets
 	fakeRegion.getStore().buckets = nil
-	s.cache.insertRegionToCache(fakeRegion)
+	s.cache.insertRegionToCache(fakeRegion, true)
 	cachedRegion = s.getRegion([]byte("a"))
 	s.Equal(defaultBuckets, cachedRegion.getStore().buckets)
 	// stale buckets
 	fakeRegion.getStore().buckets = &metapb.Buckets{Version: defaultBuckets.Version - 1}
-	s.cache.insertRegionToCache(fakeRegion)
+	s.cache.insertRegionToCache(fakeRegion, true)
 	cachedRegion = s.getRegion([]byte("a"))
 	s.Equal(defaultBuckets, cachedRegion.getStore().buckets)
 	// new buckets
@@ -1481,7 +1481,7 @@ func (s *testRegionCacheSuite) TestBuckets() {
 		Keys:     buckets.Keys,
 	}
 	fakeRegion.getStore().buckets = newBuckets
-	s.cache.insertRegionToCache(fakeRegion)
+	s.cache.insertRegionToCache(fakeRegion, true)
 	cachedRegion = s.getRegion([]byte("a"))
 	s.Equal(newBuckets, cachedRegion.getStore().buckets)
 
@@ -1614,7 +1614,7 @@ func (s *testRegionCacheSuite) TestRemoveIntersectingRegions() {
 	region, err := s.cache.loadRegion(s.bo, []byte("c"), false)
 	s.Nil(err)
 	s.Equal(region.GetID(), regions[0])
-	s.cache.insertRegionToCache(region)
+	s.cache.insertRegionToCache(region, true)
 	loc, err = s.cache.LocateKey(s.bo, []byte{'c'})
 	s.Nil(err)
 	s.Equal(loc.Region.GetID(), regions[0])
@@ -1625,7 +1625,7 @@ func (s *testRegionCacheSuite) TestRemoveIntersectingRegions() {
 	region, err = s.cache.loadRegion(s.bo, []byte("e"), false)
 	s.Nil(err)
 	s.Equal(region.GetID(), regions[0])
-	s.cache.insertRegionToCache(region)
+	s.cache.insertRegionToCache(region, true)
 	loc, err = s.cache.LocateKey(s.bo, []byte{'e'})
 	s.Nil(err)
 	s.Equal(loc.Region.GetID(), regions[0])
@@ -1739,7 +1739,7 @@ func (s *testRegionRequestToSingleStoreSuite) TestRefreshCache() {
 	v2 := region.Region.confVer + 1
 	r2 := metapb.Region{Id: region.Region.id, RegionEpoch: &metapb.RegionEpoch{Version: region.Region.ver, ConfVer: v2}, StartKey: []byte{1}}
 	st := &Store{storeID: s.store}
-	s.cache.insertRegionToCache(&Region{meta: &r2, store: unsafe.Pointer(st), lastAccess: time.Now().Unix()})
+	s.cache.insertRegionToCache(&Region{meta: &r2, store: unsafe.Pointer(st), lastAccess: time.Now().Unix()}, true)
 
 	r, _ = s.cache.scanRegionsFromCache(s.bo, []byte{}, nil, 10)
 	s.Equal(len(r), 2)
diff --git a/internal/locate/region_request3_test.go b/internal/locate/region_request3_test.go
index e05c2ca..f56b402 100644
--- a/internal/locate/region_request3_test.go
+++ b/internal/locate/region_request3_test.go
@@ -37,6 +37,7 @@ package locate
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -45,15 +46,18 @@ import (
 	"github.com/pingcap/kvproto/pkg/errorpb"
 	"github.com/pingcap/kvproto/pkg/kvrpcpb"
 	"github.com/pingcap/kvproto/pkg/metapb"
+	"github.com/pingcap/log"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/suite"
 	tikverr "github.com/tikv/client-go/v2/error"
 	"github.com/tikv/client-go/v2/internal/apicodec"
+	"github.com/tikv/client-go/v2/internal/client"
 	"github.com/tikv/client-go/v2/internal/mockstore/mocktikv"
 	"github.com/tikv/client-go/v2/internal/retry"
 	"github.com/tikv/client-go/v2/kv"
 	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikvrpc"
+	"go.uber.org/zap"
 )
 
 func TestRegionRequestToThreeStores(t *testing.T) {
@@ -278,6 +282,12 @@ func refreshEpochs(regionStore *regionStore) {
 	}
 }
 
+func refreshLivenessStates(regionStore *regionStore) {
+	for _, store := range regionStore.stores {
+		atomic.StoreUint32(&store.livenessState, uint32(reachable))
+	}
+}
+
 func AssertRPCCtxEqual(s *testRegionRequestToThreeStoresSuite, rpcCtx *RPCContext, target *replica, proxy *replica) {
 	s.Equal(rpcCtx.Store, target.store)
 	s.Equal(rpcCtx.Peer, target.peer)
@@ -322,7 +332,7 @@ func (s *testRegionRequestToThreeStoresSuite) TestLearnerReplicaSelector() {
 	cache := NewRegionCache(s.cache.pdClient)
 	defer cache.Close()
 	cache.mu.Lock()
-	cache.insertRegionToCache(region)
+	cache.insertRegionToCache(region, true)
 	cache.mu.Unlock()
 
 	// Test accessFollower state with kv.ReplicaReadLearner request type.
@@ -373,7 +383,7 @@ func (s *testRegionRequestToThreeStoresSuite) TestReplicaSelector() {
 	cache := NewRegionCache(s.cache.pdClient)
 	defer cache.Close()
 	cache.mu.Lock()
-	cache.insertRegionToCache(region)
+	cache.insertRegionToCache(region, true)
 	cache.mu.Unlock()
 
 	// Verify creating the replicaSelector.
@@ -567,6 +577,7 @@ func (s *testRegionRequestToThreeStoresSuite) TestReplicaSelector() {
 	// Test accessFollower state with kv.ReplicaReadFollower request type.
 	req = tikvrpc.NewReplicaReadRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{}, kv.ReplicaReadFollower, nil)
 	refreshEpochs(regionStore)
+	refreshLivenessStates(regionStore)
 	replicaSelector, err = newReplicaSelector(cache, regionLoc.Region, req)
 	s.Nil(err)
 	s.NotNil(replicaSelector)
@@ -680,10 +691,13 @@ func (s *testRegionRequestToThreeStoresSuite) TestSendReqWithReplicaSelector() {
 	region, err := s.cache.LocateRegionByID(s.bo, s.regionID)
 	s.Nil(err)
 	s.NotNil(region)
+	regionStore := s.cache.GetCachedRegionWithRLock(region.Region).getStore()
+	s.NotNil(regionStore)
 
 	reloadRegion := func() {
 		s.regionRequestSender.replicaSelector.region.invalidate(Other)
 		region, _ = s.cache.LocateRegionByID(s.bo, s.regionID)
+		regionStore = s.cache.GetCachedRegionWithRLock(region.Region).getStore()
 	}
 
 	hasFakeRegionError := func(resp *tikvrpc.Response) bool {
@@ -700,7 +714,7 @@ func (s *testRegionRequestToThreeStoresSuite) TestSendReqWithReplicaSelector() {
 	// Normal
 	bo := retry.NewBackoffer(context.Background(), -1)
 	sender := s.regionRequestSender
-	resp, _, err := sender.SendReq(bo, req, region.Region, time.Second)
+	resp, _, err := sender.SendReq(bo, req, region.Region, client.ReadTimeoutShort)
 	s.Nil(err)
 	s.NotNil(resp)
 	s.True(bo.GetTotalBackoffTimes() == 0)
@@ -709,16 +723,18 @@ func (s *testRegionRequestToThreeStoresSuite) TestSendReqWithReplicaSelector() {
 	bo = retry.NewBackoffer(context.Background(), -1)
 	s.cluster.ChangeLeader(s.regionID, s.peerIDs[1])
 	s.cluster.StopStore(s.storeIDs[0])
-	resp, _, err = sender.SendReq(bo, req, region.Region, time.Second)
+	resp, _, err = sender.SendReq(bo, req, region.Region, client.ReadTimeoutShort)
 	s.Nil(err)
 	s.NotNil(resp)
 	s.Equal(sender.replicaSelector.targetIdx, AccessIndex(1))
 	s.True(bo.GetTotalBackoffTimes() == 1)
 	s.cluster.StartStore(s.storeIDs[0])
+	atomic.StoreUint32(&regionStore.stores[0].livenessState, uint32(reachable))
 
 	// Leader is updated because of send success, so no backoff.
+	reloadRegion()
 	bo = retry.NewBackoffer(context.Background(), -1)
-	resp, _, err = sender.SendReq(bo, req, region.Region, time.Second)
+	resp, _, err = sender.SendReq(bo, req, region.Region, client.ReadTimeoutShort)
 	s.Nil(err)
 	s.NotNil(resp)
 	s.Equal(sender.replicaSelector.targetIdx, AccessIndex(1))
@@ -734,6 +750,7 @@ func (s *testRegionRequestToThreeStoresSuite) TestSendReqWithReplicaSelector() {
 	s.True(hasFakeRegionError(resp))
 	s.Equal(bo.GetTotalBackoffTimes(), 1)
 	s.cluster.StartStore(s.storeIDs[1])
+	atomic.StoreUint32(&regionStore.stores[1].livenessState, uint32(reachable))
 
 	// Leader is changed. No backoff.
 	reloadRegion()
@@ -750,7 +767,7 @@ func (s *testRegionRequestToThreeStoresSuite) TestSendReqWithReplicaSelector() {
 	resp, _, err = sender.SendReq(bo, req, region.Region, time.Second)
 	s.Nil(err)
 	s.True(hasFakeRegionError(resp))
-	s.Equal(bo.GetTotalBackoffTimes(), 2) // The unreachable leader is skipped
+	s.Equal(bo.GetTotalBackoffTimes(), 3)
 	s.False(sender.replicaSelector.region.isValid())
 	s.cluster.ChangeLeader(s.regionID, s.peerIDs[0])
 
@@ -1052,3 +1069,218 @@ func (s *testRegionRequestToThreeStoresSuite) TestReplicaReadWithFlashbackInProg
 		s.Equal(resp.Resp.(*kvrpcpb.GetResponse).Value, []byte("value"))
 	}
 }
+
+func (s *testRegionRequestToThreeStoresSuite) TestAccessFollowerAfter1TiKVDown() {
+	var leaderAddr string
+	s.regionRequestSender.client = &fnClient{fn: func(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (response *tikvrpc.Response, err error) {
+		// Returns error when accesses non-leader.
+		if leaderAddr != addr {
+			return nil, context.DeadlineExceeded
+		}
+		return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{
+			Value: []byte("value"),
+		}}, nil
+	}}
+
+	req := tikvrpc.NewRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{
+		Key: []byte("key"),
+	})
+	req.ReplicaReadType = kv.ReplicaReadMixed
+
+	loc, err := s.cache.LocateKey(s.bo, []byte("key"))
+	s.Nil(err)
+	region := s.cache.GetCachedRegionWithRLock(loc.Region)
+	s.NotNil(region)
+	regionStore := region.getStore()
+	leaderAddr = regionStore.stores[regionStore.workTiKVIdx].addr
+	s.NotEqual(leaderAddr, "")
+	for i := 0; i < 10; i++ {
+		bo := retry.NewBackofferWithVars(context.Background(), 100, nil)
+		resp, _, _, err := s.regionRequestSender.SendReqCtx(bo, req, loc.Region, client.ReadTimeoutShort, tikvrpc.TiKV)
+		s.Nil(err)
+		s.NotNil(resp)
+
+		// Since send req to follower will receive error, then all follower will be marked as unreachable and epoch stale.
+		allFollowerStoreEpochStale := true
+		for i, store := range regionStore.stores {
+			if i == int(regionStore.workTiKVIdx) {
+				continue
+			}
+			if store.epoch == regionStore.storeEpochs[i] {
+				allFollowerStoreEpochStale = false
+				break
+			} else {
+				s.Equal(store.getLivenessState(), unreachable)
+			}
+		}
+		if allFollowerStoreEpochStale {
+			break
+		}
+	}
+
+	// mock for GC leader reload all regions.
+	bo := retry.NewBackofferWithVars(context.Background(), 10, nil)
+	_, err = s.cache.BatchLoadRegionsWithKeyRange(bo, []byte(""), nil, 1)
+	s.Nil(err)
+
+	loc, err = s.cache.LocateKey(s.bo, []byte("key"))
+	s.Nil(err)
+	region = s.cache.GetCachedRegionWithRLock(loc.Region)
+	s.NotNil(region)
+	regionStore = region.getStore()
+	for i, store := range regionStore.stores {
+		if i == int(regionStore.workTiKVIdx) {
+			continue
+		}
+		// After reload region, the region epoch will be updated, but the store liveness state is still unreachable.
+		s.Equal(store.epoch, regionStore.storeEpochs[i])
+		s.Equal(store.getLivenessState(), unreachable)
+	}
+
+	for i := 0; i < 100; i++ {
+		bo := retry.NewBackofferWithVars(context.Background(), 1, nil)
+		resp, _, retryTimes, err := s.regionRequestSender.SendReqCtx(bo, req, loc.Region, client.ReadTimeoutShort, tikvrpc.TiKV)
+		s.Nil(err)
+		s.NotNil(resp)
+		// since all follower'store is unreachable, the request will be sent to leader, the backoff times should be 0.
+		s.Equal(0, bo.GetTotalBackoffTimes())
+		s.Equal(0, retryTimes)
+	}
+}
+
+func (s *testRegionRequestToThreeStoresSuite) TestStaleReadFallback() {
+	leaderStore, _ := s.loadAndGetLeaderStore()
+	leaderLabel := []*metapb.StoreLabel{
+		{
+			Key:   "id",
+			Value: strconv.FormatUint(leaderStore.StoreID(), 10),
+		},
+	}
+	regionLoc, err := s.cache.LocateRegionByID(s.bo, s.regionID)
+	s.Nil(err)
+	s.NotNil(regionLoc)
+	value := []byte("value")
+	isFirstReq := true
+
+	s.regionRequestSender.client = &fnClient{fn: func(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (response *tikvrpc.Response, err error) {
+		select {
+		case <-ctx.Done():
+			return nil, errors.New("timeout")
+		default:
+		}
+		// Return `DataIsNotReady` for the first time on leader.
+		if isFirstReq {
+			isFirstReq = false
+			return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{RegionError: &errorpb.Error{
+				DataIsNotReady: &errorpb.DataIsNotReady{},
+			}}}, nil
+		}
+		return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{Value: value}}, nil
+	}}
+
+	region := s.cache.getRegionByIDFromCache(regionLoc.Region.GetID())
+	s.True(region.isValid())
+
+	req := tikvrpc.NewReplicaReadRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{Key: []byte("key")}, kv.ReplicaReadLeader, nil)
+	req.ReadReplicaScope = oracle.GlobalTxnScope
+	req.TxnScope = oracle.GlobalTxnScope
+	req.EnableStaleRead()
+	req.ReplicaReadType = kv.ReplicaReadMixed
+	var ops []StoreSelectorOption
+	ops = append(ops, WithMatchLabels(leaderLabel))
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	bo := retry.NewBackoffer(ctx, -1)
+	s.Nil(err)
+	resp, _, _, err := s.regionRequestSender.SendReqCtx(bo, req, regionLoc.Region, time.Second, tikvrpc.TiKV, ops...)
+	s.Nil(err)
+
+	regionErr, err := resp.GetRegionError()
+	s.Nil(err)
+	s.Nil(regionErr)
+	getResp, ok := resp.Resp.(*kvrpcpb.GetResponse)
+	s.True(ok)
+	s.Equal(getResp.Value, value)
+}
+
+func (s *testRegionRequestToThreeStoresSuite) TestSendReqFirstTimeout() {
+	leaderAddr := ""
+	reqTargetAddrs := make(map[string]struct{})
+	s.regionRequestSender.RegionRequestRuntimeStats = NewRegionRequestRuntimeStats()
+	bo := retry.NewBackoffer(context.Background(), 10000)
+	mockRPCClient := &fnClient{fn: func(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (*tikvrpc.Response, error) {
+		reqTargetAddrs[addr] = struct{}{}
+		if req.Context.MaxExecutionDurationMs < 10 {
+			return nil, context.DeadlineExceeded
+		}
+		if addr != leaderAddr && !req.Context.ReplicaRead && !req.Context.StaleRead {
+			return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{RegionError: &errorpb.Error{NotLeader: &errorpb.NotLeader{}}}}, nil
+		}
+		return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{Value: []byte("value")}}, nil
+	}}
+	getLocFn := func() *KeyLocation {
+		loc, err := s.regionRequestSender.regionCache.LocateKey(bo, []byte("a"))
+		s.Nil(err)
+		region := s.regionRequestSender.regionCache.GetCachedRegionWithRLock(loc.Region)
+		leaderStore, _, _, _ := region.WorkStorePeer(region.getStore())
+		leaderAddr, err = s.regionRequestSender.regionCache.getStoreAddr(s.bo, region, leaderStore)
+		s.Nil(err)
+		return loc
+	}
+	resetStats := func() {
+		reqTargetAddrs = make(map[string]struct{})
+		s.regionRequestSender = NewRegionRequestSender(s.cache, mockRPCClient)
+		s.regionRequestSender.RegionRequestRuntimeStats = NewRegionRequestRuntimeStats()
+	}
+
+	//Test different read type.
+	staleReadTypes := []bool{false, true}
+	replicaReadTypes := []kv.ReplicaReadType{kv.ReplicaReadLeader, kv.ReplicaReadFollower, kv.ReplicaReadMixed}
+	for _, staleRead := range staleReadTypes {
+		for _, tp := range replicaReadTypes {
+			log.Info("TestSendReqFirstTimeout", zap.Bool("stale-read", staleRead), zap.String("replica-read-type", tp.String()))
+			resetStats()
+			req := tikvrpc.NewRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{Key: []byte("a")}, kvrpcpb.Context{})
+			if staleRead {
+				req.EnableStaleRead()
+			} else {
+				req.ReplicaRead = tp.IsFollowerRead()
+				req.ReplicaReadType = tp
+			}
+			loc := getLocFn()
+			resp, _, _, err := s.regionRequestSender.SendReqCtx(bo, req, loc.Region, time.Millisecond, tikvrpc.TiKV)
+			s.Nil(err)
+			regionErr, err := resp.GetRegionError()
+			s.Nil(err)
+			s.True(IsFakeRegionError(regionErr))
+			s.Equal(1, len(s.regionRequestSender.Stats))
+			if staleRead {
+				rpcNum := s.regionRequestSender.Stats[tikvrpc.CmdGet].Count
+				s.True(rpcNum == 1 || rpcNum == 2) // 1 rpc or 2 rpc
+			} else {
+				s.Equal(int64(3), s.regionRequestSender.Stats[tikvrpc.CmdGet].Count) // 3 rpc
+				s.Equal(3, len(reqTargetAddrs))                                      // each rpc to a different store.
+			}
+			s.Equal(0, bo.GetTotalBackoffTimes()) // no backoff since fast retry.
+			// warn: must rest MaxExecutionDurationMs before retry.
+			resetStats()
+			if staleRead {
+				req.EnableStaleRead()
+			} else {
+				req.ReplicaRead = tp.IsFollowerRead()
+				req.ReplicaReadType = tp
+			}
+			req.Context.MaxExecutionDurationMs = 0
+			resp, _, _, err = s.regionRequestSender.SendReqCtx(bo, req, loc.Region, time.Second, tikvrpc.TiKV)
+			s.Nil(err)
+			regionErr, err = resp.GetRegionError()
+			s.Nil(err)
+			s.Nil(regionErr)
+			s.Equal([]byte("value"), resp.Resp.(*kvrpcpb.GetResponse).Value)
+			s.Equal(1, len(s.regionRequestSender.Stats))
+			s.Equal(int64(1), s.regionRequestSender.Stats[tikvrpc.CmdGet].Count) // 1 rpc
+			s.Equal(0, bo.GetTotalBackoffTimes())                                // no backoff since fast retry.
+		}
+	}
+}
diff --git a/internal/locate/region_request_test.go b/internal/locate/region_request_test.go
index 6f637bd..069159b 100644
--- a/internal/locate/region_request_test.go
+++ b/internal/locate/region_request_test.go
@@ -407,6 +407,10 @@ func (s *mockTikvGrpcServer) IsAlive(context.Context, *mpp.IsAliveRequest) (*mpp
 	return nil, errors.New("unreachable")
 }
 
+func (s *mockTikvGrpcServer) ReportMPPTaskStatus(context.Context, *mpp.ReportTaskStatusRequest) (*mpp.ReportTaskStatusResponse, error) {
+	return nil, errors.New("unreachable")
+}
+
 func (s *mockTikvGrpcServer) EstablishMPPConnection(*mpp.EstablishMPPConnectionRequest, tikvpb.Tikv_EstablishMPPConnectionServer) error {
 	return errors.New("unreachable")
 }
@@ -414,10 +418,6 @@ func (s *mockTikvGrpcServer) CancelMPPTask(context.Context, *mpp.CancelTaskReque
 	return nil, errors.New("unreachable")
 }
 
-func (s *mockTikvGrpcServer) ReportMPPTaskStatus(ctx context.Context, request *mpp.ReportTaskStatusRequest) (*mpp.ReportTaskStatusResponse, error) {
-	return nil, errors.New("unreachable")
-}
-
 func (s *mockTikvGrpcServer) Raft(tikvpb.Tikv_RaftServer) error {
 	return errors.New("unreachable")
 }
@@ -613,7 +613,7 @@ func (s *testRegionRequestToSingleStoreSuite) TestGetRegionByIDFromCache() {
 	v2 := region.Region.confVer + 1
 	r2 := metapb.Region{Id: region.Region.id, RegionEpoch: &metapb.RegionEpoch{Version: region.Region.ver, ConfVer: v2}, StartKey: []byte{1}}
 	st := &Store{storeID: s.store}
-	s.cache.insertRegionToCache(&Region{meta: &r2, store: unsafe.Pointer(st), lastAccess: time.Now().Unix()})
+	s.cache.insertRegionToCache(&Region{meta: &r2, store: unsafe.Pointer(st), lastAccess: time.Now().Unix()}, true)
 	region, err = s.cache.LocateRegionByID(s.bo, s.region)
 	s.Nil(err)
 	s.NotNil(region)
@@ -623,7 +623,7 @@ func (s *testRegionRequestToSingleStoreSuite) TestGetRegionByIDFromCache() {
 	v3 := region.Region.confVer + 1
 	r3 := metapb.Region{Id: region.Region.id, RegionEpoch: &metapb.RegionEpoch{Version: v3, ConfVer: region.Region.confVer}, StartKey: []byte{2}}
 	st = &Store{storeID: s.store}
-	s.cache.insertRegionToCache(&Region{meta: &r3, store: unsafe.Pointer(st), lastAccess: time.Now().Unix()})
+	s.cache.insertRegionToCache(&Region{meta: &r3, store: unsafe.Pointer(st), lastAccess: time.Now().Unix()}, true)
 	region, err = s.cache.LocateRegionByID(s.bo, s.region)
 	s.Nil(err)
 	s.NotNil(region)
@@ -663,42 +663,38 @@ func (s *testRegionRequestToSingleStoreSuite) TestCloseConnectionOnStoreNotMatch
 	s.Equal(target, client.closedAddr)
 }
 
-func (s *testRegionRequestToSingleStoreSuite) TestCountReplicaNumber() {
-	fmt.Println("TestCountReplicaNumber")
-	fmt.Println(s.cache.storeMu.stores)
-	tikvLabels := []*metapb.StoreLabel{{Key: "engine", Value: "tikv"}}
-	tiflashLabels := []*metapb.StoreLabel{{Key: "engine", Value: "tiflash"}}
-	tiflashWNLabels := []*metapb.StoreLabel{{Key: "engine", Value: "tiflash"}, {Key: "engine_role", Value: "write"}}
-
-	s.cache.SetRegionCacheStore(1, "", "", tikvrpc.TiKV, 0, tikvLabels)
-	s.cache.SetRegionCacheStore(2, "", "", tikvrpc.TiKV, 0, tikvLabels)
-	s.cache.SetRegionCacheStore(3, "", "", tikvrpc.TiKV, 0, tikvLabels)
-	s.cache.SetRegionCacheStore(4, "", "", tikvrpc.TiFlash, 0, tiflashLabels)
-	s.cache.SetRegionCacheStore(5, "", "", tikvrpc.TiFlash, 0, tiflashLabels)
-	{
-		peers := []*metapb.Peer{{StoreId: 1, Role: metapb.PeerRole_Voter}}
-		s.Equal(1, s.regionRequestSender.countReplicaNumber(peers))
-		peers = append(peers, &metapb.Peer{StoreId: 2, Role: metapb.PeerRole_Voter})
-		s.Equal(2, s.regionRequestSender.countReplicaNumber(peers))
-		peers = append(peers, &metapb.Peer{StoreId: 3, Role: metapb.PeerRole_Voter})
-		s.Equal(3, s.regionRequestSender.countReplicaNumber(peers))
-		peers = append(peers, &metapb.Peer{StoreId: 4, Role: metapb.PeerRole_Learner})
-		s.Equal(4, s.regionRequestSender.countReplicaNumber(peers))
-		peers = append(peers, &metapb.Peer{StoreId: 5, Role: metapb.PeerRole_Learner})
-		s.Equal(5, s.regionRequestSender.countReplicaNumber(peers))
-	}
-	s.cache.SetRegionCacheStore(4, "", "", tikvrpc.TiFlash, 0, tiflashWNLabels)
-	s.cache.SetRegionCacheStore(5, "", "", tikvrpc.TiFlash, 0, tiflashWNLabels)
-	{
-		peers := []*metapb.Peer{{StoreId: 1, Role: metapb.PeerRole_Voter}}
-		s.Equal(1, s.regionRequestSender.countReplicaNumber(peers))
-		peers = append(peers, &metapb.Peer{StoreId: 2, Role: metapb.PeerRole_Voter})
-		s.Equal(2, s.regionRequestSender.countReplicaNumber(peers))
-		peers = append(peers, &metapb.Peer{StoreId: 3, Role: metapb.PeerRole_Voter})
-		s.Equal(3, s.regionRequestSender.countReplicaNumber(peers))
-		peers = append(peers, &metapb.Peer{StoreId: 4, Role: metapb.PeerRole_Learner})
-		s.Equal(4, s.regionRequestSender.countReplicaNumber(peers))
-		peers = append(peers, &metapb.Peer{StoreId: 5, Role: metapb.PeerRole_Learner})
-		s.Equal(4, s.regionRequestSender.countReplicaNumber(peers)) // Only count 1 tiflash replica for tiflash write-nodes.
-	}
+func (s *testRegionRequestToSingleStoreSuite) TestStaleReadRetry() {
+	req := tikvrpc.NewRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{
+		Key: []byte("key"),
+	})
+	req.EnableStaleRead()
+	req.ReadReplicaScope = "z1" // not global stale read.
+	region, err := s.cache.LocateRegionByID(s.bo, s.region)
+	s.Nil(err)
+	s.NotNil(region)
+
+	oc := s.regionRequestSender.client
+	defer func() {
+		s.regionRequestSender.client = oc
+	}()
+
+	s.regionRequestSender.client = &fnClient{fn: func(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (response *tikvrpc.Response, err error) {
+		if req.StaleRead {
+			// Mock for stale-read request always return DataIsNotReady error when tikv `ResolvedTS` is blocked.
+			response = &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{
+				RegionError: &errorpb.Error{DataIsNotReady: &errorpb.DataIsNotReady{}},
+			}}
+		} else {
+			response = &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{Value: []byte("value")}}
+		}
+		return response, nil
+	}}
+
+	bo := retry.NewBackofferWithVars(context.Background(), 5, nil)
+	resp, _, err := s.regionRequestSender.SendReq(bo, req, region.Region, time.Second)
+	s.Nil(err)
+	s.NotNil(resp)
+	regionErr, _ := resp.GetRegionError()
+	s.Nil(regionErr)
+	s.Equal([]byte("value"), resp.Resp.(*kvrpcpb.GetResponse).Value)
 }
diff --git a/internal/unionstore/memdb_test.go b/internal/unionstore/memdb_test.go
index 8ce0a66..e5fe206 100644
--- a/internal/unionstore/memdb_test.go
+++ b/internal/unionstore/memdb_test.go
@@ -98,14 +98,34 @@ func TestIterator(t *testing.T) {
 	}
 	assert.Equal(i, cnt)
 
-	i--
-	for it, _ := db.IterReverse(nil); it.Valid(); it.Next() {
+	for it, _ := db.IterReverse(nil, nil); it.Valid(); it.Next() {
+		binary.BigEndian.PutUint32(buf[:], uint32(i-1))
+		assert.Equal(it.Key(), buf[:])
+		assert.Equal(it.Value(), buf[:])
+		i--
+	}
+	assert.Equal(i, 0)
+
+	var upperBoundBytes, lowerBoundBytes [4]byte
+	bound := 400
+	binary.BigEndian.PutUint32(upperBoundBytes[:], uint32(bound))
+	for it, _ := db.Iter(nil, upperBoundBytes[:]); it.Valid(); it.Next() {
 		binary.BigEndian.PutUint32(buf[:], uint32(i))
 		assert.Equal(it.Key(), buf[:])
 		assert.Equal(it.Value(), buf[:])
+		i++
+	}
+	assert.Equal(i, bound)
+
+	i = cnt
+	binary.BigEndian.PutUint32(lowerBoundBytes[:], uint32(bound))
+	for it, _ := db.IterReverse(nil, lowerBoundBytes[:]); it.Valid(); it.Next() {
+		binary.BigEndian.PutUint32(buf[:], uint32(i-1))
+		assert.Equal(it.Key(), buf[:])
+		assert.Equal(it.Value(), buf[:])
 		i--
 	}
-	assert.Equal(i, -1)
+	assert.Equal(i, bound)
 }
 
 func TestDiscard(t *testing.T) {
@@ -139,7 +159,7 @@ func TestDiscard(t *testing.T) {
 	assert.Equal(i, cnt)
 
 	i--
-	for it, _ := db.IterReverse(nil); it.Valid(); it.Next() {
+	for it, _ := db.IterReverse(nil, nil); it.Valid(); it.Next() {
 		binary.BigEndian.PutUint32(buf[:], uint32(i))
 		assert.Equal(it.Key(), buf[:])
 		assert.Equal(it.Value(), buf[:])
@@ -197,7 +217,7 @@ func TestFlushOverwrite(t *testing.T) {
 	assert.Equal(i, cnt)
 
 	i--
-	for it, _ := db.IterReverse(nil); it.Valid(); it.Next() {
+	for it, _ := db.IterReverse(nil, nil); it.Valid(); it.Next() {
 		binary.BigEndian.PutUint32(kbuf[:], uint32(i))
 		binary.BigEndian.PutUint32(vbuf[:], uint32(i+1))
 		assert.Equal(it.Key(), kbuf[:])
@@ -279,7 +299,7 @@ func TestNestedSandbox(t *testing.T) {
 	assert.Equal(i, 200)
 
 	i--
-	for it, _ := db.IterReverse(nil); it.Valid(); it.Next() {
+	for it, _ := db.IterReverse(nil, nil); it.Valid(); it.Next() {
 		binary.BigEndian.PutUint32(kbuf[:], uint32(i))
 		binary.BigEndian.PutUint32(vbuf[:], uint32(i))
 		if i < 100 {
@@ -336,7 +356,7 @@ func TestOverwrite(t *testing.T) {
 	assert.Equal(i, cnt)
 
 	i--
-	for it, _ := db.IterReverse(nil); it.Valid(); it.Next() {
+	for it, _ := db.IterReverse(nil, nil); it.Valid(); it.Next() {
 		binary.BigEndian.PutUint32(buf[:], uint32(i))
 		assert.Equal(it.Key(), buf[:])
 		v := binary.BigEndian.Uint32(it.Value())
@@ -569,7 +589,7 @@ func checkConsist(t *testing.T, p1 *MemDB, p2 *leveldb.DB) {
 		assert.Equal(it.Value(), it2.Value())
 
 		if prevKey != nil {
-			it, _ = p1.IterReverse(it2.Key())
+			it, _ = p1.IterReverse(it2.Key(), nil)
 			assert.Equal(it.Key(), prevKey)
 			assert.Equal(it.Value(), prevVal)
 		}
@@ -579,7 +599,7 @@ func checkConsist(t *testing.T, p1 *MemDB, p2 *leveldb.DB) {
 		prevVal = it2.Value()
 	}
 
-	it1, _ = p1.IterReverse(nil)
+	it1, _ = p1.IterReverse(nil, nil)
 	for it2.Last(); it2.Valid(); it2.Prev() {
 		assert.Equal(it1.Key(), it2.Key())
 		assert.Equal(it1.Value(), it2.Value())
diff --git a/internal/unionstore/union_store_test.go b/internal/unionstore/union_store_test.go
index 8356bf3..7d4aeca 100644
--- a/internal/unionstore/union_store_test.go
+++ b/internal/unionstore/union_store_test.go
@@ -125,25 +125,33 @@ func TestUnionStoreIterReverse(t *testing.T) {
 	err = store.Set([]byte("3"), []byte("3"))
 	assert.Nil(err)
 
-	iter, err := us.IterReverse(nil)
+	iter, err := us.IterReverse(nil, nil)
 	assert.Nil(err)
 	checkIterator(t, iter, [][]byte{[]byte("3"), []byte("2"), []byte("1")}, [][]byte{[]byte("3"), []byte("2"), []byte("1")})
 
-	iter, err = us.IterReverse([]byte("3"))
+	iter, err = us.IterReverse([]byte("3"), []byte("1"))
+	assert.Nil(err)
+	checkIterator(t, iter, [][]byte{[]byte("2"), []byte("1")}, [][]byte{[]byte("2"), []byte("1")})
+
+	iter, err = us.IterReverse([]byte("3"), nil)
 	assert.Nil(err)
 	checkIterator(t, iter, [][]byte{[]byte("2"), []byte("1")}, [][]byte{[]byte("2"), []byte("1")})
 
 	err = us.GetMemBuffer().Set([]byte("0"), []byte("0"))
 	assert.Nil(err)
-	iter, err = us.IterReverse([]byte("3"))
+	iter, err = us.IterReverse([]byte("3"), nil)
 	assert.Nil(err)
 	checkIterator(t, iter, [][]byte{[]byte("2"), []byte("1"), []byte("0")}, [][]byte{[]byte("2"), []byte("1"), []byte("0")})
 
 	err = us.GetMemBuffer().Delete([]byte("1"))
 	assert.Nil(err)
-	iter, err = us.IterReverse([]byte("3"))
+	iter, err = us.IterReverse([]byte("3"), nil)
 	assert.Nil(err)
 	checkIterator(t, iter, [][]byte{[]byte("2"), []byte("0")}, [][]byte{[]byte("2"), []byte("0")})
+
+	iter, err = us.IterReverse([]byte("3"), []byte("1"))
+	assert.Nil(err)
+	checkIterator(t, iter, [][]byte{[]byte("2")}, [][]byte{[]byte("2")})
 }
 
 func checkIterator(t *testing.T, iter Iterator, keys [][]byte, values [][]byte) {
diff --git a/tikv/test_probe.go b/tikv/test_probe.go
index d470d69..5971480 100644
--- a/tikv/test_probe.go
+++ b/tikv/test_probe.go
@@ -35,6 +35,7 @@
 package tikv
 
 import (
+	"bytes"
 	"context"
 
 	"github.com/pingcap/kvproto/pkg/metapb"
@@ -114,6 +115,42 @@ func (s StoreProbe) GCResolveLockPhase(ctx context.Context, safepoint uint64, co
 	return s.resolveLocks(ctx, safepoint, concurrency)
 }
 
+func (s StoreProbe) ScanLocks(ctx context.Context, startKey, endKey []byte, maxVersion uint64) ([]*txnlock.Lock, error) {
+	bo := NewGcResolveLockMaxBackoffer(ctx)
+	const limit = 1024
+
+	var result []*txnlock.Lock
+
+outerLoop:
+	for {
+		locks, loc, err := s.KVStore.scanLocksInRegionWithStartKey(bo, startKey, maxVersion, limit)
+		if err != nil {
+			return nil, err
+		}
+		for _, l := range locks {
+			if bytes.Compare(endKey, l.Key) <= 0 {
+				// Finished scanning the given range.
+				break outerLoop
+			}
+			result = append(result, l)
+		}
+
+		if len(locks) < limit {
+			if len(loc.EndKey) == 0 {
+				// Scanned to the very end.
+				break outerLoop
+			}
+			// The current region is completely scanned.
+			startKey = loc.EndKey
+		} else {
+			// The current region may still have more locks.
+			startKey = append(locks[len(locks)-1].Key, 0)
+		}
+	}
+
+	return result, nil
+}
+
 // LockResolverProbe wraps a LockResolver and exposes internal stats for testing purpose.
 type LockResolverProbe struct {
 	*txnlock.LockResolverProbe
diff --git a/util/request_source_test.go b/util/request_source_test.go
index 4f991a9..433bc42 100644
--- a/util/request_source_test.go
+++ b/util/request_source_test.go
@@ -43,19 +43,19 @@ func TestGetRequestSource(t *testing.T) {
 
 	// Test nil pointer
 	rs = nil
-	expected = "unknown_default"
+	expected = "unknown"
 	actual = rs.GetRequestSource()
 	assert.Equal(t, expected, actual)
 
 	// Test empty RequestSourceType and ExplicitRequestSourceType
 	rs = &RequestSource{}
-	expected = "unknown_default"
+	expected = "unknown"
 	actual = rs.GetRequestSource()
 	assert.Equal(t, expected, actual)
 
 	// Test empty ExplicitRequestSourceType
 	rs.RequestSourceType = "test"
-	expected = "external_test_default"
+	expected = "external_test"
 	actual = rs.GetRequestSource()
 	assert.Equal(t, expected, actual)
 
@@ -66,3 +66,30 @@ func TestGetRequestSource(t *testing.T) {
 	actual = rs.GetRequestSource()
 	assert.Equal(t, expected, actual)
 }
+
+func TestBuildRequestSource(t *testing.T) {
+	// Test internal request
+	expected := "internal_test_lightning"
+	actual := BuildRequestSource(true, "test", "lightning")
+	assert.Equal(t, expected, actual)
+
+	// Test external request
+	expected = "external_test_lightning"
+	actual = BuildRequestSource(false, "test", "lightning")
+	assert.Equal(t, expected, actual)
+
+	// Test empty ExplicitRequestSourceType
+	expected = "external_test"
+	actual = BuildRequestSource(false, "test", "")
+	assert.Equal(t, expected, actual)
+
+	// Test empty RequestSourceType
+	expected = "external_unknown_lightning"
+	actual = BuildRequestSource(false, "", "lightning")
+	assert.Equal(t, expected, actual)
+
+	// Test RequestSourceType && ExplicitRequestSourceType both empty
+	expected = "unknown"
+	actual = BuildRequestSource(true, "", "")
+	assert.Equal(t, expected, actual)
+}