Repository: tikv/client-go
Pull request #1458 (2024-09-05 11:07:18)

feat: Broadcast min_commit_ts for pipelined transactions
Ref: https://github.com/tikv/tikv/issues/17459
Prerequires https://github.com/pingcap/kvproto/pull/1265

Changes
1. Periodic updates of min_commit_ts
2. Broadcast min_commit_ts
3. Periodic updates of full store list

ti-chi-bot[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
**Once this PR has been reviewed and has the lgtm label**, please ask for approval from [ekexium](https://github.com/ekexium), **ensuring that each of them provides their approval before proceeding**. For more information see [the Code Review Process](https://book.prow.tidb.net/#/workflows/pr).

The full list of commands accepted by this bot can be found [here](https://prow.tidb.net/command-help?repo=tikv%2Fclient-go).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/tikv/client-go/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["ekexium"]} -->
ekexium: > BTW, this is not related to this PR, can you change the `RLock/RUnlock` to `Lock/Unlock` [here](https://github.com/tikv/client-go/blob/ad0302eef1bf688a90020b52884ed64437844b95/txnkv/transaction/pipelined_flush.go#L333-L335)?

Done.

> I have a question about the future plan, have you considered apply this broadcast also for slow or large 2pc transactions?

It's possible. But I think we need more empirical data before implementing that. It's unclear now how this optimization would benefit normal large transactions.

Patch:
diff --git a/go.mod b/go.mod
index f8bbc87..d226e97 100644
--- a/go.mod
+++ b/go.mod
@@ -15,7 +15,7 @@ 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-20240620063548-118a4cab53e4
+	github.com/pingcap/kvproto v0.0.0-20240924080114-4a3e17f5e62d
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.18.0
diff --git a/go.sum b/go.sum
index 74adbea..029e0d0 100644
--- a/go.sum
+++ b/go.sum
@@ -74,8 +74,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-20240620063548-118a4cab53e4 h1:6aIKNB2YGAec4IUDLw6G2eDECiGiufZcgEbZSCELBx0=
-github.com/pingcap/kvproto v0.0.0-20240620063548-118a4cab53e4/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8=
+github.com/pingcap/kvproto v0.0.0-20240924080114-4a3e17f5e62d h1:vSdKTrF6kpcd56G5BLP0Bz88Nho2tDo7IR1+oSsBAfc=
+github.com/pingcap/kvproto v0.0.0-20240924080114-4a3e17f5e62d/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8=
 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.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
diff --git a/internal/locate/region_cache.go b/internal/locate/region_cache.go
index ef575f8..01dfa38 100644
--- a/internal/locate/region_cache.go
+++ b/internal/locate/region_cache.go
@@ -726,9 +726,47 @@ func NewRegionCache(pdClient pd.Client, opt ...RegionCacheOpt) *RegionCache {
 		// cache GC is incompatible with cache refresh
 		c.bg.schedule(c.gcRoundFunc(cleanRegionNumPerRound), cleanCacheInterval)
 	}
+	c.bg.schedule(
+		func(ctx context.Context, _ time.Time) bool {
+			refreshFullStoreList(ctx, c.stores)
+			return false
+		}, refreshStoreListInterval,
+	)
 	return c
 }
 
+// Try to refresh full store list. Errors are ignored.
+func refreshFullStoreList(ctx context.Context, stores storeCache) {
+	storeList, err := stores.fetchAllStores(ctx)
+	if err != nil {
+		logutil.Logger(ctx).Info("refresh full store list failed", zap.Error(err))
+		return
+	}
+	for _, store := range storeList {
+		_, exist := stores.get(store.GetId())
+		if exist {
+			continue
+		}
+		// GetAllStores is supposed to return only Up and Offline stores.
+		// This check is being defensive and to make it consistent with store resolve code.
+		if store == nil || store.GetState() == metapb.StoreState_Tombstone {
+			continue
+		}
+		addr := store.GetAddress()
+		if addr == "" {
+			continue
+		}
+		s := stores.getOrInsertDefault(store.GetId())
+		// TODO: maybe refactor this, together with other places initializing Store
+		s.addr = addr
+		s.peerAddr = store.GetPeerAddress()
+		s.saddr = store.GetStatusAddress()
+		s.storeType = tikvrpc.GetStoreTypeByMeta(store)
+		s.labels = store.GetLabels()
+		s.changeResolveStateTo(unresolved, resolved)
+	}
+}
+
 // only used fot test.
 func newTestRegionCache() *RegionCache {
 	c := &RegionCache{}
@@ -2649,6 +2687,7 @@ func (c *RegionCache) UpdateBucketsIfNeeded(regionID RegionVerID, latestBucketsV
 
 const cleanCacheInterval = time.Second
 const cleanRegionNumPerRound = 50
+const refreshStoreListInterval = 10 * time.Second
 
 // gcScanItemHook is only used for testing
 var gcScanItemHook = new(atomic.Pointer[func(*btreeItem)])
diff --git a/internal/unionstore/art/art.go b/internal/unionstore/art/art.go
index 29f0ea5..d54a2dc 100644
--- a/internal/unionstore/art/art.go
+++ b/internal/unionstore/art/art.go
@@ -528,16 +528,24 @@ func (t *ART) MemHookSet() bool {
 
 // GetKeyByHandle returns key by handle.
 func (t *ART) GetKeyByHandle(handle arena.MemKeyHandle) []byte {
-	panic("unimplemented")
+	lf := t.allocator.getLeaf(handle.ToAddr())
+	return lf.GetKey()
 }
 
 // GetValueByHandle returns value by handle.
 func (t *ART) GetValueByHandle(handle arena.MemKeyHandle) ([]byte, bool) {
-	panic("unimplemented")
+	if t.vlogInvalid {
+		return nil, false
+	}
+	lf := t.allocator.getLeaf(handle.ToAddr())
+	if lf.vAddr.IsNull() {
+		return nil, false
+	}
+	return t.allocator.vlogAllocator.GetValue(lf.vAddr), true
 }
 
 func (t *ART) SetEntrySizeLimit(entryLimit, bufferLimit uint64) {
-	panic("unimplemented")
+	t.entrySizeLimit, t.bufferSizeLimit = entryLimit, bufferLimit
 }
 
 func (t *ART) RemoveFromBuffer(key []byte) {
diff --git a/internal/unionstore/art/art_iterator.go b/internal/unionstore/art/art_iterator.go
index 5e032c4..3c84475 100644
--- a/internal/unionstore/art/art_iterator.go
+++ b/internal/unionstore/art/art_iterator.go
@@ -14,18 +14,489 @@
 
 package art
 
-func (*ART) Iter([]byte, []byte) (*Iterator, error) {
-	panic("unimplemented")
+import (
+	"bytes"
+	"sort"
+
+	"github.com/pkg/errors"
+	"github.com/tikv/client-go/v2/internal/unionstore/arena"
+	"github.com/tikv/client-go/v2/kv"
+)
+
+func (t *ART) Iter(lowerBound, upperBound []byte) (*Iterator, error) {
+	return t.iter(lowerBound, upperBound, false, false)
+}
+
+func (t *ART) IterReverse(upperBound, lowerBound []byte) (*Iterator, error) {
+	return t.iter(lowerBound, upperBound, true, false)
+}
+
+func (t *ART) IterWithFlags(lowerBound, upperBound []byte) *Iterator {
+	it, _ := t.iter(lowerBound, upperBound, false, true)
+	return it
+}
+
+func (t *ART) IterReverseWithFlags(upperBound []byte) *Iterator {
+	it, _ := t.iter(nil, upperBound, true, true)
+	return it
+}
+
+func (t *ART) iter(lowerBound, upperBound []byte, reverse, includeFlags bool) (*Iterator, error) {
+	it := &Iterator{
+		tree:         t,
+		reverse:      reverse,
+		valid:        true,
+		includeFlags: includeFlags,
+		inner: &baseIter{
+			allocator: &t.allocator,
+		},
+		currAddr: arena.BadAddr, // the default value of currAddr is not equal to any valid address
+		endAddr:  arena.NullAddr,
+	}
+	it.init(lowerBound, upperBound)
+	if !it.valid {
+		return it, nil
+	}
+	if err := it.Next(); err != nil {
+		return nil, err
+	}
+	return it, nil
+}
+
+type Iterator struct {
+	tree         *ART
+	reverse      bool
+	valid        bool
+	includeFlags bool
+	inner        *baseIter
+	currLeaf     *artLeaf
+	currAddr     arena.MemdbArenaAddr
+	endAddr      arena.MemdbArenaAddr
+}
+
+func (it *Iterator) Valid() bool        { return it.valid }
+func (it *Iterator) Key() []byte        { return it.currLeaf.GetKey() }
+func (it *Iterator) Flags() kv.KeyFlags { return it.currLeaf.GetKeyFlags() }
+func (it *Iterator) Value() []byte {
+	if it.currLeaf.vAddr.IsNull() {
+		return nil
+	}
+	return it.tree.allocator.vlogAllocator.GetValue(it.currLeaf.vAddr)
+}
+
+func (it *Iterator) Next() error {
+	if !it.valid {
+		// iterate is finished
+		return errors.New("Art: iterator is finished")
+	}
+	if it.currAddr == it.endAddr {
+		it.valid = false
+		return nil
+	}
+
+	var nextLeaf artNode
+	for {
+		if it.reverse {
+			nextLeaf = it.inner.prev()
+		} else {
+			nextLeaf = it.inner.next()
+		}
+		if nextLeaf.addr.IsNull() {
+			it.valid = false
+			return nil
+		}
+		it.setCurrLeaf(nextLeaf.addr)
+		if it.currLeaf.vAddr.IsNull() {
+			// if it.includeFlags is true, the iterator should return even the value is null.
+			if it.includeFlags && !it.currLeaf.isDeleted() {
+				return nil
+			}
+			if nextLeaf.addr == it.endAddr {
+				it.valid = false
+				return nil
+			}
+			continue
+		}
+		return nil
+	}
 }
 
-func (*ART) IterReverse([]byte, []byte) (*Iterator, error) {
-	panic("unimplemented")
+func (it *Iterator) setCurrLeaf(node arena.MemdbArenaAddr) {
+	it.currAddr = node
+	it.currLeaf = it.tree.allocator.getLeaf(node)
 }
 
-type Iterator struct{}
+func (it *Iterator) Close() {}
 
-func (i *Iterator) Valid() bool   { panic("unimplemented") }
-func (i *Iterator) Key() []byte   { panic("unimplemented") }
-func (i *Iterator) Value() []byte { panic("unimplemented") }
-func (i *Iterator) Next() error   { panic("unimplemented") }
-func (i *Iterator) Close()        { panic("unimplemented") }
+func (it *Iterator) Handle() arena.MemKeyHandle {
+	return it.currAddr.ToHandle()
+}
+
+func (it *Iterator) init(lowerBound, upperBound []byte) {
+	if it.tree.root.addr.IsNull() {
+		it.valid = false
+		return
+	}
+	if len(lowerBound) > 0 && len(upperBound) > 0 {
+		if bytes.Compare(lowerBound, upperBound) >= 0 {
+			it.valid = false
+			return
+		}
+	}
+
+	if it.reverse {
+		it.inner.idxes, it.inner.nodes = it.seek(upperBound)
+		if len(lowerBound) == 0 {
+			it.endAddr = arena.NullAddr
+		} else {
+			helper := new(baseIter)
+			helper.allocator = &it.tree.allocator
+			helper.idxes, helper.nodes = it.seek(lowerBound)
+			if it.inner.compare(helper) > 0 {
+				// lowerBound is inclusive, call next to find the smallest leaf node that >= lowerBound.
+				it.endAddr = helper.next().addr
+				if it.inner.compare(helper) < 0 || len(helper.idxes) == 0 {
+					it.valid = false
+				}
+				return
+			}
+			it.valid = false
+		}
+		return
+	}
+
+	it.inner.idxes, it.inner.nodes = it.seek(lowerBound)
+	if len(upperBound) == 0 {
+		it.endAddr = arena.NullAddr
+	} else {
+		helper := new(baseIter)
+		helper.allocator = &it.tree.allocator
+		helper.idxes, helper.nodes = it.seek(upperBound)
+		if it.inner.compare(helper) < 0 {
+			// upperBound is exclusive, so we move the helper cursor to the previous node, which is the true endAddr.
+			it.endAddr = helper.prev().addr
+			if it.inner.compare(helper) > 0 || len(helper.idxes) == 0 {
+				it.valid = false
+			}
+			return
+		}
+		it.valid = false
+		return
+	}
+}
+
+// seek the first node and index that >= key, return the indexes and nodes of the lookup path
+// nodes[0] is the root node
+func (it *Iterator) seek(key artKey) ([]int, []artNode) {
+	curr := it.tree.root
+	depth := uint32(0)
+	idxes := make([]int, 0, 8)
+	nodes := make([]artNode, 0, 8)
+	if len(key) == 0 {
+		// if the seek key is empty, it means -inf or +inf, return root node directly.
+		nodes = append(nodes, curr)
+		if it.reverse {
+			idxes = append(idxes, node256cap)
+		} else {
+			idxes = append(idxes, inplaceIndex)
+		}
+		return idxes, nodes
+	}
+	var node *nodeBase
+	for {
+		if curr.isLeaf() {
+			if key.valid(int(depth)) {
+				lf := curr.asLeaf(&it.tree.allocator)
+				if bytes.Compare(key, lf.GetKey()) > 0 {
+					// the seek key is not exist, and it's longer and larger than the current leaf's key.
+					// e.g. key: [1, 1, 1], leaf: [1, 1].
+					idxes[len(idxes)-1]++
+				}
+			}
+			break
+		}
+
+		node = curr.asNode(&it.tree.allocator)
+		if node.prefixLen > 0 {
+			mismatchIdx := node.matchDeep(&it.tree.allocator, &curr, key, depth)
+			if mismatchIdx < node.prefixLen {
+				// no leaf node is match with the seek key
+				leafNode := minimum(&it.tree.allocator, curr)
+				leafKey := leafNode.asLeaf(&it.tree.allocator).GetKey()
+				if mismatchIdx+depth == uint32(len(key)) || key[depth+mismatchIdx] < leafKey[depth+mismatchIdx] {
+					// key < leafKey, set index to -1 means all the children are larger than the seek key
+					idxes = append(idxes, -1)
+				} else {
+					// key > leafKey, set index to 256 means all the children are less than the seek key
+					idxes = append(idxes, node256cap)
+				}
+				nodes = append(nodes, curr)
+				return idxes, nodes
+			}
+			depth += min(mismatchIdx, node.prefixLen)
+		}
+
+		nodes = append(nodes, curr)
+		char := key.charAt(int(depth))
+		idx, next := curr.findChild(&it.tree.allocator, char, !key.valid(int(depth)))
+		if next.addr.IsNull() {
+			nextIdx := 0
+			switch curr.kind {
+			case typeNode4:
+				n4 := curr.asNode4(&it.tree.allocator)
+				for ; nextIdx < int(n4.nodeNum); nextIdx++ {
+					if n4.keys[nextIdx] >= char {
+						break
+					}
+				}
+			case typeNode16:
+				n16 := curr.asNode16(&it.tree.allocator)
+				nextIdx, _ = sort.Find(int(n16.nodeNum), func(i int) int {
+					if n16.keys[i] < char {
+						return 1
+					}
+					return -1
+				})
+			case typeNode48:
+				n48 := curr.asNode48(&it.tree.allocator)
+				nextIdx = n48.nextPresentIdx(int(char))
+			case typeNode256:
+				n256 := curr.asNode256(&it.tree.allocator)
+				nextIdx = n256.nextPresentIdx(int(char))
+			}
+			idxes = append(idxes, nextIdx)
+			return idxes, nodes
+		}
+		idxes = append(idxes, idx)
+		curr = next
+		depth++
+	}
+	return idxes, nodes
+}
+
+type baseIter struct {
+	allocator *artAllocator
+	idxes     []int
+	nodes     []artNode
+}
+
+// compare compares the path of nodes, return 1 if self > other, -1 if self < other, 0 if self == other
+func (it *baseIter) compare(other *baseIter) int {
+	l1, l2 := len(it.idxes), len(other.idxes)
+	for i, l := 0, min(l1, l2); i < l; i++ {
+		if it.idxes[i] == other.idxes[i] {
+			continue
+		}
+		if it.idxes[i] > other.idxes[i] {
+			return 1
+		}
+		if it.idxes[i] < other.idxes[i] {
+			return -1
+		}
+	}
+	if l1 == l2 {
+		return 0
+	} else if l1 < l2 {
+		return -1
+	} else {
+		return 1
+	}
+}
+
+// next returns the next leaf node
+// it returns nullArtNode if there is no more leaf node
+func (it *baseIter) next() artNode {
+	depth := len(it.nodes) - 1
+	curr := it.nodes[depth]
+	idx := it.idxes[depth]
+	switch curr.kind {
+	case typeNode4:
+		n4 := it.allocator.getNode4(curr.addr)
+		if idx == inplaceIndex {
+			idx = 0 // mark in-place leaf is visited
+			it.idxes[depth] = idx
+			if !n4.inplaceLeaf.addr.IsNull() {
+				return n4.inplaceLeaf
+			}
+		} else if idx == node4cap {
+			break
+		}
+		if idx < int(n4.nodeNum) {
+			it.idxes[depth] = idx
+			child := n4.children[idx]
+			if child.kind == typeLeaf {
+				it.idxes[depth]++
+				return child
+			}
+			it.nodes = append(it.nodes, child)
+			it.idxes = append(it.idxes, inplaceIndex)
+			return it.next()
+		}
+	case typeNode16:
+		n16 := it.allocator.getNode16(curr.addr)
+		if idx == inplaceIndex {
+			idx = 0 // mark in-place leaf is visited
+			it.idxes[depth] = idx
+			if !n16.inplaceLeaf.addr.IsNull() {
+				return n16.inplaceLeaf
+			}
+		} else if idx == node16cap {
+			break
+		}
+		if idx < int(n16.nodeNum) {
+			it.idxes[depth] = idx
+			child := n16.children[idx]
+			if child.kind == typeLeaf {
+				it.idxes[depth]++
+				return child
+			}
+			it.nodes = append(it.nodes, child)
+			it.idxes = append(it.idxes, inplaceIndex)
+			return it.next()
+		}
+	case typeNode48:
+		n48 := it.allocator.getNode48(curr.addr)
+		if idx == inplaceIndex {
+			idx = 0 // mark in-place leaf is visited
+			it.idxes[depth] = idx
+			if !n48.inplaceLeaf.addr.IsNull() {
+				return n48.inplaceLeaf
+			}
+		} else if idx == node256cap {
+			break
+		}
+		idx = n48.nextPresentIdx(idx)
+		if idx < node256cap {
+			it.idxes[depth] = idx
+			child := n48.children[n48.keys[idx]]
+			if child.kind == typeLeaf {
+				it.idxes[depth]++
+				return child
+			}
+			it.nodes = append(it.nodes, child)
+			it.idxes = append(it.idxes, inplaceIndex)
+			return it.next()
+		}
+	case typeNode256:
+		n256 := it.allocator.getNode256(curr.addr)
+		if idx == inplaceIndex {
+			idx = 0 // mark in-place leaf is visited
+			it.idxes[depth] = idx
+			if !n256.inplaceLeaf.addr.IsNull() {
+				return n256.inplaceLeaf
+			}
+		} else if idx == 256 {
+			break
+		}
+		idx = n256.nextPresentIdx(idx)
+		if idx < 256 {
+			it.idxes[depth] = idx
+			child := n256.children[idx]
+			if child.kind == typeLeaf {
+				it.idxes[depth]++
+				return child
+			}
+			it.nodes = append(it.nodes, child)
+			it.idxes = append(it.idxes, inplaceIndex)
+			return it.next()
+		}
+	}
+	it.nodes = it.nodes[:depth]
+	it.idxes = it.idxes[:depth]
+	if depth == 0 {
+		return nullArtNode
+	}
+	it.idxes[depth-1]++
+	return it.next()
+}
+
+func (it *baseIter) prev() artNode {
+	depth := len(it.nodes) - 1
+	curr := it.nodes[depth]
+	idx := it.idxes[depth]
+	idx--
+	switch curr.kind {
+	case typeNode4:
+		n4 := it.allocator.getNode4(curr.addr)
+		idx = min(idx, int(n4.nodeNum)-1)
+		if idx >= 0 {
+			it.idxes[depth] = idx
+			child := n4.children[idx]
+			if child.kind == typeLeaf {
+				return child
+			}
+			it.nodes = append(it.nodes, child)
+			it.idxes = append(it.idxes, node256cap)
+			return it.prev()
+		} else if idx == inplaceIndex {
+			it.idxes[depth] = idx
+			if !n4.inplaceLeaf.addr.IsNull() {
+				return n4.inplaceLeaf
+			}
+		}
+	case typeNode16:
+		n16 := it.allocator.getNode16(curr.addr)
+		idx = min(idx, int(n16.nodeNum)-1)
+		if idx >= 0 {
+			it.idxes[depth] = idx
+			child := n16.children[idx]
+			if child.kind == typeLeaf {
+				return child
+			}
+			it.nodes = append(it.nodes, child)
+			it.idxes = append(it.idxes, node256cap)
+			return it.prev()
+		} else if idx == inplaceIndex {
+			it.idxes[depth] = idx
+			if !n16.inplaceLeaf.addr.IsNull() {
+				return n16.inplaceLeaf
+			}
+		}
+	case typeNode48:
+		n48 := it.allocator.getNode48(curr.addr)
+		if idx >= 0 {
+			idx = n48.prevPresentIdx(idx)
+		}
+		if idx >= 0 {
+			it.idxes[depth] = idx
+			child := n48.children[n48.keys[idx]]
+			if child.kind == typeLeaf {
+				return child
+			}
+			it.nodes = append(it.nodes, child)
+			it.idxes = append(it.idxes, node256cap)
+			return it.prev()
+		} else if idx == inplaceIndex {
+			it.idxes[depth] = idx
+			if !n48.inplaceLeaf.addr.IsNull() {
+				return n48.inplaceLeaf
+			}
+		}
+	case typeNode256:
+		n256 := it.allocator.getNode256(curr.addr)
+		if idx >= 0 {
+			idx = n256.prevPresentIdx(idx)
+		}
+		if idx >= 0 {
+			it.idxes[depth] = idx
+			child := n256.children[idx]
+			if child.kind == typeLeaf {
+				return child
+			}
+			it.nodes = append(it.nodes, child)
+			it.idxes = append(it.idxes, node256cap)
+			return it.prev()
+		} else if idx == -1 {
+			it.idxes[depth] = idx
+			if !n256.inplaceLeaf.addr.IsNull() {
+				return n256.inplaceLeaf
+			}
+		}
+	}
+	it.nodes = it.nodes[:depth]
+	it.idxes = it.idxes[:depth]
+	if depth == 0 {
+		return nullArtNode
+	}
+	return it.prev()
+}
diff --git a/internal/unionstore/art/art_node.go b/internal/unionstore/art/art_node.go
index f94794c..75c8d11 100644
--- a/internal/unionstore/art/art_node.go
+++ b/internal/unionstore/art/art_node.go
@@ -148,8 +148,6 @@ func (an *artNode) asNode(a *artAllocator) *nodeBase {
 }
 
 // at returns the nth child of the node, used for test.
-//
-//nolint:unused
 func (an *artNode) at(a *artAllocator, idx int) artNode {
 	switch an.kind {
 	case typeNode4:
@@ -425,7 +423,7 @@ func (an *artNode) findChild(a *artAllocator, c byte, inplace bool) (int, artNod
 		n48 := an.asNode48(a)
 		idx := n48.findChild(c)
 		if idx != notExistIndex {
-			return idx, n48.children[idx]
+			return idx, n48.children[n48.keys[idx]]
 		}
 	case typeNode256:
 		n256 := an.asNode256(a)
@@ -465,7 +463,7 @@ func (n16 *node16) findChild(c byte) int {
 
 func (n48 *node48) findChild(c byte) int {
 	if n48.present[c>>n48s]&(1<<(c%n48m)) != 0 {
-		return int(n48.keys[c])
+		return int(c)
 	}
 	return notExistIndex
 }
@@ -497,7 +495,7 @@ func (an *artNode) replaceChild(a *artAllocator, c byte, child artNode) {
 		n48 := an.asNode48(a)
 		idx := n48.findChild(c)
 		if idx != notExistIndex {
-			n48.children[idx] = child
+			n48.children[n48.keys[idx]] = child
 			return
 		}
 	case typeNode256:
diff --git a/tikv/gc.go b/tikv/gc.go
index a3b4eef..da0022f 100644
--- a/tikv/gc.go
+++ b/tikv/gc.go
@@ -309,7 +309,7 @@ const unsafeDestroyRangeTimeout = 5 * time.Minute
 // multiple times on an single range.
 func (s *KVStore) UnsafeDestroyRange(ctx context.Context, startKey []byte, endKey []byte) error {
 	// Get all stores every time deleting a region. So the store list is less probably to be stale.
-	stores, err := s.listStoresForUnsafeDestory(ctx)
+	stores, err := s.listStoresForUnsafeDestroy(ctx)
 	if err != nil {
 		metrics.TiKVUnsafeDestroyRangeFailuresCounterVec.WithLabelValues("get_stores").Inc()
 		return err
@@ -366,7 +366,7 @@ func (s *KVStore) UnsafeDestroyRange(ctx context.Context, startKey []byte, endKe
 	return nil
 }
 
-func (s *KVStore) listStoresForUnsafeDestory(ctx context.Context) ([]*metapb.Store, error) {
+func (s *KVStore) listStoresForUnsafeDestroy(ctx context.Context) ([]*metapb.Store, error) {
 	stores, err := s.pdClient.GetAllStores(ctx)
 	if err != nil {
 		return nil, errors.WithStack(err)
diff --git a/tikvrpc/tikvrpc.go b/tikvrpc/tikvrpc.go
index 350c5b7..96a480c 100644
--- a/tikvrpc/tikvrpc.go
+++ b/tikvrpc/tikvrpc.go
@@ -100,6 +100,7 @@ const (
 	CmdLockWaitInfo
 
 	CmdGetHealthFeedback
+	CmdBroadcastTxnStatus
 
 	CmdCop CmdType = 512 + iota
 	CmdCopStream
@@ -221,6 +222,8 @@ func (t CmdType) String() string {
 		return "LockWaitInfo"
 	case CmdGetHealthFeedback:
 		return "GetHealthFeedback"
+	case CmdBroadcastTxnStatus:
+		return "BroadcastTxnStatus"
 	case CmdFlashbackToVersion:
 		return "FlashbackToVersion"
 	case CmdPrepareFlashbackToVersion:
@@ -568,6 +571,10 @@ func (req *Request) GetHealthFeedback() *kvrpcpb.GetHealthFeedbackRequest {
 	return req.Req.(*kvrpcpb.GetHealthFeedbackRequest)
 }
 
+func (req *Request) BroadcastTxnStatus() *kvrpcpb.BroadcastTxnStatusRequest {
+	return req.Req.(*kvrpcpb.BroadcastTxnStatusRequest)
+}
+
 // FlashbackToVersion returns FlashbackToVersionRequest in request.
 func (req *Request) FlashbackToVersion() *kvrpcpb.FlashbackToVersionRequest {
 	return req.Req.(*kvrpcpb.FlashbackToVersionRequest)
@@ -653,6 +660,8 @@ func (req *Request) ToBatchCommandsRequest() *tikvpb.BatchCommandsRequest_Reques
 		return &tikvpb.BatchCommandsRequest_Request{Cmd: &tikvpb.BatchCommandsRequest_Request_BufferBatchGet{BufferBatchGet: req.BufferBatchGet()}}
 	case CmdGetHealthFeedback:
 		return &tikvpb.BatchCommandsRequest_Request{Cmd: &tikvpb.BatchCommandsRequest_Request_GetHealthFeedback{GetHealthFeedback: req.GetHealthFeedback()}}
+	case CmdBroadcastTxnStatus:
+		return &tikvpb.BatchCommandsRequest_Request{Cmd: &tikvpb.BatchCommandsRequest_Request_BroadcastTxnStatus{BroadcastTxnStatus: req.BroadcastTxnStatus()}}
 	}
 	return nil
 }
@@ -730,6 +739,8 @@ func FromBatchCommandsResponse(res *tikvpb.BatchCommandsResponse_Response) (*Res
 		return &Response{Resp: res.BufferBatchGet}, nil
 	case *tikvpb.BatchCommandsResponse_Response_GetHealthFeedback:
 		return &Response{Resp: res.GetHealthFeedback}, nil
+	case *tikvpb.BatchCommandsResponse_Response_BroadcastTxnStatus:
+		return &Response{Resp: res.BroadcastTxnStatus}, nil
 	}
 	panic("unreachable")
 }
@@ -1143,6 +1154,8 @@ func CallRPC(ctx context.Context, client tikvpb.TikvClient, req *Request) (*Resp
 		resp.Resp, err = client.KvBufferBatchGet(ctx, req.BufferBatchGet())
 	case CmdGetHealthFeedback:
 		resp.Resp, err = client.GetHealthFeedback(ctx, req.GetHealthFeedback())
+	case CmdBroadcastTxnStatus:
+		resp.Resp, err = client.BroadcastTxnStatus(ctx, req.BroadcastTxnStatus())
 	default:
 		return nil, errors.Errorf("invalid request type: %v", req.Type)
 	}
diff --git a/txnkv/transaction/2pc.go b/txnkv/transaction/2pc.go
index fbc8a85..2a01bef 100644
--- a/txnkv/transaction/2pc.go
+++ b/txnkv/transaction/2pc.go
@@ -164,7 +164,7 @@ type twoPhaseCommitter struct {
 	}
 
 	useAsyncCommit    uint32
-	minCommitTS       uint64
+	minCommitTSMgr    *minCommitTsManager
 	maxCommitTS       uint64
 	prewriteStarted   bool
 	prewriteCancelled uint32
@@ -477,6 +477,7 @@ func newTwoPhaseCommitter(txn *KVTxn, sessionID uint64) (*twoPhaseCommitter, err
 		binlog:            txn.binlog,
 		diskFullOpt:       kvrpcpb.DiskFullOpt_NotAllowedOnFull,
 		resourceGroupName: txn.resourceGroupName,
+		minCommitTSMgr:    newMinCommitTsManager(),
 	}
 	return committer, nil
 }
@@ -1137,6 +1138,69 @@ const (
 	stateClosed
 )
 
+// WriteAccessLevel represents the level of write access required to modify the value
+type WriteAccessLevel int
+
+const (
+	ttlAccess   WriteAccessLevel = 1
+	twoPCAccess WriteAccessLevel = 2
+)
+
+// minCommitTsManager manages a minimum commit timestamp with different write access levels.
+type minCommitTsManager struct {
+	mutex               sync.Mutex
+	value               uint64
+	requiredWriteAccess WriteAccessLevel
+}
+
+// newMinCommitTsManager creates and returns a new minCommitTsManager.
+func newMinCommitTsManager() *minCommitTsManager {
+	return &minCommitTsManager{requiredWriteAccess: ttlAccess}
+}
+
+// tryUpdate update the value if the provided write access level is sufficient and
+// the new value is greater.
+func (m *minCommitTsManager) tryUpdate(newValue uint64, writeAccess WriteAccessLevel) {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	if writeAccess < m.requiredWriteAccess {
+		return
+	}
+
+	if newValue > m.value {
+		m.value = newValue
+	}
+}
+
+// elevateWriteAccess elevates the required write access level.
+// It returns the current value.
+func (m *minCommitTsManager) elevateWriteAccess(newLevel WriteAccessLevel) uint64 {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	if newLevel > m.requiredWriteAccess {
+		m.requiredWriteAccess = newLevel
+	}
+	return m.value
+}
+
+// get returns the current value. This is a read operation and doesn't require write access.
+func (m *minCommitTsManager) get() uint64 {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	return m.value
+}
+
+// getRequiredWriteAccess returns the current required write access level.
+func (m *minCommitTsManager) getRequiredWriteAccess() WriteAccessLevel {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	return m.requiredWriteAccess
+}
+
 type ttlManager struct {
 	state   ttlManagerState
 	ch      chan struct{}
@@ -1175,7 +1239,11 @@ func (tm *ttlManager) reset() {
 const keepAliveMaxBackoff = 20000
 const pessimisticLockMaxBackoff = 20000
 const maxConsecutiveFailure = 10
+const broadcastGracePeriod = 5 * time.Second
+const broadcastMaxBackoff = 10000
 
+// keepAlive keeps sending heartbeat to update the primary key's TTL
+// For pipelined transactions, it also updates min_commit_ts, and broadcasts it to all TiKVs.
 func keepAlive(
 	c *twoPhaseCommitter, closeCh chan struct{}, tm *ttlManager, primaryKey []byte,
 	lockCtx *kv.LockCtx, isPipelinedTxn bool,
@@ -1234,6 +1302,14 @@ func keepAlive(
 				return
 			}
 
+			// update minCommitTS, if it's a non-async-commit pipelined transaction
+			if isPipelinedTxn &&
+				!c.isOnePC() &&
+				!c.isAsyncCommit() &&
+				c.minCommitTSMgr.getRequiredWriteAccess() <= ttlAccess {
+				c.minCommitTSMgr.tryUpdate(now, ttlAccess)
+			}
+
 			newTTL := uptime + atomic.LoadUint64(&ManagedLockTTL)
 			logutil.Logger(bo.GetCtx()).Info("send TxnHeartBeat",
 				zap.Uint64("startTS", c.startTS),
@@ -1241,7 +1317,9 @@ func keepAlive(
 				zap.Bool("isPipelinedTxn", isPipelinedTxn),
 			)
 			startTime := time.Now()
-			_, stopHeartBeat, err := sendTxnHeartBeat(bo, c.store, primaryKey, c.startTS, newTTL)
+			_, stopHeartBeat, err := sendTxnHeartBeat(
+				bo, c.store, primaryKey, c.startTS, newTTL, c.minCommitTSMgr.get(),
+			)
 			if err != nil {
 				keepFail++
 				metrics.TxnHeartBeatHistogramError.Observe(time.Since(startTime).Seconds())
@@ -1265,19 +1343,122 @@ func keepAlive(
 					}
 					return
 				}
-				continue
+			} else {
+				keepFail = 0
+				metrics.TxnHeartBeatHistogramOK.Observe(time.Since(startTime).Seconds())
+			}
+
+			// broadcast to all stores
+			if isPipelinedTxn {
+				broadcastToAllStores(
+					c.txn,
+					c.store,
+					retry.NewBackofferWithVars(
+						context.Background(),
+						broadcastMaxBackoff,
+						c.txn.vars,
+					),
+					&kvrpcpb.TxnStatus{
+						StartTs:     c.startTS,
+						MinCommitTs: c.minCommitTSMgr.get(),
+						CommitTs:    0,
+						RolledBack:  false,
+						IsCompleted: false,
+					},
+					c.resourceGroupName,
+					c.resourceGroupTag,
+				)
 			}
-			keepFail = 0
-			metrics.TxnHeartBeatHistogramOK.Observe(time.Since(startTime).Seconds())
 		}
 	}
 }
 
-func sendTxnHeartBeat(bo *retry.Backoffer, store kvstore, primary []byte, startTS, ttl uint64) (newTTL uint64, stopHeartBeat bool, err error) {
+const broadcastRpcTimeout = time.Second * 5
+const broadcastMaxConcurrency = 10
+
+// broadcastToAllStores asynchronously broadcasts the transaction status to all stores.
+// Errors are ignored.
+func broadcastToAllStores(
+	txn *KVTxn,
+	store kvstore,
+	bo *retry.Backoffer,
+	status *kvrpcpb.TxnStatus,
+	resourceGroupName string,
+	resourceGroupTag []byte,
+) {
+	broadcastFunc := func() {
+		stores := store.GetRegionCache().GetStoresByType(tikvrpc.TiKV)
+		concurrency := min(broadcastMaxConcurrency, len(stores))
+		rateLimit := make(chan struct{}, concurrency)
+
+		var wg sync.WaitGroup
+
+		for _, s := range stores {
+			rateLimit <- struct{}{}
+			wg.Add(1)
+			target := s
+
+			err := txn.spawnWithStorePool(func() {
+				defer wg.Done()
+				defer func() { <-rateLimit }()
+
+				req := tikvrpc.NewRequest(
+					tikvrpc.CmdBroadcastTxnStatus, &kvrpcpb.BroadcastTxnStatusRequest{
+						TxnStatus: []*kvrpcpb.TxnStatus{status},
+					},
+				)
+				req.Context.ClusterId = store.GetClusterID()
+				req.Context.ResourceControlContext = &kvrpcpb.ResourceControlContext{
+					ResourceGroupName: resourceGroupName,
+				}
+				req.Context.ResourceGroupTag = resourceGroupTag
+
+				_, err := store.GetTiKVClient().SendRequest(
+					bo.GetCtx(),
+					target.GetAddr(),
+					req,
+					broadcastRpcTimeout,
+				)
+				if err != nil {
+					logutil.Logger(store.Ctx()).Info(
+						"broadcast txn status failed",
+						zap.Uint64("storeID", target.StoreID()),
+						zap.String("storeAddr", target.GetAddr()),
+						zap.Stringer("status", status),
+						zap.Error(err),
+					)
+				}
+			})
+
+			if err != nil {
+				// If spawning the goroutine fails, release the slot and mark done
+				<-rateLimit
+				wg.Done()
+				logutil.Logger(store.Ctx()).Error("failed to spawn worker goroutine", zap.Error(err))
+			}
+		}
+
+		wg.Wait()
+	}
+
+	if err := txn.spawnWithStorePool(broadcastFunc); err != nil {
+		logutil.Logger(store.Ctx()).Error("failed to spawn goroutine for broadcasting txn status",
+			zap.Error(err))
+	}
+}
+
+func sendTxnHeartBeat(
+	bo *retry.Backoffer,
+	store kvstore,
+	primary []byte,
+	startTS, ttl uint64,
+	minCommitTS uint64,
+) (newTTL uint64, stopHeartBeat bool, err error) {
 	req := tikvrpc.NewRequest(tikvrpc.CmdTxnHeartBeat, &kvrpcpb.TxnHeartBeatRequest{
 		PrimaryLock:   primary,
 		StartVersion:  startTS,
 		AdviseLockTtl: ttl,
+		MinCommitTs:   minCommitTS,
 	})
 	for {
 		loc, err := store.GetRegionCache().LocateKey(bo, primary)
@@ -1424,6 +1605,7 @@ func (c *twoPhaseCommitter) cleanup(ctx context.Context) {
 		var err error
 		if c.txn.IsPipelined() {
 			// TODO: cleanup pipelined txn
+			// TODO: broadcast txn status
 		} else if !c.isOnePC() {
 			err = c.cleanupMutations(retry.NewBackofferWithVars(cleanupKeysCtx, cleanupMaxBackoff, c.txn.vars), c.mutations)
 		} else if c.isPessimistic {
@@ -1444,6 +1626,7 @@ func (c *twoPhaseCommitter) cleanup(ctx context.Context) {
 
 // execute executes the two-phase commit protocol.
 func (c *twoPhaseCommitter) execute(ctx context.Context) (err error) {
+	c.minCommitTSMgr.elevateWriteAccess(twoPCAccess)
 	var binlogSkipped bool
 	defer func() {
 		if c.isOnePC() {
@@ -1547,7 +1730,7 @@ func (c *twoPhaseCommitter) execute(ctx context.Context) (err error) {
 		}
 		commitDetail.GetLatestTsTime = time.Since(start)
 		// Plus 1 to avoid producing the same commit TS with previously committed transactions
-		c.minCommitTS = latestTS + 1
+		c.minCommitTSMgr.tryUpdate(latestTS+1, twoPCAccess)
 	}
 	// Calculate maxCommitTS if necessary
 	if commitTSMayBeCalculated {
@@ -1666,10 +1849,10 @@ func (c *twoPhaseCommitter) execute(ctx context.Context) (err error) {
 	}
 
 	if c.isAsyncCommit() {
-		if c.minCommitTS == 0 {
+		if c.minCommitTSMgr.get() == 0 {
 			return errors.Errorf("session %d invalid minCommitTS for async commit protocol after prewrite, startTS=%v", c.sessionID, c.startTS)
 		}
-		commitTS = c.minCommitTS
+		commitTS = c.minCommitTSMgr.get()
 	} else {
 		start = time.Now()
 		logutil.Event(ctx, "start get commit ts")
diff --git a/txnkv/transaction/pipelined_flush.go b/txnkv/transaction/pipelined_flush.go
index 3ba39d2..8cadf63 100644
--- a/txnkv/transaction/pipelined_flush.go
+++ b/txnkv/transaction/pipelined_flush.go
@@ -330,14 +330,32 @@ func (c *twoPhaseCommitter) commitFlushedMutations(bo *retry.Backoffer) error {
 	if err = c.commitMutations(bo, &primaryMutation); err != nil {
 		return errors.Trace(err)
 	}
-	c.mu.RLock()
+	c.mu.Lock()
 	c.mu.committed = true
-	c.mu.RUnlock()
+	c.mu.Unlock()
 	logutil.Logger(bo.GetCtx()).Info(
 		"[pipelined dml] transaction is committed",
 		zap.Uint64("startTS", c.startTS),
 		zap.Uint64("commitTS", commitTS),
 	)
+	broadcastToAllStores(
+		c.txn,
+		c.store,
+		retry.NewBackofferWithVars(
+			bo.GetCtx(),
+			broadcastMaxBackoff,
+			c.txn.vars,
+		),
+		&kvrpcpb.TxnStatus{
+			StartTs:     c.startTS,
+			MinCommitTs: c.minCommitTSMgr.get(),
+			CommitTs:    commitTS,
+			RolledBack:  false,
+			IsCompleted: false,
+		},
+		c.resourceGroupName,
+		c.resourceGroupTag,
+	)
 
 	if _, err := util.EvalFailpoint("pipelinedSkipResolveLock"); err == nil {
 		return nil
@@ -439,13 +457,17 @@ func (c *twoPhaseCommitter) resolveFlushedLocks(bo *retry.Backoffer, start, end
 	const RESOLVE_CONCURRENCY = 8
 	var resolved atomic.Uint64
 	handler, err := c.buildPipelinedResolveHandler(commit, &resolved)
+	commitTs := uint64(0)
+	if commit {
+		commitTs = atomic.LoadUint64(&c.commitTS)
+	}
 	if err != nil {
 		logutil.Logger(bo.GetCtx()).Error(
 			"[pipelined dml] build buildPipelinedResolveHandler error",
 			zap.Error(err),
 			zap.Uint64("resolved regions", resolved.Load()),
 			zap.Uint64("startTS", c.startTS),
-			zap.Uint64("commitTS", atomic.LoadUint64(&c.commitTS)),
+			zap.Uint64("commitTS", commitTs),
 		)
 		return
 	}
@@ -470,7 +492,7 @@ func (c *twoPhaseCommitter) resolveFlushedLocks(bo *retry.Backoffer, start, end
 				zap.String("txn-status", status),
 				zap.Uint64("resolved regions", resolved.Load()),
 				zap.Uint64("startTS", c.startTS),
-				zap.Uint64("commitTS", atomic.LoadUint64(&c.commitTS)),
+				zap.Uint64("commitTS", commitTs),
 				zap.Uint64("session", c.sessionID),
 				zap.Error(err),
 			)
@@ -479,9 +501,33 @@ func (c *twoPhaseCommitter) resolveFlushedLocks(bo *retry.Backoffer, start, end
 				zap.String("txn-status", status),
 				zap.Uint64("resolved regions", resolved.Load()),
 				zap.Uint64("startTS", c.startTS),
-				zap.Uint64("commitTS", atomic.LoadUint64(&c.commitTS)),
+				zap.Uint64("commitTS", commitTs),
 				zap.Uint64("session", c.sessionID),
 			)
+
+			// wait a while before notifying txn_status_cache to evict the txn,
+			// which tolerates slow followers and avoids the situation that the
+			// txn is evicted before the follower catches up.
+			time.Sleep(broadcastGracePeriod)
+
+			broadcastToAllStores(
+				c.txn,
+				c.store,
+				retry.NewBackofferWithVars(
+					bo.GetCtx(),
+					broadcastMaxBackoff,
+					c.txn.vars,
+				),
+				&kvrpcpb.TxnStatus{
+					StartTs:     c.startTS,
+					MinCommitTs: 0,
+					CommitTs:    commitTs,
+					RolledBack:  !commit,
+					IsCompleted: true,
+				},
+				c.resourceGroupName,
+				c.resourceGroupTag,
+			)
 		}
 	}()
 }
diff --git a/txnkv/transaction/prewrite.go b/txnkv/transaction/prewrite.go
index a50e407..26e317e 100644
--- a/txnkv/transaction/prewrite.go
+++ b/txnkv/transaction/prewrite.go
@@ -116,7 +116,7 @@ func (c *twoPhaseCommitter) buildPrewriteRequest(batch batchMutations, txnSize u
 		}
 	}
 	c.mu.Lock()
-	minCommitTS := c.minCommitTS
+	minCommitTS := c.minCommitTSMgr.get()
 	c.mu.Unlock()
 	if c.forUpdateTS > 0 && c.forUpdateTS >= minCommitTS {
 		minCommitTS = c.forUpdateTS + 1
@@ -387,7 +387,7 @@ func (action actionPrewrite) handleSingleBatch(
 					c.setOnePC(false)
 					c.setAsyncCommit(false)
 				} else {
-					// For 1PC, there's no racing to access to access `onePCCommmitTS` so it's safe
+					// For 1PC, there's no racing to access `onePCCommitTS` so it's safe
 					// not to lock the mutex.
 					if c.onePCCommitTS != 0 {
 						logutil.Logger(bo.GetCtx()).Fatal(
@@ -419,8 +419,8 @@ func (action actionPrewrite) handleSingleBatch(
 					c.setAsyncCommit(false)
 				} else {
 					c.mu.Lock()
-					if prewriteResp.MinCommitTs > c.minCommitTS {
-						c.minCommitTS = prewriteResp.MinCommitTs
+					if prewriteResp.MinCommitTs > c.minCommitTSMgr.get() {
+						c.minCommitTSMgr.tryUpdate(prewriteResp.MinCommitTs, twoPCAccess)
 					}
 					c.mu.Unlock()
 				}
diff --git a/txnkv/transaction/txn.go b/txnkv/transaction/txn.go
index dfa1461..7c124a6 100644
--- a/txnkv/transaction/txn.go
+++ b/txnkv/transaction/txn.go
@@ -841,7 +841,26 @@ func (txn *KVTxn) Rollback() error {
 		txn.committer.ttlManager.close()
 		// no need to clean up locks when no flush triggered.
 		pipelinedStart, pipelinedEnd := txn.committer.pipelinedCommitInfo.pipelinedStart, txn.committer.pipelinedCommitInfo.pipelinedEnd
-		if len(pipelinedStart) != 0 && len(pipelinedEnd) != 0 {
+		needCleanUpLocks := len(pipelinedStart) != 0 && len(pipelinedEnd) != 0
+		broadcastToAllStores(
+			txn,
+			txn.committer.store,
+			retry.NewBackofferWithVars(
+				txn.store.Ctx(),
+				broadcastMaxBackoff,
+				txn.committer.txn.vars,
+			),
+			&kvrpcpb.TxnStatus{
+				StartTs:     txn.startTS,
+				MinCommitTs: txn.committer.minCommitTSMgr.get(),
+				CommitTs:    0,
+				RolledBack:  true,
+				IsCompleted: !needCleanUpLocks,
+			},
+			txn.resourceGroupName,
+			txn.resourceGroupTag,
+		)
+		if needCleanUpLocks {
 			rollbackBo := retry.NewBackofferWithVars(txn.store.Ctx(), CommitSecondaryMaxBackoff, txn.vars)
 			txn.committer.resolveFlushedLocks(rollbackBo, pipelinedStart, pipelinedEnd, false)
 		}


Test patch:
diff --git a/integration_tests/go.mod b/integration_tests/go.mod
index 8e2f7d9..c5b8f4a 100644
--- a/integration_tests/go.mod
+++ b/integration_tests/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/ninedraft/israce v0.0.3
 	github.com/pingcap/errors v0.11.5-0.20240318064555-6bd07397691f
 	github.com/pingcap/failpoint v0.0.0-20240527053858-9b3b6e34194a
-	github.com/pingcap/kvproto v0.0.0-20240620063548-118a4cab53e4
+	github.com/pingcap/kvproto v0.0.0-20240924080114-4a3e17f5e62d
 	github.com/pingcap/tidb v1.1.0-beta.0.20240703042657-230bbc2ef5ef
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.9.0
diff --git a/integration_tests/go.sum b/integration_tests/go.sum
index d50e94f..77a57eb 100644
--- a/integration_tests/go.sum
+++ b/integration_tests/go.sum
@@ -357,8 +357,8 @@ github.com/pingcap/fn v1.0.0/go.mod h1:u9WZ1ZiOD1RpNhcI42RucFh/lBuzTu6rw88a+oF2Z
 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-20240620063548-118a4cab53e4 h1:6aIKNB2YGAec4IUDLw6G2eDECiGiufZcgEbZSCELBx0=
-github.com/pingcap/kvproto v0.0.0-20240620063548-118a4cab53e4/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8=
+github.com/pingcap/kvproto v0.0.0-20240924080114-4a3e17f5e62d h1:vSdKTrF6kpcd56G5BLP0Bz88Nho2tDo7IR1+oSsBAfc=
+github.com/pingcap/kvproto v0.0.0-20240924080114-4a3e17f5e62d/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8=
 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.20240314023424-862ccc32f18d h1:y3EueKVfVykdpTyfUnQGqft0ud+xVFuCdp1XkVL0X1E=
diff --git a/internal/locate/region_request_test.go b/internal/locate/region_request_test.go
index 8e10e30..c43aecb 100644
--- a/internal/locate/region_request_test.go
+++ b/internal/locate/region_request_test.go
@@ -582,6 +582,10 @@ func (s *mockTikvGrpcServer) GetHealthFeedback(ctx context.Context, request *kvr
 	return nil, errors.New("unreachable")
 }
 
+func (s *mockTikvGrpcServer) BroadcastTxnStatus(ctx context.Context, request *kvrpcpb.BroadcastTxnStatusRequest) (*kvrpcpb.BroadcastTxnStatusResponse, error) {
+	return nil, errors.New("unreachable")
+}
+
 func (s *testRegionRequestToSingleStoreSuite) TestNoReloadRegionForGrpcWhenCtxCanceled() {
 	// prepare a mock tikv grpc server
 	addr := "localhost:56341"
diff --git a/internal/unionstore/art/art_iterator_test.go b/internal/unionstore/art/art_iterator_test.go
new file mode 100644
index 0000000..7dcd504
--- /dev/null
+++ b/internal/unionstore/art/art_iterator_test.go
@@ -0,0 +1,224 @@
+// Copyright 2024 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 art
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/tikv/client-go/v2/internal/unionstore/arena"
+)
+
+func TestIterateNodeCapacity(t *testing.T) {
+	check := func(tree *ART, startKey, endKey []byte, startVal, endVal int) {
+		// iter
+		it, err := tree.Iter(startKey, endKey)
+		require.Nil(t, err)
+		handles := make([]arena.MemKeyHandle, 0, endVal-startVal+1)
+		for i := startVal; i <= endVal; i++ {
+			require.True(t, it.Valid(), i)
+			require.Equal(t, it.Key(), []byte{byte(i)})
+			require.Equal(t, it.Value(), []byte{byte(i)})
+			handles = append(handles, it.Handle())
+			require.Nil(t, it.Next())
+		}
+		require.False(t, it.Valid())
+		require.Error(t, it.Next())
+		for i, handle := range handles {
+			require.Equal(t, tree.GetKeyByHandle(handle), []byte{byte(startVal + i)})
+			val, valid := tree.GetValueByHandle(handle)
+			require.True(t, valid)
+			require.Equal(t, val, []byte{byte(startVal + i)})
+		}
+		// reverse iter
+		it, err = tree.IterReverse(endKey, startKey)
+		require.Nil(t, err)
+		handles = handles[:0]
+		for i := endVal; i >= startVal; i-- {
+			require.True(t, it.Valid(), i)
+			require.Equal(t, it.Key(), []byte{byte(i)})
+			require.Equal(t, it.Value(), []byte{byte(i)})
+			handles = append(handles, it.Handle())
+			require.Nil(t, it.Next())
+		}
+		require.False(t, it.Valid())
+		require.Error(t, it.Next())
+		for i, handle := range handles {
+			require.Equal(t, tree.GetKeyByHandle(handle), []byte{byte(endVal - i)})
+			val, valid := tree.GetValueByHandle(handle)
+			require.True(t, valid)
+			require.Equal(t, val, []byte{byte(endVal - i)})
+		}
+	}
+	for _, capacity := range []int{node4cap, node16cap, node48cap, node256cap} {
+		tree := New()
+		for i := 0; i < capacity; i++ {
+			key := []byte{byte(i)}
+			require.Nil(t, tree.Set(key, key))
+		}
+		check(tree, nil, nil, 0, capacity-1)
+		mid := capacity / 2
+		check(tree, []byte{byte(mid)}, nil, mid, capacity-1) // lower bound is inclusive
+		check(tree, nil, []byte{byte(mid)}, 0, mid-1)        // upper bound is exclusive
+	}
+}
+
+func TestIterSeekLeaf(t *testing.T) {
+	for _, capacity := range []int{node4cap, node16cap, node48cap, node256cap} {
+		tree := New()
+		for i := 0; i < capacity; i++ {
+			key := []byte{byte(i)}
+			require.Nil(t, tree.Set(key, key))
+		}
+		for i := 0; i < capacity; i++ {
+			key := []byte{byte(i)}
+			it, err := tree.Iter(key, nil)
+			require.Nil(t, err)
+			idxes, nodes := it.seek(key)
+			require.Greater(t, len(idxes), 0)
+			require.Equal(t, len(idxes), len(nodes))
+			leafNode := nodes[len(nodes)-1].at(&tree.allocator, idxes[len(idxes)-1])
+			require.NotEqual(t, leafNode, nullArtNode)
+			leaf := leafNode.asLeaf(&tree.allocator)
+			require.Equal(t, leaf.GetKey(), key)
+		}
+	}
+}
+
+func TestMultiLevelIterate(t *testing.T) {
+	tree := New()
+	var keys [][]byte
+	for i := 0; i < 20; i++ {
+		key := make([]byte, i+1)
+		keys = append(keys, key)
+	}
+	for _, key := range keys {
+		require.Nil(t, tree.Set(key, key))
+	}
+	// iter
+	it, err := tree.Iter(nil, nil)
+	require.Nil(t, err)
+	handles := make([]arena.MemKeyHandle, 0, len(keys))
+	for _, key := range keys {
+		require.True(t, it.Valid())
+		require.Equal(t, it.Key(), key)
+		require.Equal(t, it.Value(), key)
+		handles = append(handles, it.Handle())
+		require.Nil(t, it.Next())
+	}
+	require.False(t, it.Valid())
+	require.Error(t, it.Next())
+	for i, handle := range handles {
+		require.Equal(t, tree.GetKeyByHandle(handle), keys[i])
+		val, valid := tree.GetValueByHandle(handle)
+		require.True(t, valid)
+		require.Equal(t, val, keys[i])
+	}
+	// reverse iter
+	it, err = tree.IterReverse(nil, nil)
+	require.Nil(t, err)
+	handles = handles[:0]
+	for i := len(keys) - 1; i >= 0; i-- {
+		require.True(t, it.Valid())
+		require.Equal(t, it.Key(), keys[i])
+		require.Equal(t, it.Value(), keys[i])
+		handles = append(handles, it.Handle())
+		require.Nil(t, it.Next())
+	}
+	require.False(t, it.Valid())
+	require.Error(t, it.Next())
+	for i, handle := range handles {
+		require.Equal(t, tree.GetKeyByHandle(handle), keys[len(keys)-i-1])
+		val, valid := tree.GetValueByHandle(handle)
+		require.True(t, valid)
+		require.Equal(t, val, keys[len(keys)-i-1])
+	}
+}
+
+func TestSeekMeetLeaf(t *testing.T) {
+	tree := New()
+	require.Nil(t, tree.Set([]byte{1}, []byte{1}))
+	iter, err := tree.Iter([]byte{1}, []byte{1, 1})
+	require.Nil(t, err)
+	require.True(t, iter.Valid())
+	require.Equal(t, iter.Key(), []byte{1})
+	require.Equal(t, iter.Value(), []byte{1})
+	require.Nil(t, iter.Next())
+	require.False(t, iter.Valid())
+	require.Nil(t, tree.Set([]byte{2}, []byte{2}))
+	iter, err = tree.IterReverse([]byte{2, 2}, []byte{1, 1})
+	require.Nil(t, err)
+	require.True(t, iter.Valid())
+	require.Equal(t, iter.Key(), []byte{2})
+	require.Equal(t, iter.Value(), []byte{2})
+	require.Nil(t, iter.Next())
+	require.False(t, iter.Valid())
+
+	tree = New()
+	require.Nil(t, tree.Set([]byte{1, 1}, []byte{1, 1}))
+	iter, err = tree.Iter([]byte{1, 0, 0}, []byte{1, 2, 0})
+	require.Nil(t, err)
+	require.True(t, iter.Valid())
+	require.Equal(t, iter.Key(), []byte{1, 1})
+	require.Equal(t, iter.Value(), []byte{1, 1})
+	require.Nil(t, iter.Next())
+	require.False(t, iter.Valid())
+}
+
+func TestSeekInExistNode(t *testing.T) {
+	check := func(tree *ART, kind nodeKind) {
+		cnt := 0
+		switch kind {
+		case typeNode4:
+			cnt = node4cap
+		case typeNode16:
+			cnt = node16cap
+		case typeNode48:
+			cnt = node48cap
+		case typeNode256:
+			cnt = node48cap + 1
+		default:
+			panic("invalid node kind")
+		}
+		for i := 0; i < cnt; i++ {
+			key := []byte{byte(2 * i)}
+			require.Nil(t, tree.Set(key, key))
+		}
+		require.Equal(t, tree.root.kind, kind)
+		for i := 0; i < cnt-1; i++ {
+			it := &Iterator{
+				tree: tree,
+			}
+			idxes, _ := it.seek([]byte{byte(2*i + 1)})
+			expect := 0
+			switch kind {
+			case typeNode4, typeNode16:
+				// node4 and node16 is not indexed by char
+				expect = i + 1
+			case typeNode48, typeNode256:
+				// node48 and node256 is indexed by char
+				expect = 2 * (i + 1)
+			default:
+				panic("invalid node kind")
+			}
+			require.Equal(t, idxes, []int{expect})
+		}
+	}
+
+	check(New(), typeNode4)
+	check(New(), typeNode16)
+	check(New(), typeNode48)
+	check(New(), typeNode256)
+}
diff --git a/internal/unionstore/memdb_bench_test.go b/internal/unionstore/memdb_bench_test.go
index 2229609..2fc57de 100644
--- a/internal/unionstore/memdb_bench_test.go
+++ b/internal/unionstore/memdb_bench_test.go
@@ -177,10 +177,7 @@ func BenchmarkMemDbIter(b *testing.B) {
 	}
 
 	b.Run("RBT", func(b *testing.B) { fn(b, newRbtDBWithContext()) })
-	b.Run("ART", func(b *testing.B) {
-		b.Skip("unimplemented")
-		fn(b, newArtDBWithContext())
-	})
+	b.Run("ART", func(b *testing.B) { fn(b, newArtDBWithContext()) })
 }
 
 func BenchmarkMemDbCreation(b *testing.B) {
diff --git a/internal/unionstore/memdb_norace_test.go b/internal/unionstore/memdb_norace_test.go
index ffd4b3d..0718371 100644
--- a/internal/unionstore/memdb_norace_test.go
+++ b/internal/unionstore/memdb_norace_test.go
@@ -175,6 +175,17 @@ func TestRandomAB(t *testing.T) {
 func testRandomAB(t *testing.T, bufferA, bufferB MemBuffer) {
 	require := require.New(t)
 
+	checkIters := func(iter1, iter2 Iterator, k []byte) {
+		for iter1.Valid() {
+			require.True(iter2.Valid())
+			require.Equal(iter1.Key(), iter2.Key())
+			require.Equal(iter1.Value(), iter2.Value())
+			require.Nil(iter1.Next())
+			require.Nil(iter2.Next())
+		}
+		require.False(iter2.Valid())
+	}
+
 	const cnt = 50000
 	keys := make([][]byte, cnt)
 	for i := 0; i < cnt; i++ {
@@ -195,13 +206,30 @@ func testRandomAB(t *testing.T, bufferA, bufferB MemBuffer) {
 			bufferB.Release(h)
 		}
 
-		require.Equal(bufferA.Dirty(), bufferB.Dirty())
-		require.Equal(bufferA.Len(), bufferB.Len())
-		require.Equal(bufferA.Size(), bufferB.Size(), i)
-		key := keys[rand.Intn(i+1)]
-		v1, err1 := bufferA.Get(context.Background(), key)
-		v2, err2 := bufferB.Get(context.Background(), key)
-		require.Equal(err1, err2)
-		require.Equal(v1, v2)
+		if i%5000 == 0 {
+			// iter test is slow, run it less frequently
+			key := make([]byte, rand.Intn(19)+1)
+			rand2.Read(key)
+
+			require.Equal(bufferA.Dirty(), bufferB.Dirty())
+			require.Equal(bufferA.Len(), bufferB.Len())
+			require.Equal(bufferA.Size(), bufferB.Size(), i)
+			v1, err1 := bufferA.Get(context.Background(), key)
+			v2, err2 := bufferB.Get(context.Background(), key)
+			require.Equal(err1, err2)
+			require.Equal(v1, v2)
+
+			iter1, err := bufferA.Iter(key, nil)
+			require.Nil(err)
+			iter2, err := bufferB.Iter(key, nil)
+			require.Nil(err)
+			checkIters(iter1, iter2, key)
+
+			iter1, err = bufferA.IterReverse(nil, key)
+			require.Nil(err)
+			iter2, err = bufferB.IterReverse(nil, key)
+			require.Nil(err)
+			checkIters(iter1, iter2, key)
+		}
 	}
 }
diff --git a/internal/unionstore/memdb_test.go b/internal/unionstore/memdb_test.go
index f3996de..ac6f6d6 100644
--- a/internal/unionstore/memdb_test.go
+++ b/internal/unionstore/memdb_test.go
@@ -72,6 +72,7 @@ func testGetSet(t *testing.T, db MemBuffer) {
 
 func TestIterator(t *testing.T) {
 	testIterator(t, newRbtDBWithContext())
+	testIterator(t, newArtDBWithContext())
 }
 
 func testIterator(t *testing.T, db MemBuffer) {
@@ -122,6 +123,7 @@ func testIterator(t *testing.T, db MemBuffer) {
 
 func TestDiscard(t *testing.T) {
 	testDiscard(t, newRbtDBWithContext())
+	testDiscard(t, newArtDBWithContext())
 }
 
 func testDiscard(t *testing.T, db MemBuffer) {
@@ -174,6 +176,7 @@ func testDiscard(t *testing.T, db MemBuffer) {
 
 func TestFlushOverwrite(t *testing.T) {
 	testFlushOverwrite(t, newRbtDBWithContext())
+	testFlushOverwrite(t, newArtDBWithContext())
 }
 
 func testFlushOverwrite(t *testing.T, db MemBuffer) {
@@ -254,6 +257,7 @@ func testComplexUpdate(t *testing.T, db MemBuffer) {
 
 func TestNestedSandbox(t *testing.T) {
 	testNestedSandbox(t, newRbtDBWithContext())
+	testNestedSandbox(t, newArtDBWithContext())
 }
 
 func testNestedSandbox(t *testing.T, db MemBuffer) {
@@ -313,6 +317,7 @@ func testNestedSandbox(t *testing.T, db MemBuffer) {
 
 func TestOverwrite(t *testing.T) {
 	testOverwrite(t, newRbtDBWithContext())
+	testOverwrite(t, newArtDBWithContext())
 }
 
 func testOverwrite(t *testing.T, db MemBuffer) {
@@ -375,6 +380,7 @@ func testOverwrite(t *testing.T, db MemBuffer) {
 
 func TestReset(t *testing.T) {
 	testReset(t, newRbtDBWithContext())
+	testReset(t, newArtDBWithContext())
 }
 
 func testReset(t *testing.T, db interface {
@@ -488,6 +494,7 @@ func testDirty(t *testing.T, createDb func() MemBuffer) {
 
 func TestFlags(t *testing.T) {
 	testFlags(t, newRbtDBWithContext(), func(db MemBuffer) Iterator { return db.(*rbtDBWithContext).IterWithFlags(nil, nil) })
+	testFlags(t, newArtDBWithContext(), func(db MemBuffer) Iterator { return db.(*artDBWithContext).IterWithFlags(nil, nil) })
 }
 
 func testFlags(t *testing.T, db MemBuffer, iterWithFlags func(db MemBuffer) Iterator) {
@@ -614,7 +621,7 @@ const (
 	indexStep  = 2
 )
 
-func insertData(t *testing.T, buffer *MemDB) {
+func insertData(t *testing.T, buffer MemBuffer) {
 	for i := startIndex; i < testCount; i++ {
 		val := encodeInt(i * indexStep)
 		err := buffer.Set(val, val)
@@ -637,7 +644,7 @@ func valToStr(iter Iterator) string {
 	return string(val)
 }
 
-func checkNewIterator(t *testing.T, buffer *MemDB) {
+func checkNewIterator(t *testing.T, buffer MemBuffer) {
 	assert := assert.New(t)
 	for i := startIndex; i < testCount; i++ {
 		val := encodeInt(i * indexStep)
@@ -683,7 +690,7 @@ func checkNewIterator(t *testing.T, buffer *MemDB) {
 	iter.Close()
 }
 
-func mustGet(t *testing.T, buffer *MemDB) {
+func mustGet(t *testing.T, buffer MemBuffer) {
 	for i := startIndex; i < testCount; i++ {
 		s := encodeInt(i * indexStep)
 		val, err := buffer.Get(context.Background(), s)
@@ -693,14 +700,22 @@ func mustGet(t *testing.T, buffer *MemDB) {
 }
 
 func TestKVGetSet(t *testing.T) {
-	buffer := NewMemDB()
+	testKVGetSet(t, newRbtDBWithContext())
+	testKVGetSet(t, newArtDBWithContext())
+}
+
+func testKVGetSet(t *testing.T, buffer MemBuffer) {
 	insertData(t, buffer)
 	mustGet(t, buffer)
 }
 
 func TestNewIterator(t *testing.T) {
+	testNewIterator(t, newRbtDBWithContext())
+	testNewIterator(t, newArtDBWithContext())
+}
+
+func testNewIterator(t *testing.T, buffer MemBuffer) {
 	assert := assert.New(t)
-	buffer := NewMemDB()
 	// should be invalid
 	iter, err := buffer.Iter(nil, nil)
 	assert.Nil(err)
@@ -727,8 +742,12 @@ func NextUntil(it Iterator, fn FnKeyCmp) error {
 }
 
 func TestIterNextUntil(t *testing.T) {
+	testIterNextUntil(t, newRbtDBWithContext())
+	testIterNextUntil(t, newArtDBWithContext())
+}
+
+func testIterNextUntil(t *testing.T, buffer MemBuffer) {
 	assert := assert.New(t)
-	buffer := NewMemDB()
 	insertData(t, buffer)
 
 	iter, err := buffer.Iter(nil, nil)
@@ -742,14 +761,23 @@ func TestIterNextUntil(t *testing.T) {
 }
 
 func TestBasicNewIterator(t *testing.T) {
+	testBasicNewIterator(t, newRbtDBWithContext())
+	testBasicNewIterator(t, newArtDBWithContext())
+}
+
+func testBasicNewIterator(t *testing.T, buffer MemBuffer) {
 	assert := assert.New(t)
-	buffer := NewMemDB()
 	it, err := buffer.Iter([]byte("2"), nil)
 	assert.Nil(err)
 	assert.False(it.Valid())
 }
 
 func TestNewIteratorMin(t *testing.T) {
+	testNewIteratorMin(t, newRbtDBWithContext())
+	testNewIteratorMin(t, newArtDBWithContext())
+}
+
+func testNewIteratorMin(t *testing.T, buffer MemBuffer) {
 	assert := assert.New(t)
 	kvs := []struct {
 		key   string
@@ -762,7 +790,6 @@ func TestNewIteratorMin(t *testing.T) {
 		{"DATA_test_main_db_tbl_tbl_test_record__00000000000000000002_0002", "2"},
 		{"DATA_test_main_db_tbl_tbl_test_record__00000000000000000002_0003", "hello"},
 	}
-	buffer := NewMemDB()
 	for _, kv := range kvs {
 		err := buffer.Set([]byte(kv.key), []byte(kv.value))
 		assert.Nil(err)
@@ -928,6 +955,7 @@ func testMemDBCheckpoint(t *testing.T, buffer MemBuffer) {
 
 func TestBufferLimit(t *testing.T) {
 	testBufferLimit(t, newRbtDBWithContext())
+	testBufferLimit(t, newArtDBWithContext())
 }
 
 func testBufferLimit(t *testing.T, buffer MemBuffer) {
@@ -950,13 +978,17 @@ func testBufferLimit(t *testing.T, buffer MemBuffer) {
 }
 
 func TestUnsetTemporaryFlag(t *testing.T) {
+	testUnsetTemporaryFlag(t, newRbtDBWithContext())
+	testUnsetTemporaryFlag(t, newArtDBWithContext())
+}
+
+func testUnsetTemporaryFlag(t *testing.T, buffer MemBuffer) {
 	require := require.New(t)
-	db := NewMemDB()
 	key := []byte{1}
 	value := []byte{2}
-	db.SetWithFlags(key, value, kv.SetNeedConstraintCheckInPrewrite)
-	db.Set(key, value)
-	flags, err := db.GetFlags(key)
+	buffer.SetWithFlags(key, value, kv.SetNeedConstraintCheckInPrewrite)
+	buffer.Set(key, value)
+	flags, err := buffer.GetFlags(key)
 	require.Nil(err)
 	require.False(flags.HasNeedConstraintCheckInPrewrite())
 }
@@ -1047,3 +1079,31 @@ func testCleanupKeepPersistentFlag(t *testing.T, db MemBuffer) {
 	assert.True(flag.HasLocked())
 	assert.False(flag.HasPresumeKeyNotExists())
 }
+
+func TestIterNoResult(t *testing.T) {
+	testIterNoResult(t, newRbtDBWithContext())
+	testIterNoResult(t, newArtDBWithContext())
+}
+
+func testIterNoResult(t *testing.T, buffer MemBuffer) {
+	assert := assert.New(t)
+
+	assert.Nil(buffer.Set([]byte{1, 1}, []byte{1, 1}))
+	// Test lower bound and upper bound seek same position
+	iter, err := buffer.Iter([]byte{1, 0, 0}, []byte{1, 0, 1})
+	assert.Nil(err)
+	assert.False(iter.Valid())
+	iter, err = buffer.IterReverse([]byte{1, 0, 1}, []byte{1, 0, 0})
+	assert.Nil(err)
+	assert.False(iter.Valid())
+	// Test lower bound >= upper bound
+	iter, err = buffer.Iter([]byte{1, 0, 1}, []byte{1, 0, 0})
+	assert.Nil(err)
+	assert.False(iter.Valid())
+	iter, err = buffer.IterReverse([]byte{1, 0, 0}, []byte{1, 0, 1})
+	assert.Nil(err)
+	assert.False(iter.Valid())
+	iter, err = buffer.Iter([]byte{1, 1}, []byte{1, 1})
+	assert.Nil(err)
+	assert.False(iter.Valid())
+}
diff --git a/tikv/kv_test.go b/tikv/kv_test.go
index c2e4631..1a9a461 100644
--- a/tikv/kv_test.go
+++ b/tikv/kv_test.go
@@ -51,7 +51,6 @@ type testKVSuite struct {
 func (s *testKVSuite) SetupTest() {
 	client, cluster, pdClient, err := testutils.NewMockTiKV("", nil)
 	s.Require().Nil(err)
-	testutils.BootstrapWithSingleStore(cluster)
 	s.setGetMinResolvedTSByStoresIDs(func(ctx context.Context, ids []uint64) (uint64, map[uint64]uint64, error) {
 		return 0, nil, nil
 	})
diff --git a/txnkv/transaction/2pc_test.go b/txnkv/transaction/2pc_test.go
new file mode 100644
index 0000000..3e1a80d
--- /dev/null
+++ b/txnkv/transaction/2pc_test.go
@@ -0,0 +1,124 @@
+// Copyright 2024 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.
+
+// NOTE: The code in this file is based on code from the
+// TiDB project, licensed under the Apache License v 2.0
+//
+// https://github.com/pingcap/tidb/tree/cc5e161ac06827589c4966674597c137cc9e809c/store/tikv/tests/snapshot_test.go
+//
+
+// Copyright 2016 PingCAP, Inc.
+//
+// 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 transaction
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMinCommitTsManager(t *testing.T) {
+	t.Run(
+		"Initial state", func(t *testing.T) {
+			manager := newMinCommitTsManager()
+			assert.Equal(t, uint64(0), manager.get(), "Initial value should be 0")
+			assert.Equal(
+				t,
+				ttlAccess,
+				manager.getRequiredWriteAccess(),
+				"Initial write access should be ttlAccess",
+			)
+		},
+	)
+
+	t.Run(
+		"TTL updates", func(t *testing.T) {
+			manager := newMinCommitTsManager()
+
+			manager.tryUpdate(10, ttlAccess)
+			assert.Equal(t, uint64(10), manager.get(), "Value should be 10")
+
+			manager.tryUpdate(5, ttlAccess)
+			assert.Equal(t, uint64(10), manager.get(), "Value should remain 10")
+		},
+	)
+
+	t.Run(
+		"Elevate write access", func(t *testing.T) {
+			manager := newMinCommitTsManager()
+			manager.tryUpdate(10, ttlAccess)
+
+			currentValue := manager.elevateWriteAccess(twoPCAccess)
+			assert.Equal(t, uint64(10), currentValue, "Current value should be 10")
+			assert.Equal(
+				t,
+				twoPCAccess,
+				manager.getRequiredWriteAccess(),
+				"Required write access should be twoPCAccess",
+			)
+		},
+	)
+
+	t.Run(
+		"Updates after elevation", func(t *testing.T) {
+			manager := newMinCommitTsManager()
+			manager.tryUpdate(10, ttlAccess)
+			manager.elevateWriteAccess(twoPCAccess)
+
+			manager.tryUpdate(20, ttlAccess)
+			assert.Equal(t, uint64(10), manager.get(), "Value should remain 10")
+
+			manager.tryUpdate(30, twoPCAccess)
+			assert.Equal(t, uint64(30), manager.get(), "Value should be 30")
+		},
+	)
+
+	t.Run(
+		"Concurrent updates", func(t *testing.T) {
+			manager := newMinCommitTsManager()
+			done := make(chan bool)
+
+			go func() {
+				for i := 0; i < 1000; i++ {
+					manager.tryUpdate(uint64(i), ttlAccess)
+				}
+				done <- true
+			}()
+
+			go func() {
+				for i := 0; i < 1000; i++ {
+					manager.tryUpdate(uint64(1000+i), ttlAccess)
+				}
+				done <- true
+			}()
+
+			<-done
+			<-done
+
+			assert.Equal(t, manager.get(), uint64(1999))
+		},
+	)
+}
diff --git a/txnkv/transaction/test_probe.go b/txnkv/transaction/test_probe.go
index 598bd70..b782a3a 100644
--- a/txnkv/transaction/test_probe.go
+++ b/txnkv/transaction/test_probe.go
@@ -194,12 +194,12 @@ func (c CommitterProbe) GetCommitTS() uint64 {
 
 // GetMinCommitTS returns the minimal commit ts can be used.
 func (c CommitterProbe) GetMinCommitTS() uint64 {
-	return c.minCommitTS
+	return c.minCommitTSMgr.get()
 }
 
 // SetMinCommitTS sets the minimal commit ts can be used.
 func (c CommitterProbe) SetMinCommitTS(ts uint64) {
-	c.minCommitTS = ts
+	c.minCommitTSMgr.tryUpdate(ts, twoPCAccess)
 }
 
 // SetMaxCommitTS sets the max commit ts can be used.
@@ -381,7 +381,7 @@ func (c CommitterProbe) ResolveFlushedLocks(bo *retry.Backoffer, start, end []by
 
 // SendTxnHeartBeat renews a txn's ttl.
 func SendTxnHeartBeat(bo *retry.Backoffer, store kvstore, primary []byte, startTS, ttl uint64) (newTTL uint64, stopHeartBeat bool, err error) {
-	return sendTxnHeartBeat(bo, store, primary, startTS, ttl)
+	return sendTxnHeartBeat(bo, store, primary, startTS, ttl, 0)
 }
 
 // ConfigProbe exposes configurations and global variables for testing purpose.