Repository: tikv/client-go
Pull request #1119 (2024-01-17 04:07:41)

Support txn file transaction
Txn File transaction is used for transactions of big size (>= 16MiB). By building chunks on external worker and then ingesting to TiKV, txn file transactions will reduce cost and load of TiKV comparing to normal transactions.


Patch:
diff --git a/config/client.go b/config/client.go
index b29c7b9..931172a 100644
--- a/config/client.go
+++ b/config/client.go
@@ -90,6 +90,15 @@ type TiKVClient struct {
 
 	// RemoteCoprocessorAddr is the address of the remote coprocessor.
 	RemoteCoprocessorAddr string `toml:"remote-coprocessor-addr" json:"remote-coprocessor-addr"`
+
+	// TxnChunkWriterAddr is the address of the txn chunk writer for file-based txn.
+	TxnChunkWriterAddr string `toml:"txn-chunk-writer-addr" json:"txn-chunk-writer-addr"`
+	// TxnChunkWriterConcurrency is the concurrency to request the txn chunk writer for file-based txn.
+	TxnChunkWriterConcurrency uint `toml:"txn-chunk-writer-concurrency" json:"txn-chunk-writer-concurrency"`
+	// TxnChunkMaxSize is the maximum size of a txn chunk of file-based txn.
+	TxnChunkMaxSize uint64 `toml:"txn-chunk-max-size" json:"txn-chunk-max-size"`
+	// TxnFileMinMutationSize is the minimum size of mutations to use file-based txn.
+	TxnFileMinMutationSize uint64 `toml:"txn-file-min-mutation-size" json:"txn-file-min-mutation-size"`
 }
 
 // AsyncCommit is the config for the async commit feature. The switch to enable it is a system variable.
@@ -159,6 +168,10 @@ func DefaultTiKVClient() TiKVClient {
 		CoprReqTimeout: 60 * time.Second,
 
 		ResolveLockLiteThreshold: 16,
+
+		TxnChunkWriterConcurrency: 4,
+		TxnChunkMaxSize:           128 * 1024 * 1024,
+		TxnFileMinMutationSize:    16 * 1024 * 1024,
 	}
 }
 
diff --git a/examples/gcworker/go.mod b/examples/gcworker/go.mod
index e4a6bf5..a47f527 100644
--- a/examples/gcworker/go.mod
+++ b/examples/gcworker/go.mod
@@ -20,7 +20,7 @@ require (
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
 	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20231222062942-c0c73f41d0b2 // indirect
+	github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0 // indirect
@@ -29,7 +29,7 @@ require (
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20240318054542-da3b6e381c9b // indirect
+	github.com/tikv/pd/client v0.0.0-20240410074522-b9240a065877 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
diff --git a/examples/rawkv/go.mod b/examples/rawkv/go.mod
index 23c2540..7945319 100644
--- a/examples/rawkv/go.mod
+++ b/examples/rawkv/go.mod
@@ -20,7 +20,7 @@ require (
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
 	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20231222062942-c0c73f41d0b2 // indirect
+	github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0 // indirect
@@ -29,7 +29,7 @@ require (
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20240318054542-da3b6e381c9b // indirect
+	github.com/tikv/pd/client v0.0.0-20240410074522-b9240a065877 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
diff --git a/examples/txnkv/1pc_txn/go.mod b/examples/txnkv/1pc_txn/go.mod
index 73c44a2..3820a60 100644
--- a/examples/txnkv/1pc_txn/go.mod
+++ b/examples/txnkv/1pc_txn/go.mod
@@ -20,7 +20,7 @@ require (
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
 	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20231222062942-c0c73f41d0b2 // indirect
+	github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0 // indirect
@@ -29,7 +29,7 @@ require (
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20240318054542-da3b6e381c9b // indirect
+	github.com/tikv/pd/client v0.0.0-20240410074522-b9240a065877 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
diff --git a/examples/txnkv/async_commit/go.mod b/examples/txnkv/async_commit/go.mod
index af7ce9c..6b53262 100644
--- a/examples/txnkv/async_commit/go.mod
+++ b/examples/txnkv/async_commit/go.mod
@@ -20,7 +20,7 @@ require (
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
 	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20231222062942-c0c73f41d0b2 // indirect
+	github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0 // indirect
@@ -29,7 +29,7 @@ require (
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20240318054542-da3b6e381c9b // indirect
+	github.com/tikv/pd/client v0.0.0-20240410074522-b9240a065877 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
diff --git a/examples/txnkv/delete_range/go.mod b/examples/txnkv/delete_range/go.mod
index df9f9ea..c486ef8 100644
--- a/examples/txnkv/delete_range/go.mod
+++ b/examples/txnkv/delete_range/go.mod
@@ -20,7 +20,7 @@ require (
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
 	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20231222062942-c0c73f41d0b2 // indirect
+	github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0 // indirect
@@ -29,7 +29,7 @@ require (
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20240318054542-da3b6e381c9b // indirect
+	github.com/tikv/pd/client v0.0.0-20240410074522-b9240a065877 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
diff --git a/examples/txnkv/go.mod b/examples/txnkv/go.mod
index 61f072c..6babdd6 100644
--- a/examples/txnkv/go.mod
+++ b/examples/txnkv/go.mod
@@ -20,7 +20,7 @@ require (
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
 	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20231222062942-c0c73f41d0b2 // indirect
+	github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0 // indirect
@@ -29,7 +29,7 @@ require (
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20240318054542-da3b6e381c9b // indirect
+	github.com/tikv/pd/client v0.0.0-20240410074522-b9240a065877 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
diff --git a/examples/txnkv/pessimistic_txn/go.mod b/examples/txnkv/pessimistic_txn/go.mod
index 1977104..41632b9 100644
--- a/examples/txnkv/pessimistic_txn/go.mod
+++ b/examples/txnkv/pessimistic_txn/go.mod
@@ -20,7 +20,7 @@ require (
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
 	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20231222062942-c0c73f41d0b2 // indirect
+	github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0 // indirect
@@ -29,7 +29,7 @@ require (
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20240318054542-da3b6e381c9b // indirect
+	github.com/tikv/pd/client v0.0.0-20240410074522-b9240a065877 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
diff --git a/examples/txnkv/unsafedestoryrange/go.mod b/examples/txnkv/unsafedestoryrange/go.mod
index 66a46a1..d704071 100644
--- a/examples/txnkv/unsafedestoryrange/go.mod
+++ b/examples/txnkv/unsafedestoryrange/go.mod
@@ -20,7 +20,7 @@ require (
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
 	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20231222062942-c0c73f41d0b2 // indirect
+	github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0 // indirect
@@ -29,7 +29,7 @@ require (
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20240318054542-da3b6e381c9b // indirect
+	github.com/tikv/pd/client v0.0.0-20240410074522-b9240a065877 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
diff --git a/go.mod b/go.mod
index 887d038..29cb803 100644
--- a/go.mod
+++ b/go.mod
@@ -14,7 +14,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-20231222062942-c0c73f41d0b2
+	github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9
 	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 546cc68..21804c4 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-20231222062942-c0c73f41d0b2 h1:364A6VCS+l0oHBKZKotX9LzmfEtIO/NTccTIQcPp3Ug=
-github.com/pingcap/kvproto v0.0.0-20231222062942-c0c73f41d0b2/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8=
+github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9 h1:tAfzWZi+hVqX8tM+jnltSiC2k3X+305L/aGiblhX6dk=
+github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9/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=
@@ -111,10 +111,6 @@ 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-20240318054542-da3b6e381c9b h1:BiFQvT54X/UMNjqO/UXqp7IWl9bvIaOfNaraqdpo++g=
-github.com/tikv/pd/client v0.0.0-20240318054542-da3b6e381c9b/go.mod h1:Z/QAgOt29zvwBTd0H6pdx45VO6KRNc/O/DzGkVmSyZg=
-github.com/tikv/pd/client v0.0.0-20240407053619-a173a08acd81 h1:MCfYSh3oZIkHjXf+H65LI1abtdnRh6zdNYPwYDSodms=
-github.com/tikv/pd/client v0.0.0-20240407053619-a173a08acd81/go.mod h1:Qdzax77EzM8h6NbzwKymp96AB0DUvtkHCcC1jwxoiGU=
 github.com/tikv/pd/client v0.0.0-20240410074522-b9240a065877 h1:8Poo/LLJo9sMO2Yak5+0yWEiN04NenJvvBGBafSERGg=
 github.com/tikv/pd/client v0.0.0-20240410074522-b9240a065877/go.mod h1:Qdzax77EzM8h6NbzwKymp96AB0DUvtkHCcC1jwxoiGU=
 github.com/twmb/murmur3 v1.1.3 h1:D83U0XYKcHRYwYIpBKf3Pks91Z0Byda/9SJ8B6EMRcA=
diff --git a/kv/variables.go b/kv/variables.go
index cae78c9..495e198 100644
--- a/kv/variables.go
+++ b/kv/variables.go
@@ -49,6 +49,9 @@ type Variables struct {
 	// When its value is 0, it's not killed
 	// When its value is not 0, it's killed, the value indicates concrete reason.
 	Killed *uint32
+
+	// EnableTxnFile specifies whether file-based txn is enabled.
+	EnableTxnFile bool
 }
 
 // NewVariables create a new Variables instance with default values.
@@ -57,6 +60,7 @@ func NewVariables(killed *uint32) *Variables {
 		BackoffLockFast: DefBackoffLockFast,
 		BackOffWeight:   DefBackOffWeight,
 		Killed:          killed,
+		EnableTxnFile:   false,
 	}
 }
 
diff --git a/txnkv/transaction/2pc.go b/txnkv/transaction/2pc.go
index 7d5ec2e..a697123 100644
--- a/txnkv/transaction/2pc.go
+++ b/txnkv/transaction/2pc.go
@@ -195,6 +195,8 @@ type twoPhaseCommitter struct {
 	isInternal bool
 
 	forUpdateTSConstraints map[string]uint64
+
+	txnFileCtx txnFileCtx
 }
 
 type memBufferMutations struct {
@@ -734,6 +736,9 @@ func (c *twoPhaseCommitter) primary() []byte {
 		if c.mutations != nil {
 			return c.mutations.GetKey(0)
 		}
+		if c.txnFileCtx.slice.len() > 0 {
+			return c.txnFileCtx.slice.chunkRanges[0].smallest
+		}
 		return nil
 	}
 	return c.primaryKey
@@ -1215,7 +1220,7 @@ func keepAlive(c *twoPhaseCommitter, closeCh chan struct{}, primaryKey []byte, l
 			logutil.Logger(bo.GetCtx()).Info("send TxnHeartBeat",
 				zap.Uint64("startTS", c.startTS), zap.Uint64("newTTL", newTTL))
 			startTime := time.Now()
-			_, stopHeartBeat, err := sendTxnHeartBeat(bo, c.store, primaryKey, c.startTS, newTTL)
+			_, stopHeartBeat, err := sendTxnHeartBeat(bo, c.store, primaryKey, c.startTS, newTTL, c.txnFileCtx.slice.len() > 0)
 			if err != nil {
 				keepFail++
 				metrics.TxnHeartBeatHistogramError.Observe(time.Since(startTime).Seconds())
@@ -1237,11 +1242,12 @@ func keepAlive(c *twoPhaseCommitter, closeCh chan struct{}, primaryKey []byte, l
 	}
 }
 
-func sendTxnHeartBeat(bo *retry.Backoffer, store kvstore, primary []byte, startTS, ttl uint64) (newTTL uint64, stopHeartBeat bool, err error) {
+func sendTxnHeartBeat(bo *retry.Backoffer, store kvstore, primary []byte, startTS, ttl uint64, isTxnFile bool) (newTTL uint64, stopHeartBeat bool, err error) {
 	req := tikvrpc.NewRequest(tikvrpc.CmdTxnHeartBeat, &kvrpcpb.TxnHeartBeatRequest{
 		PrimaryLock:   primary,
 		StartVersion:  startTS,
 		AdviseLockTtl: ttl,
+		IsTxnFile:     isTxnFile,
 	})
 	for {
 		loc, err := store.GetRegionCache().LocateKey(bo, primary)
diff --git a/txnkv/transaction/txn.go b/txnkv/transaction/txn.go
index 94ef359..04c7859 100644
--- a/txnkv/transaction/txn.go
+++ b/txnkv/transaction/txn.go
@@ -502,6 +502,14 @@ func (txn *KVTxn) Commit(ctx context.Context) error {
 			}
 		}
 	}()
+	if committer.useTxnFile() {
+		err = committer.executeTxnFile(ctx)
+		// TODO: fall back to normal 2PC when tikv-worker is unavailable.
+		if val == nil || sessionID > 0 {
+			txn.onCommitted(err)
+		}
+		return err
+	}
 	// latches disabled
 	// pessimistic transaction should also bypass latch.
 	if txn.store.TxnLatches() == nil || txn.IsPessimistic() {
diff --git a/txnkv/transaction/txn_file.go b/txnkv/transaction/txn_file.go
new file mode 100644
index 0000000..775191e
--- /dev/null
+++ b/txnkv/transaction/txn_file.go
@@ -0,0 +1,936 @@
+// 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 transaction
+
+import (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"encoding/json"
+	"fmt"
+	"hash/crc32"
+	"io"
+	"net/http"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/pingcap/kvproto/pkg/errorpb"
+	"github.com/pingcap/kvproto/pkg/kvrpcpb"
+	"github.com/pkg/errors"
+	"github.com/tikv/client-go/v2/config"
+	tikverr "github.com/tikv/client-go/v2/error"
+	"github.com/tikv/client-go/v2/internal/client"
+	"github.com/tikv/client-go/v2/internal/locate"
+	"github.com/tikv/client-go/v2/internal/logutil"
+	"github.com/tikv/client-go/v2/internal/retry"
+	"github.com/tikv/client-go/v2/kv"
+	"github.com/tikv/client-go/v2/metrics"
+	"github.com/tikv/client-go/v2/tikvrpc"
+	"github.com/tikv/client-go/v2/txnkv/txnlock"
+	"github.com/tikv/client-go/v2/util"
+	atomicutil "go.uber.org/atomic"
+	"go.uber.org/zap"
+)
+
+var (
+	// BuildTxnFileMaxBackoff is max sleep time (in millisecond) to build TxnFile.
+	BuildTxnFileMaxBackoff = atomicutil.NewUint64(60000)
+
+	buildChunkErrMsg string = "txn file: build chunk failed"
+)
+
+const PreSplitRegionChunks = 4
+
+type txnFileCtx struct {
+	slice txnChunkSlice
+}
+
+type chunkBatch struct {
+	txnChunkSlice
+	region    *locate.KeyLocation
+	isPrimary bool
+}
+
+func (b chunkBatch) String() string {
+	return fmt.Sprintf("chunkBatch{region: %s, isPrimary: %t, txnChunkSlice: %s}", b.region.String(), b.isPrimary, b.txnChunkSlice.String())
+}
+
+// chunkBatch.txnChunkSlice should be sorted by smallest.
+func (b *chunkBatch) getSampleKeys() [][]byte {
+	keys := make([][]byte, 0)
+	start := sort.Search(b.txnChunkSlice.len(), func(i int) bool {
+		return bytes.Compare(b.region.StartKey, b.txnChunkSlice.chunkRanges[i].smallest) <= 0
+	})
+	end := b.txnChunkSlice.len()
+	if len(b.region.EndKey) > 0 {
+		end = sort.Search(b.txnChunkSlice.len(), func(i int) bool {
+			return bytes.Compare(b.txnChunkSlice.chunkRanges[i].smallest, b.region.EndKey) >= 0
+		})
+	}
+	for i := start; i < end; i++ {
+		keys = append(keys, b.txnChunkSlice.chunkRanges[i].smallest)
+	}
+	return keys
+}
+
+// txnChunkSlice should be sorted by txnChunkRange.smallest and no overlapping.
+type txnChunkSlice struct {
+	chunkIDs    []uint64
+	chunkRanges []txnChunkRange
+}
+
+func (s txnChunkSlice) String() string {
+	slice := make([]string, len(s.chunkRanges))
+	for i, ran := range s.chunkRanges {
+		slice[i] = fmt.Sprintf("txnChunkSlice{%v: [%s, %s]}", s.chunkIDs[i], kv.StrKey(ran.smallest), kv.StrKey(ran.biggest))
+	}
+	return fmt.Sprintf("[%s]", strings.Join(slice, ", "))
+}
+
+func (s *txnChunkSlice) Smallest() []byte {
+	if len(s.chunkRanges) == 0 {
+		return nil
+	}
+	return s.chunkRanges[0].smallest
+}
+
+func (s *txnChunkSlice) Biggest() []byte {
+	if len(s.chunkRanges) == 0 {
+		return nil
+	}
+	return s.chunkRanges[len(s.chunkRanges)-1].biggest
+}
+
+func (cs *txnChunkSlice) appendSlice(other *txnChunkSlice) {
+	if cs.len() == 0 {
+		cs.chunkIDs = append(cs.chunkIDs, other.chunkIDs...)
+		cs.chunkRanges = append(cs.chunkRanges, other.chunkRanges...)
+		return
+	}
+	lastChunkRange := cs.chunkRanges[len(cs.chunkIDs)-1]
+	for i, chunkRange := range other.chunkRanges {
+		if bytes.Compare(lastChunkRange.smallest, chunkRange.smallest) < 0 {
+			cs.chunkIDs = append(cs.chunkIDs, other.chunkIDs[i:]...)
+			cs.chunkRanges = append(cs.chunkRanges, other.chunkRanges[i:]...)
+			break
+		}
+	}
+}
+
+func (cs *txnChunkSlice) append(chunkID uint64, chunkRange txnChunkRange) {
+	cs.chunkIDs = append(cs.chunkIDs, chunkID)
+	cs.chunkRanges = append(cs.chunkRanges, chunkRange)
+}
+
+func (cs *txnChunkSlice) len() int {
+	return len(cs.chunkIDs)
+}
+
+// []chunkBatch is sorted by region.StartKey.
+// Note: regions may be overlapping.
+func (cs *txnChunkSlice) groupToBatches(c *locate.RegionCache, bo *retry.Backoffer) ([]chunkBatch, error) {
+	batchMap := make(map[locate.RegionVerID]*chunkBatch)
+	for i, chunkRange := range cs.chunkRanges {
+		regions, err := chunkRange.getOverlapRegions(c, bo)
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		for _, r := range regions {
+			if batchMap[r.Region] == nil {
+				batchMap[r.Region] = &chunkBatch{
+					region: r,
+				}
+			}
+			batchMap[r.Region].append(cs.chunkIDs[i], chunkRange)
+		}
+	}
+
+	batches := make([]chunkBatch, 0, len(batchMap))
+	for _, batch := range batchMap {
+		batches = append(batches, *batch)
+	}
+	sort.Slice(batches, func(i, j int) bool {
+		return bytes.Compare(batches[i].region.StartKey, batches[j].region.StartKey) < 0
+	})
+
+	logutil.Logger(bo.GetCtx()).Debug("txn file group to batches", zap.Stringers("batches", batches))
+	return batches, nil
+}
+
+type txnChunkRange struct {
+	smallest []byte
+	biggest  []byte
+}
+
+func (r txnChunkRange) String() string {
+	return fmt.Sprintf("txnChunkRange[%s,%s]", kv.StrKey(r.smallest), kv.StrKey(r.biggest))
+}
+
+func newTxnChunkRange(smallest []byte, biggest []byte) txnChunkRange {
+	return txnChunkRange{
+		smallest: smallest,
+		biggest:  biggest,
+	}
+}
+
+func (r *txnChunkRange) getOverlapRegions(c *locate.RegionCache, bo *retry.Backoffer) ([]*locate.KeyLocation, error) {
+	regions := make([]*locate.KeyLocation, 0)
+	startKey := r.smallest
+	for bytes.Compare(startKey, r.biggest) <= 0 {
+		loc, err := c.LocateKey(bo, startKey)
+		if err != nil {
+			logutil.Logger(bo.GetCtx()).Error("locate key failed", zap.Error(err), zap.String("startKey", kv.StrKey(startKey)))
+			return nil, errors.Wrap(err, "locate key failed")
+		}
+		regions = append(regions, loc)
+		if len(loc.EndKey) == 0 {
+			break
+		}
+		startKey = loc.EndKey
+	}
+	return regions, nil
+}
+
+type txnFileAction interface {
+	executeBatch(c *twoPhaseCommitter, bo *retry.Backoffer, batch chunkBatch) (*tikvrpc.Response, error)
+	onPrimarySuccess(c *twoPhaseCommitter)
+	extractKeyError(resp *tikvrpc.Response) *kvrpcpb.KeyError
+	asyncExecuteSecondaries() bool
+	String() string
+}
+
+type txnFilePrewriteAction struct{}
+
+var _ txnFileAction = (*txnFilePrewriteAction)(nil)
+
+func (a txnFilePrewriteAction) executeBatch(c *twoPhaseCommitter, bo *retry.Backoffer, batch chunkBatch) (*tikvrpc.Response, error) {
+	primaryLock := c.txnFileCtx.slice.chunkRanges[0].smallest
+	req := tikvrpc.NewRequest(tikvrpc.CmdPrewrite, &kvrpcpb.PrewriteRequest{
+		StartVersion:   c.startTS,
+		PrimaryLock:    primaryLock,
+		LockTtl:        c.lockTTL,
+		MaxCommitTs:    c.maxCommitTS,
+		AssertionLevel: kvrpcpb.AssertionLevel_Off,
+		TxnFileChunks:  batch.chunkIDs,
+	}, kvrpcpb.Context{
+		Priority:               c.priority,
+		SyncLog:                c.syncLog,
+		ResourceGroupTag:       c.resourceGroupTag,
+		DiskFullOpt:            c.diskFullOpt,
+		TxnSource:              c.txnSource,
+		MaxExecutionDurationMs: uint64(client.ReadTimeoutMedium.Milliseconds()),
+		RequestSource:          c.txn.GetRequestSource(),
+		ResourceControlContext: &kvrpcpb.ResourceControlContext{
+			ResourceGroupName: c.resourceGroupName,
+		},
+	})
+	sender := locate.NewRegionRequestSender(c.store.GetRegionCache(), c.store.GetTiKVClient())
+	var resolvingRecordToken *int
+	for {
+		resp, _, err := sender.SendReq(bo, req, batch.region.Region, client.ReadTimeoutMedium)
+		if err != nil {
+			return nil, err
+		}
+		if resp.Resp == nil {
+			return nil, errors.WithStack(tikverr.ErrBodyMissing)
+		}
+		prewriteResp := resp.Resp.(*kvrpcpb.PrewriteResponse)
+		keyErrs := prewriteResp.GetErrors()
+		if len(keyErrs) == 0 {
+			return resp, nil
+		}
+		var locks []*txnlock.Lock
+		for _, keyErr := range keyErrs {
+			// Check already exists error
+			if alreadyExist := keyErr.GetAlreadyExist(); alreadyExist != nil {
+				e := &tikverr.ErrKeyExist{AlreadyExist: alreadyExist}
+				return nil, c.extractKeyExistsErr(e)
+			}
+
+			// Extract lock from key error
+			lock, err1 := txnlock.ExtractLockFromKeyErr(keyErr)
+			if err1 != nil {
+				return nil, err1
+			}
+			logutil.Logger(bo.GetCtx()).Info(
+				"prewrite txn file encounters lock",
+				zap.Uint64("session", c.sessionID),
+				zap.Uint64("txnID", c.startTS),
+				zap.Stringer("lock", lock),
+			)
+			// If an optimistic transaction encounters a lock with larger TS, this transaction will certainly
+			// fail due to a WriteConflict error. So we can construct and return an error here early.
+			// Pessimistic transactions don't need such an optimization. If this key needs a pessimistic lock,
+			// TiKV will return a PessimisticLockNotFound error directly if it encounters a different lock. Otherwise,
+			// TiKV returns lock.TTL = 0, and we still need to resolve the lock.
+			if lock.TxnID > c.startTS {
+				return nil, tikverr.NewErrWriteConflictWithArgs(
+					c.startTS,
+					lock.TxnID,
+					0,
+					lock.Key,
+					kvrpcpb.WriteConflict_Optimistic,
+				)
+			}
+			locks = append(locks, lock)
+		}
+		if resolvingRecordToken == nil {
+			token := c.store.GetLockResolver().RecordResolvingLocks(locks, c.startTS)
+			resolvingRecordToken = &token
+			defer c.store.GetLockResolver().ResolveLocksDone(c.startTS, *resolvingRecordToken)
+		} else {
+			c.store.GetLockResolver().UpdateResolvingLocks(locks, c.startTS, *resolvingRecordToken)
+		}
+		resolveLockOpts := txnlock.ResolveLocksOptions{
+			CallerStartTS: c.startTS,
+			Locks:         locks,
+			Detail:        &c.getDetail().ResolveLock,
+		}
+		resolveLockRes, err := c.store.GetLockResolver().ResolveLocksWithOpts(bo, resolveLockOpts)
+		if err != nil {
+			return nil, err
+		}
+		msBeforeExpired := resolveLockRes.TTL
+		if msBeforeExpired > 0 {
+			err = bo.BackoffWithCfgAndMaxSleep(
+				retry.BoTxnLock,
+				int(msBeforeExpired),
+				errors.Errorf("2PC txn file prewrite lockedKeys: %d", len(locks)),
+			)
+			if err != nil {
+				return nil, err
+			}
+		}
+	}
+}
+
+func (a txnFilePrewriteAction) onPrimarySuccess(c *twoPhaseCommitter) {
+	c.run(c, nil)
+}
+
+func (a txnFilePrewriteAction) extractKeyError(resp *tikvrpc.Response) *kvrpcpb.KeyError {
+	prewriteResp, _ := resp.Resp.(*kvrpcpb.PrewriteResponse)
+	errs := prewriteResp.GetErrors()
+	if len(errs) > 0 {
+		return errs[0]
+	}
+	return nil
+}
+
+func (a txnFilePrewriteAction) asyncExecuteSecondaries() bool {
+	return false
+}
+
+func (a txnFilePrewriteAction) String() string {
+	return "txnFilePrewrite"
+}
+
+type txnFileCommitAction struct {
+	commitTS uint64
+}
+
+var _ txnFileAction = (*txnFileCommitAction)(nil)
+
+func (a txnFileCommitAction) executeBatch(c *twoPhaseCommitter, bo *retry.Backoffer, batch chunkBatch) (*tikvrpc.Response, error) {
+	req := tikvrpc.NewRequest(tikvrpc.CmdCommit, &kvrpcpb.CommitRequest{
+		Keys:          batch.getSampleKeys(), // To help detect duplicated request.
+		StartVersion:  c.startTS,
+		CommitVersion: a.commitTS,
+		IsTxnFile:     true,
+	}, kvrpcpb.Context{
+		Priority:               c.priority,
+		SyncLog:                c.syncLog,
+		ResourceGroupTag:       c.resourceGroupTag,
+		DiskFullOpt:            c.diskFullOpt,
+		TxnSource:              c.txnSource,
+		MaxExecutionDurationMs: uint64(client.ReadTimeoutMedium.Milliseconds()),
+		RequestSource:          c.txn.GetRequestSource(),
+		ResourceControlContext: &kvrpcpb.ResourceControlContext{
+			ResourceGroupName: c.resourceGroupName,
+		},
+	})
+	sender := locate.NewRegionRequestSender(c.store.GetRegionCache(), c.store.GetTiKVClient())
+	for {
+		resp, _, err := sender.SendReq(bo, req, batch.region.Region, client.ReadTimeoutMedium)
+		if batch.isPrimary && sender.GetRPCError() != nil {
+			c.setUndeterminedErr(errors.WithStack(sender.GetRPCError()))
+		}
+		// Unexpected error occurs, return it.
+		if err != nil {
+			return nil, err
+		}
+		if resp.Resp == nil {
+			return nil, errors.WithStack(tikverr.ErrBodyMissing)
+		}
+		commitResp := resp.Resp.(*kvrpcpb.CommitResponse)
+		if keyErr := commitResp.GetError(); keyErr != nil {
+			if rejected := keyErr.GetCommitTsExpired(); rejected != nil {
+				logutil.Logger(bo.GetCtx()).Info("2PC commitTS rejected by TiKV, retry with a newer commitTS",
+					zap.Uint64("txnStartTS", c.startTS),
+					zap.Stringer("info", logutil.Hex(rejected)))
+
+				// Do not retry for a txn which has a too large MinCommitTs
+				// 3600000 << 18 = 943718400000
+				if rejected.MinCommitTs-rejected.AttemptedCommitTs > 943718400000 {
+					return nil, errors.Errorf("2PC MinCommitTS is too large, we got MinCommitTS: %d, and AttemptedCommitTS: %d",
+						rejected.MinCommitTs, rejected.AttemptedCommitTs)
+				}
+
+				// Update commit ts and retry.
+				commitTS, err1 := c.store.GetTimestampWithRetry(bo, c.txn.GetScope())
+				if err1 != nil {
+					logutil.Logger(bo.GetCtx()).Warn("2PC get commitTS failed",
+						zap.Error(err1),
+						zap.Uint64("txnStartTS", c.startTS))
+					return nil, err1
+				}
+
+				c.mu.Lock()
+				c.commitTS = commitTS
+				c.mu.Unlock()
+				// Update the commitTS of the request and retry.
+				req.Commit().CommitVersion = commitTS
+				continue
+			}
+			return nil, tikverr.ExtractKeyErr(keyErr)
+		}
+		return resp, nil
+	}
+}
+
+func (a txnFileCommitAction) onPrimarySuccess(c *twoPhaseCommitter) {
+	c.mu.Lock()
+	c.mu.committed = true
+	c.mu.Unlock()
+}
+
+func (a txnFileCommitAction) extractKeyError(resp *tikvrpc.Response) *kvrpcpb.KeyError {
+	commitResp, _ := resp.Resp.(*kvrpcpb.CommitResponse)
+	return commitResp.GetError()
+}
+
+func (a txnFileCommitAction) asyncExecuteSecondaries() bool {
+	return true
+}
+
+func (a txnFileCommitAction) String() string {
+	return "txnFileCommit"
+}
+
+type txnFileRollbackAction struct{}
+
+var _ txnFileAction = (*txnFileRollbackAction)(nil)
+
+func (a txnFileRollbackAction) executeBatch(c *twoPhaseCommitter, bo *retry.Backoffer, batch chunkBatch) (*tikvrpc.Response, error) {
+	req := tikvrpc.NewRequest(tikvrpc.CmdBatchRollback, &kvrpcpb.BatchRollbackRequest{
+		Keys:         batch.getSampleKeys(), // To help detect duplicated request.
+		StartVersion: c.startTS,
+		IsTxnFile:    true,
+	}, kvrpcpb.Context{
+		Priority:               c.priority,
+		SyncLog:                c.syncLog,
+		ResourceGroupTag:       c.resourceGroupTag,
+		DiskFullOpt:            c.diskFullOpt,
+		TxnSource:              c.txnSource,
+		MaxExecutionDurationMs: uint64(client.ReadTimeoutShort.Milliseconds()),
+		RequestSource:          c.txn.GetRequestSource(),
+		ResourceControlContext: &kvrpcpb.ResourceControlContext{
+			ResourceGroupName: c.resourceGroupName,
+		},
+	})
+	sender := locate.NewRegionRequestSender(c.store.GetRegionCache(), c.store.GetTiKVClient())
+	resp, _, err1 := sender.SendReq(bo, req, batch.region.Region, client.ReadTimeoutShort)
+	if err1 != nil {
+		return nil, err1
+	}
+	return resp, nil
+}
+
+func (a txnFileRollbackAction) onPrimarySuccess(_ *twoPhaseCommitter) {
+}
+
+func (a txnFileRollbackAction) extractKeyError(resp *tikvrpc.Response) *kvrpcpb.KeyError {
+	rollbackResp, _ := resp.Resp.(*kvrpcpb.BatchRollbackResponse)
+	return rollbackResp.GetError()
+}
+
+func (a txnFileRollbackAction) asyncExecuteSecondaries() bool {
+	return true
+}
+
+func (a txnFileRollbackAction) String() string {
+	return "txnFileRollback"
+}
+
+type step struct {
+	name string
+	dur  time.Duration
+}
+
+func (s step) String() string {
+	return fmt.Sprintf("%s:%s", s.name, s.dur.String())
+}
+
+func (c *twoPhaseCommitter) executeTxnFile(ctx context.Context) (err error) {
+	start := time.Now()
+	steps := make([]step, 0)
+	stepDone := func(name string) {
+		now := time.Now()
+		s := step{name: name, dur: now.Sub(start)}
+		steps = append(steps, s)
+		start = now
+	}
+
+	defer func() {
+		// Always clean up all written keys if the txn does not commit.
+		c.mu.RLock()
+		committed := c.mu.committed
+		undetermined := c.mu.undeterminedErr != nil
+		c.mu.RUnlock()
+		if !committed && !undetermined {
+			if c.txnFileCtx.slice.len() > 0 {
+				err1 := c.executeTxnFileAction(retry.NewBackofferWithVars(ctx, int(CommitMaxBackoff), c.txn.vars), c.txnFileCtx.slice, txnFileRollbackAction{})
+				if err1 != nil {
+					logutil.Logger(ctx).Error("txn file: rollback on error failed", zap.Error(err1))
+				}
+			}
+			metrics.TwoPCTxnCounterError.Inc()
+		} else {
+			metrics.TwoPCTxnCounterOk.Inc()
+		}
+		c.txn.commitTS = c.commitTS
+
+		logutil.Logger(ctx).Info("execute txn file finished",
+			zap.Uint64("startTS", c.startTS),
+			zap.Uint64("commitTS", c.commitTS),
+			zap.Error(err),
+			zap.Stringers("steps", steps))
+	}()
+
+	logutil.Logger(ctx).Debug("execute txn file", zap.Uint64("startTS", c.startTS))
+
+	buildBo := retry.NewBackofferWithVars(ctx, int(BuildTxnFileMaxBackoff.Load()), c.txn.vars)
+	err = c.buildTxnFiles(buildBo, c.mutations)
+	stepDone("build")
+	if err != nil {
+		return
+	}
+
+	err = c.preSplitTxnFileRegions(buildBo)
+	stepDone("pre-split")
+	if err != nil {
+		return
+	}
+
+	prewriteBo := retry.NewBackofferWithVars(ctx, int(PrewriteMaxBackoff.Load()), c.txn.vars)
+	err = c.executeTxnFileAction(prewriteBo, c.txnFileCtx.slice, txnFilePrewriteAction{})
+	stepDone("prewrite")
+	if err != nil {
+		return
+	}
+
+	commitBo := retry.NewBackofferWithVars(ctx, int(CommitMaxBackoff), c.txn.vars)
+	c.commitTS, err = c.store.GetTimestampWithRetry(commitBo, c.txn.GetScope())
+	if err != nil {
+		return
+	}
+	err = c.executeTxnFileAction(commitBo, c.txnFileCtx.slice, txnFileCommitAction{commitTS: c.commitTS})
+	stepDone("commit")
+	return
+}
+
+func (c *twoPhaseCommitter) executeTxnFileSlice(bo *retry.Backoffer, chunkSlice txnChunkSlice, batches []chunkBatch, action txnFileAction, successRanges *util.MergeRanges) (txnChunkSlice, error) {
+	var err error
+	var regionErrChunks txnChunkSlice
+
+	if batches == nil {
+		batches, err = chunkSlice.groupToBatches(c.store.GetRegionCache(), bo)
+		if err != nil {
+			return regionErrChunks, errors.Wrap(err, "txn file: group to batches failed")
+		}
+	}
+
+	for _, batch := range batches {
+		if successRanges.Covered(batch.region.StartKey, batch.region.EndKey) {
+			continue
+		}
+
+		resp, err1 := action.executeBatch(c, bo, batch)
+		logutil.Logger(bo.GetCtx()).Debug("txn file: execute batch finished",
+			zap.Uint64("startTS", c.startTS),
+			zap.Any("batch", batch),
+			zap.Stringer("action", action),
+			zap.Error(err1))
+		if err1 != nil {
+			return regionErrChunks, err1
+		}
+		if keyErr := action.extractKeyError(resp); keyErr != nil {
+			if alreadyExist := keyErr.GetAlreadyExist(); alreadyExist != nil {
+				e := &tikverr.ErrKeyExist{AlreadyExist: alreadyExist}
+				return regionErrChunks, c.extractKeyExistsErr(e)
+			}
+			lock, err2 := txnlock.ExtractLockFromKeyErr(keyErr)
+			if err2 != nil {
+				return regionErrChunks, err2
+			}
+			if lock.TxnID > c.startTS {
+				return regionErrChunks, tikverr.NewErrWriteConflictWithArgs(
+					c.startTS,
+					lock.TxnID,
+					0,
+					lock.Key,
+					kvrpcpb.WriteConflict_Optimistic,
+				)
+			}
+		}
+		regionErr, err1 := resp.GetRegionError()
+		if err1 != nil {
+			return regionErrChunks, err1
+		}
+		if regionErr != nil {
+			regionErrChunks.appendSlice(&batch.txnChunkSlice)
+			continue
+		}
+
+		successRanges.Insert(batch.region.StartKey, batch.region.EndKey)
+	}
+	return regionErrChunks, nil
+}
+
+func (c *twoPhaseCommitter) executeTxnFileSliceWithRetry(bo *retry.Backoffer, chunkSlice txnChunkSlice, batches []chunkBatch, action txnFileAction, successRanges *util.MergeRanges) error {
+	currentChunks := chunkSlice
+	currentBatches := batches
+	for {
+		var regionErrChunks txnChunkSlice
+		regionErrChunks, err := c.executeTxnFileSlice(bo, currentChunks, currentBatches, action, successRanges)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+		if regionErrChunks.len() == 0 {
+			return nil
+		}
+		currentChunks = regionErrChunks
+		currentBatches = nil
+		err = bo.Backoff(retry.BoRegionMiss, errors.Errorf("txn file: execute failed, region miss"))
+		if err != nil {
+			return errors.WithStack(err)
+		}
+	}
+}
+
+func (c *twoPhaseCommitter) executeTxnFilePrimaryBatch(bo *retry.Backoffer, firstBatch chunkBatch, action txnFileAction) (regionErr *errorpb.Error, err error) {
+	if !firstBatch.region.Contains(c.primary()) {
+		logutil.Logger(bo.GetCtx()).Error("txn file: primary out of first batch",
+			zap.Uint64("startTS", c.startTS),
+			zap.String("primary", kv.StrKey(c.primary())),
+			zap.Stringer("action", action),
+			zap.Stringer("first batch", firstBatch))
+		return nil, fmt.Errorf("txn file: primary out of first batch")
+	}
+
+	firstBatch.isPrimary = true
+	resp, err := action.executeBatch(c, bo, firstBatch)
+	logutil.Logger(bo.GetCtx()).Debug("txn file: execute primary batch finished",
+		zap.Uint64("startTS", c.startTS),
+		zap.String("primary", kv.StrKey(c.primary())),
+		zap.Stringer("action", action),
+		zap.Stringer("batch", firstBatch),
+		zap.Error(err))
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	regionErr, err = resp.GetRegionError()
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	if regionErr != nil {
+		return regionErr, nil
+	}
+	action.onPrimarySuccess(c)
+	return nil, nil
+}
+
+func (c *twoPhaseCommitter) executeTxnFileAction(bo *retry.Backoffer, chunkSlice txnChunkSlice, action txnFileAction) error {
+	for {
+		batches, err := chunkSlice.groupToBatches(c.store.GetRegionCache(), bo)
+		if err != nil {
+			return errors.Wrap(err, "txn file: group to batches failed")
+		}
+
+		regionErr, err := c.executeTxnFilePrimaryBatch(bo, batches[0], action)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+		if regionErr != nil {
+			errBo := bo.Backoff(retry.BoRegionMiss, errors.Wrap(errors.New(regionErr.String()), "txn file: execute primary batch failed"))
+			if errBo != nil {
+				return errors.WithStack(errBo)
+			}
+			continue
+		}
+
+		secondaries := batches[1:]
+		if len(secondaries) == 0 {
+			return nil
+		}
+		primaryRegion := batches[0].region
+		successRanges := util.NewMergeRanges()
+		successRanges.Insert(primaryRegion.StartKey, primaryRegion.EndKey)
+		var emptySlice txnChunkSlice
+		if !action.asyncExecuteSecondaries() {
+			return c.executeTxnFileSliceWithRetry(bo, emptySlice, secondaries, action, successRanges)
+		}
+
+		c.store.WaitGroup().Add(1)
+		errGo := c.store.Go(func() {
+			defer c.store.WaitGroup().Done()
+			err := c.executeTxnFileSliceWithRetry(bo, emptySlice, secondaries, action, successRanges)
+			logutil.Logger(bo.GetCtx()).Debug("txn file: async execute secondaries finished",
+				zap.Uint64("startTS", c.startTS),
+				zap.Stringer("action", action),
+				zap.Error(err))
+			if err != nil {
+				logutil.Logger(bo.GetCtx()).Warn("txn file: async execute secondaries failed",
+					zap.Uint64("startTS", c.startTS), zap.Stringer("action", action), zap.Error(err))
+			}
+		})
+		if errGo != nil {
+			c.store.WaitGroup().Done()
+			logutil.Logger(bo.GetCtx()).Warn("txn file: create goroutine failed",
+				zap.Uint64("startTS", c.startTS),
+				zap.Stringer("action", action),
+				zap.Error(errGo))
+		}
+		return nil
+	}
+}
+
+func (c *twoPhaseCommitter) buildTxnFiles(bo *retry.Backoffer, mutations CommitterMutations) error {
+	bo, cancel := bo.Fork()
+	defer cancel()
+
+	cfg := config.GetGlobalConfig()
+	maxTxnChunkSize := int(cfg.TiKVClient.TxnChunkMaxSize)
+	capacity := c.txn.Size() + c.txn.Len()*7 + 4
+	totalChunks := (capacity + maxTxnChunkSize - 1) / maxTxnChunkSize
+	if capacity > maxTxnChunkSize {
+		capacity = maxTxnChunkSize
+	}
+	concurrency := int(cfg.TiKVClient.TxnChunkWriterConcurrency)
+	if concurrency > totalChunks {
+		concurrency = totalChunks
+	}
+
+	writer, err := newChunkWriterClient()
+	if err != nil {
+		return errors.Wrap(err, "new chunk writer client failed")
+	}
+
+	results := make([]buildChunkResult, 0, totalChunks)
+	resultCh := make(chan buildChunkResult, concurrency)
+
+	totalSize := 0
+	inflightChunks := 0
+	buf := make([]byte, 0, capacity)
+	chunkSmallest := mutations.GetKey(0)
+	for i := 0; i < mutations.Len(); i++ {
+		key := mutations.GetKey(i)
+		op := mutations.GetOp(i)
+		val := mutations.GetValue(i)
+		entrySize := 2 + len(key) + 1 + 4 + len(val)
+		if len(buf) > 0 && len(buf)+entrySize+4 > cap(buf) {
+			totalSize += len(buf)
+			inflightChunks += 1
+			ran := newTxnChunkRange(chunkSmallest, mutations.GetKey(i-1))
+			writer.asyncBuildChunk(bo, buf, ran, resultCh)
+			chunkSmallest = key
+			buf = make([]byte, 0, capacity)
+
+			if inflightChunks >= concurrency {
+				r := <-resultCh
+				if r.err != nil {
+					logutil.Logger(bo.GetCtx()).Error(buildChunkErrMsg, zap.Error(err))
+					return errors.Wrap(r.err, buildChunkErrMsg)
+				}
+				results = append(results, r)
+				inflightChunks -= 1
+			}
+		}
+		buf = binary.LittleEndian.AppendUint16(buf, uint16(len(key)))
+		buf = append(buf, key...)
+		buf = append(buf, byte(op))
+		buf = binary.LittleEndian.AppendUint32(buf, uint32(len(val)))
+		buf = append(buf, val...)
+	}
+	if len(buf) > 0 {
+		totalSize += len(buf)
+		inflightChunks += 1
+		ran := newTxnChunkRange(chunkSmallest, mutations.GetKey(mutations.Len()-1))
+		writer.asyncBuildChunk(bo, buf, ran, resultCh)
+	}
+
+	for i := 0; i < inflightChunks; i++ {
+		r := <-resultCh
+		if r.err != nil {
+			logutil.Logger(bo.GetCtx()).Error(buildChunkErrMsg, zap.Error(err))
+			return errors.Wrap(r.err, buildChunkErrMsg)
+		}
+		results = append(results, r)
+	}
+	sort.Slice(results, func(i, j int) bool {
+		return bytes.Compare(results[i].chunkRange.smallest, results[j].chunkRange.smallest) < 0
+	})
+	for _, r := range results {
+		c.txnFileCtx.slice.append(r.chunkId, r.chunkRange)
+	}
+
+	logutil.Logger(bo.GetCtx()).Info("build txn files",
+		zap.Uint64("startTS", c.startTS),
+		zap.Int("mutationsLen", mutations.Len()),
+		zap.Int("totalChunksSize", totalSize),
+		zap.Any("chunkIDs", c.txnFileCtx.slice.chunkIDs))
+	return nil
+}
+
+func (c *twoPhaseCommitter) useTxnFile() bool {
+	if c.txn == nil || !c.txn.vars.EnableTxnFile {
+		return false
+	}
+	conf := config.GetGlobalConfig()
+	// Don't use txn file for internal request to avoid affect system tables or metadata before it is stable enough.
+	// TODO: use txn file for internal TTL & DDL tasks.
+	return !c.txn.isPessimistic &&
+		!c.txn.isInternal() &&
+		len(conf.TiKVClient.TxnChunkWriterAddr) > 0 &&
+		uint64(c.txn.GetMemBuffer().Size()) >= conf.TiKVClient.TxnFileMinMutationSize
+}
+
+func (c *twoPhaseCommitter) preSplitTxnFileRegions(bo *retry.Backoffer) error {
+	batches, err := c.txnFileCtx.slice.groupToBatches(c.store.GetRegionCache(), bo)
+	if err != nil {
+		return errors.Wrap(err, "group to batches failed")
+	}
+	var splitKeys [][]byte
+	for _, batch := range batches {
+		if batch.len() > PreSplitRegionChunks {
+			for i := PreSplitRegionChunks; i < batch.len(); i += PreSplitRegionChunks {
+				splitKeys = append(splitKeys, batch.chunkRanges[i].smallest)
+			}
+		}
+	}
+	if len(splitKeys) == 0 {
+		return nil
+	}
+	_, err = c.store.SplitRegions(bo.GetCtx(), splitKeys, false, nil)
+	return errors.Wrap(err, "pre split regions failed")
+}
+
+type chunkWriterClient struct {
+	cli         *http.Client
+	serviceAddr string
+}
+
+func newChunkWriterClient() (*chunkWriterClient, error) {
+	var (
+		cfg     = config.GetGlobalConfig()
+		scheme  = "http://"
+		timeout = time.Duration(BuildTxnFileMaxBackoff.Load()) * time.Millisecond
+		client  = &http.Client{Timeout: timeout}
+	)
+	if len(cfg.Security.ClusterSSLCA) != 0 {
+		tlsConfig, err := cfg.Security.ToTLSConfig()
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		scheme = "https://"
+		client.Transport = &http.Transport{
+			TLSClientConfig:   tlsConfig,
+			ForceAttemptHTTP2: true,
+		}
+	}
+	serviceAddr := fmt.Sprintf("%s%s/txn_chunk", scheme, cfg.TiKVClient.TxnChunkWriterAddr)
+	return &chunkWriterClient{client, serviceAddr}, nil
+}
+
+func (w *chunkWriterClient) buildChunk(bo *retry.Backoffer, buf []byte) (uint64, error) {
+	ctx := bo.GetCtx()
+	hash := crc32.New(crc32.MakeTable(crc32.IEEE))
+	hash.Write(buf)
+	crc := hash.Sum32()
+	buf = binary.LittleEndian.AppendUint32(buf, crc)
+
+	for {
+		if ctx.Err() != nil {
+			return 0, errors.WithStack(ctx.Err())
+		}
+
+		req, err := http.NewRequestWithContext(ctx, "POST", w.serviceAddr, bytes.NewReader(buf))
+		if err != nil {
+			return 0, errors.WithStack(err)
+		}
+		req.Header.Set("Content-Type", "application/octet-stream")
+
+		resp, err := w.cli.Do(req)
+		if err != nil {
+			logutil.Logger(bo.GetCtx()).Warn("build txn file request failed", zap.Error(err), zap.String("addr", w.serviceAddr))
+			err = bo.Backoff(retry.BoTiKVRPC, errors.WithMessage(err, "build txn file request failed"))
+			if err != nil {
+				return 0, errors.WithStack(err)
+			}
+			continue
+		}
+		defer resp.Body.Close()
+		if resp.StatusCode != http.StatusOK {
+			var bodyStr string
+			if data, err := io.ReadAll(resp.Body); err == nil {
+				bodyStr = string(data)
+			}
+			logutil.Logger(bo.GetCtx()).Warn("build txn file service error", zap.String("http status", resp.Status), zap.String("body", bodyStr))
+			err = bo.Backoff(retry.BoTiKVServerBusy, errors.WithMessagef(err, "build txn file service error, http status %s", resp.Status))
+			if err != nil {
+				return 0, errors.WithStack(err)
+			}
+			continue
+		}
+		data, err := io.ReadAll(resp.Body)
+		if err != nil {
+			return 0, errors.WithStack(err)
+		}
+		v := struct {
+			ChunkId uint64 `json:"chunk_id"`
+		}{}
+		if err = json.Unmarshal(data, &v); err != nil {
+			return 0, errors.Wrapf(err, "unmarshal response %s", string(data))
+		}
+		logutil.Logger(bo.GetCtx()).Debug("build txn file", zap.Int("size", len(buf)), zap.Uint64("chunkId", v.ChunkId))
+		return v.ChunkId, nil
+	}
+}
+
+type buildChunkResult struct {
+	chunkId    uint64
+	chunkRange txnChunkRange
+	err        error
+}
+
+func (w *chunkWriterClient) asyncBuildChunk(bo *retry.Backoffer, buf []byte, chunkRange txnChunkRange, resultCh chan<- buildChunkResult) {
+	go func() {
+		chunkId, err := w.buildChunk(bo.Clone(), buf)
+		resultCh <- buildChunkResult{chunkId, chunkRange, err}
+	}()
+}
diff --git a/txnkv/txnlock/lock_resolver.go b/txnkv/txnlock/lock_resolver.go
index b93b86e..86c69a0 100644
--- a/txnkv/txnlock/lock_resolver.go
+++ b/txnkv/txnlock/lock_resolver.go
@@ -171,6 +171,7 @@ type Lock struct {
 	UseAsyncCommit  bool
 	LockForUpdateTS uint64
 	MinCommitTS     uint64
+	IsTxnFile       bool
 }
 
 func (l *Lock) String() string {
@@ -179,8 +180,8 @@ func (l *Lock) String() string {
 	buf.WriteString(hex.EncodeToString(l.Key))
 	buf.WriteString(", primary: ")
 	buf.WriteString(hex.EncodeToString(l.Primary))
-	return fmt.Sprintf("%s, txnStartTS: %d, lockForUpdateTS:%d, minCommitTs:%d, ttl: %d, type: %s, UseAsyncCommit: %t, txnSize: %d",
-		buf.String(), l.TxnID, l.LockForUpdateTS, l.MinCommitTS, l.TTL, l.LockType, l.UseAsyncCommit, l.TxnSize)
+	return fmt.Sprintf("%s, txnStartTS: %d, lockForUpdateTS:%d, minCommitTs:%d, ttl: %d, type: %s, UseAsyncCommit: %t, txnSize: %d, isTxnFile: %t",
+		buf.String(), l.TxnID, l.LockForUpdateTS, l.MinCommitTS, l.TTL, l.LockType, l.UseAsyncCommit, l.TxnSize, l.IsTxnFile)
 }
 
 // NewLock creates a new *Lock.
@@ -195,6 +196,7 @@ func NewLock(l *kvrpcpb.LockInfo) *Lock {
 		UseAsyncCommit:  l.UseAsyncCommit,
 		LockForUpdateTS: l.LockForUpdateTs,
 		MinCommitTS:     l.MinCommitTs,
+		IsTxnFile:       l.IsTxnFile,
 	}
 }
 
@@ -236,6 +238,7 @@ func (lr *LockResolver) BatchResolveLocks(bo *retry.Backoffer, locks []*Lock, lo
 	expiredLocks := locks
 
 	txnInfos := make(map[uint64]uint64)
+	txnFileIDs := make(map[uint64]bool)
 	startTime := time.Now()
 	for _, l := range expiredLocks {
 		logutil.Logger(bo.GetCtx()).Debug("BatchResolveLocks handling lock", zap.Stringer("lock", l))
@@ -292,6 +295,9 @@ func (lr *LockResolver) BatchResolveLocks(bo *retry.Backoffer, locks []*Lock, lo
 		}
 
 		txnInfos[l.TxnID] = status.commitTS
+		if l.IsTxnFile {
+			txnFileIDs[l.TxnID] = true
+		}
 	}
 	logutil.BgLogger().Info("BatchResolveLocks: lookup txn status",
 		zap.Duration("cost time", time.Since(startTime)),
@@ -300,8 +306,9 @@ func (lr *LockResolver) BatchResolveLocks(bo *retry.Backoffer, locks []*Lock, lo
 	listTxnInfos := make([]*kvrpcpb.TxnInfo, 0, len(txnInfos))
 	for txnID, status := range txnInfos {
 		listTxnInfos = append(listTxnInfos, &kvrpcpb.TxnInfo{
-			Txn:    txnID,
-			Status: status,
+			Txn:       txnID,
+			Status:    status,
+			IsTxnFile: txnFileIDs[txnID],
 		})
 	}
 
@@ -738,6 +745,7 @@ func (lr *LockResolver) getTxnStatus(bo *retry.Backoffer, txnID uint64, primary
 
 	var status TxnStatus
 	resolvingPessimisticLock := lockInfo != nil && lockInfo.LockType == kvrpcpb.Op_PessimisticLock
+	isTxnFile := lockInfo != nil && lockInfo.IsTxnFile
 	req := tikvrpc.NewRequest(tikvrpc.CmdCheckTxnStatus, &kvrpcpb.CheckTxnStatusRequest{
 		PrimaryKey:               primary,
 		LockTs:                   txnID,
@@ -747,6 +755,7 @@ func (lr *LockResolver) getTxnStatus(bo *retry.Backoffer, txnID uint64, primary
 		ForceSyncCommit:          forceSyncCommit,
 		ResolvingPessimisticLock: resolvingPessimisticLock,
 		VerifyIsPrimary:          true,
+		IsTxnFile:                isTxnFile,
 	}, kvrpcpb.Context{
 		RequestSource: util.RequestSourceFromCtx(bo.GetCtx()),
 		ResourceControlContext: &kvrpcpb.ResourceControlContext{
@@ -1053,6 +1062,7 @@ func (lr *LockResolver) checkAllSecondaries(bo *retry.Backoffer, l *Lock, status
 func (lr *LockResolver) resolveRegionLocks(bo *retry.Backoffer, l *Lock, region locate.RegionVerID, keys [][]byte, status TxnStatus) error {
 	lreq := &kvrpcpb.ResolveLockRequest{
 		StartVersion: l.TxnID,
+		IsTxnFile:    l.IsTxnFile,
 	}
 	if status.IsCommitted() {
 		lreq.CommitVersion = status.CommitTS()
@@ -1125,6 +1135,7 @@ func (lr *LockResolver) resolveLock(bo *retry.Backoffer, l *Lock, status TxnStat
 		}
 		lreq := &kvrpcpb.ResolveLockRequest{
 			StartVersion: l.TxnID,
+			IsTxnFile:    l.IsTxnFile,
 		}
 		if status.IsCommitted() {
 			lreq.CommitVersion = status.CommitTS()
diff --git a/util/merge_ranges.go b/util/merge_ranges.go
new file mode 100644
index 0000000..1d7e2a5
--- /dev/null
+++ b/util/merge_ranges.go
@@ -0,0 +1,134 @@
+// 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 util
+
+import (
+	"bytes"
+	"container/list"
+)
+
+// To make life easier.
+// MergeRanges is used for txn file only, all keys are TiDB keys, and must not be bigger than following `maxEndKey`.
+var maxEndKey = []byte{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff}
+
+type keyRange struct {
+	start []byte
+	end   []byte
+}
+
+type MergeRanges struct {
+	ranges *list.List
+}
+
+func NewMergeRanges() *MergeRanges {
+	r := &MergeRanges{
+		ranges: list.New(),
+	}
+	return r
+}
+
+func (m *MergeRanges) IsEmpty() bool {
+	return m.ranges.Len() == 0
+}
+
+// searchByStart search range by start key.
+// If the key is found, return the element and true.
+// If the key is not found, return the element to insert BEFORE and false. When the mark is nil, push to the back.
+func (m *MergeRanges) searchByStart(key []byte) (mark *list.Element, ok bool) {
+	for e := m.ranges.Front(); e != nil; e = e.Next() {
+		cmp := bytes.Compare(e.Value.(*keyRange).start, key)
+		if cmp == 0 {
+			return e, true
+		} else if cmp > 0 {
+			return e, false
+		}
+	}
+	return nil, false
+}
+
+// searchOverlapped search the overlapped range for key.
+// range.start <= key <= range.end
+// If the key is found, return the element and true.
+// If the key is not found, return the element to insert BEFORE and false. When the mark is nil, push to the back.
+func (m *MergeRanges) searchOverlapped(key []byte) (mark *list.Element, overlapped bool) {
+	mark, overlapped = m.searchByStart(key)
+	if overlapped {
+		return mark, true
+	}
+
+	var prev *list.Element
+	if mark == nil {
+		prev = m.ranges.Back()
+	} else {
+		prev = mark.Prev()
+	}
+
+	if prev == nil {
+		return m.ranges.Front(), false
+	}
+	if bytes.Compare(key, prev.Value.(*keyRange).end) <= 0 {
+		return prev, true
+	}
+	return mark, false
+}
+
+// Covered returns whether the range [start,end) is fully covered.
+func (m *MergeRanges) Covered(start []byte, end []byte) bool {
+	if len(end) == 0 {
+		end = maxEndKey
+	}
+
+	mark, ok := m.searchOverlapped(start)
+	if ok {
+		return bytes.Compare(end, mark.Value.(*keyRange).end) <= 0
+	}
+	return false
+}
+
+// Insert inserts the range [start,end) into MergeRanges.
+func (m *MergeRanges) Insert(start []byte, end []byte) {
+	if len(end) == 0 {
+		end = maxEndKey
+	}
+
+	left, ok := m.searchOverlapped(start)
+	if ok {
+		if bytes.Compare(end, left.Value.(*keyRange).end) <= 0 {
+			return
+		}
+		start = left.Value.(*keyRange).start
+	}
+
+	if left == nil {
+		m.ranges.PushBack(&keyRange{start, end})
+		return
+	}
+
+	right, ok := m.searchOverlapped(end)
+	rightBound := right
+	if ok {
+		end = right.Value.(*keyRange).end
+		rightBound = right.Next()
+	}
+
+	m.ranges.InsertBefore(&keyRange{start, end}, left)
+
+	// Remove overlapped.
+	var next *list.Element
+	for e := left; e != rightBound; e = next {
+		next = e.Next()
+		m.ranges.Remove(e)
+	}
+}


Test patch:
diff --git a/integration_tests/go.mod b/integration_tests/go.mod
index 5c7dd14..6c3a255 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.20231212100244-799fae176cfb
 	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c
-	github.com/pingcap/kvproto v0.0.0-20240109063850-932639606bcf
+	github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9
 	github.com/pingcap/tidb v1.1.0-beta.0.20231025055903-f3dfd896b5e6
 	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.20240304062027-038a52519d39
-	github.com/tikv/pd/client v0.0.0-20240318054542-da3b6e381c9b
+	github.com/tikv/pd/client v0.0.0-20240410074522-b9240a065877
 	go.uber.org/goleak v1.3.0
 )
 
diff --git a/integration_tests/go.sum b/integration_tests/go.sum
index 1e32dff..b17b4fd 100644
--- a/integration_tests/go.sum
+++ b/integration_tests/go.sum
@@ -416,8 +416,8 @@ github.com/pingcap/fn v0.0.0-20200306044125-d5540d389059/go.mod h1:fMRU1BA1y+r89
 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-20240109063850-932639606bcf h1:n3FMveYjc2VuETjo6YhmsgkDx0P/yLJTvk96BJdCq6Y=
-github.com/pingcap/kvproto v0.0.0-20240109063850-932639606bcf/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8=
+github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9 h1:tAfzWZi+hVqX8tM+jnltSiC2k3X+305L/aGiblhX6dk=
+github.com/pingcap/kvproto v0.0.0-20240314010430-d552d2f77da9/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.20230317032135-a0d097d16e22 h1:2SOzvGvE8beiC1Y4g9Onkvu6UmuBBOeWRGQEjJaT/JY=
@@ -530,8 +530,8 @@ 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 v0.0.0-20230420042919-0e0313adef1f h1:1BBlbLSCua+oK3xngEjPO2dj1oLcItJxhCzqYG6XANA=
 github.com/tikv/pd v0.0.0-20230420042919-0e0313adef1f/go.mod h1:2hGdvUNGLaCKLbEXC8JMiQz2JsbxEDCKGDOBGJTt0NQ=
-github.com/tikv/pd/client v0.0.0-20240318054542-da3b6e381c9b h1:BiFQvT54X/UMNjqO/UXqp7IWl9bvIaOfNaraqdpo++g=
-github.com/tikv/pd/client v0.0.0-20240318054542-da3b6e381c9b/go.mod h1:Z/QAgOt29zvwBTd0H6pdx45VO6KRNc/O/DzGkVmSyZg=
+github.com/tikv/pd/client v0.0.0-20240410074522-b9240a065877 h1:8Poo/LLJo9sMO2Yak5+0yWEiN04NenJvvBGBafSERGg=
+github.com/tikv/pd/client v0.0.0-20240410074522-b9240a065877/go.mod h1:Qdzax77EzM8h6NbzwKymp96AB0DUvtkHCcC1jwxoiGU=
 github.com/tklauser/go-sysconf v0.3.9/go.mod h1:11DU/5sG7UexIrp/O6g35hrWzu0JxlwQ3LSFUzyeuhs=
 github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU=
 github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI=
diff --git a/internal/locate/region_request_test.go b/internal/locate/region_request_test.go
index 8563715..b82e927 100644
--- a/internal/locate/region_request_test.go
+++ b/internal/locate/region_request_test.go
@@ -356,6 +356,12 @@ func (s *mockTikvGrpcServer) KvGC(context.Context, *kvrpcpb.GCRequest) (*kvrpcpb
 func (s *mockTikvGrpcServer) KvDeleteRange(context.Context, *kvrpcpb.DeleteRangeRequest) (*kvrpcpb.DeleteRangeResponse, error) {
 	return nil, errors.New("unreachable")
 }
+func (s *mockTikvGrpcServer) KvFlush(context.Context, *kvrpcpb.FlushRequest) (*kvrpcpb.FlushResponse, error) {
+	return nil, errors.New("unreachable")
+}
+func (s *mockTikvGrpcServer) KvBufferBatchGet(context.Context, *kvrpcpb.BufferBatchGetRequest) (*kvrpcpb.BufferBatchGetResponse, error) {
+	return nil, errors.New("unreachable")
+}
 func (s *mockTikvGrpcServer) RawGet(context.Context, *kvrpcpb.RawGetRequest) (*kvrpcpb.RawGetResponse, error) {
 	return nil, errors.New("unreachable")
 }
diff --git a/txnkv/transaction/test_probe.go b/txnkv/transaction/test_probe.go
index f35017f..cc9201b 100644
--- a/txnkv/transaction/test_probe.go
+++ b/txnkv/transaction/test_probe.go
@@ -371,7 +371,7 @@ func (c CommitterProbe) CleanupMutations(ctx context.Context) error {
 
 // 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, false)
 }
 
 // ConfigProbe exposes configurations and global variables for testing purpose.
diff --git a/txnkv/transaction/txn_file_test.go b/txnkv/transaction/txn_file_test.go
new file mode 100644
index 0000000..dc1c0ab
--- /dev/null
+++ b/txnkv/transaction/txn_file_test.go
@@ -0,0 +1,81 @@
+// 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 transaction
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/tikv/client-go/v2/internal/locate"
+)
+
+func TestTxnFileChunkBatch(t *testing.T) {
+	assert := assert.New(t)
+	chunkSlice := txnChunkSlice{
+		chunkIDs: []uint64{1, 2, 3},
+		chunkRanges: []txnChunkRange{
+			{
+				smallest: []byte("k01"),
+				biggest:  []byte("k01"),
+			},
+			{
+				smallest: []byte("k02"),
+				biggest:  []byte("k04"),
+			},
+			{
+				smallest: []byte("k05"),
+				biggest:  []byte("k08"),
+			},
+		},
+	}
+
+	cases := []struct {
+		startKey string
+		endKey   string
+		expected []string
+	}{
+		{
+			"", "", []string{"k01", "k02", "k05"},
+		},
+		{
+			"k01", "k05", []string{"k01", "k02"},
+		},
+		{
+			"k010", "k05", []string{"k02"},
+		},
+		{
+			"k03", "k05", []string{},
+		},
+		{
+			"k03", "", []string{"k05"},
+		},
+	}
+
+	for _, c := range cases {
+		batch := chunkBatch{
+			txnChunkSlice: chunkSlice,
+			region: &locate.KeyLocation{
+				StartKey: []byte(c.startKey),
+				EndKey:   []byte(c.endKey),
+			},
+		}
+		keys := batch.getSampleKeys()
+		strKeys := make([]string, 0, len(keys))
+		for _, key := range keys {
+			strKeys = append(strKeys, string(key))
+		}
+		assert.Equal(c.expected, strKeys)
+	}
+}
diff --git a/util/merge_ranges_test.go b/util/merge_ranges_test.go
new file mode 100644
index 0000000..d7c558c
--- /dev/null
+++ b/util/merge_ranges_test.go
@@ -0,0 +1,225 @@
+// 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 util
+
+import (
+	"container/list"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestMergeRangesSearchOverlapped(t *testing.T) {
+	assert := require.New(t)
+
+	mr := NewMergeRanges()
+	type Case struct {
+		key        int
+		overlapped bool
+		rng        *keyRange // the range overlap with or after the key
+	}
+	checkOverlapped := func(c Case) {
+		mark, overlapped := mr.searchOverlapped(iToKey(c.key))
+		assert.Equal(c.overlapped, overlapped)
+		if c.rng == nil {
+			assert.Nil(mark)
+		} else {
+			assert.Equal(c.rng, mark.Value.(*keyRange))
+		}
+	}
+
+	// mr: empty
+	checkOverlapped(Case{0, false, nil})
+
+	insertRange(mr, 0, 1)
+	// mr: [0,1)
+	cases := []Case{
+		{0, true, makeRange(0, 1)},
+		{1, true, makeRange(0, 1)},
+		{2, false, nil},
+	}
+	for _, c := range cases {
+		checkOverlapped(c)
+	}
+
+	insertRange(mr, 3, 5)
+	// mr: [0,1), [3,5)
+	cases = []Case{
+		{0, true, makeRange(0, 1)},
+		{1, true, makeRange(0, 1)},
+		{2, false, makeRange(3, 5)},
+		{3, true, makeRange(3, 5)},
+		{4, true, makeRange(3, 5)},
+		{5, true, makeRange(3, 5)},
+		{6, false, nil},
+	}
+	for _, c := range cases {
+		checkOverlapped(c)
+	}
+}
+
+func TestMergeRangesCovered(t *testing.T) {
+	assert := require.New(t)
+
+	mr := NewMergeRanges()
+	type Case struct {
+		start   int
+		end     int
+		covered bool
+	}
+	checkCovered := func(c Case) {
+		assert.Equal(c.covered, mr.Covered(iToKey(c.start), iToKey(c.end)))
+	}
+
+	// mr: empty
+	checkCovered(Case{0, 1, false})
+
+	insertRange(mr, 0, 1)
+	// mr: [0,1)
+	cases := []Case{
+		{0, 1, true},
+		{1, 2, false},
+	}
+	for _, c := range cases {
+		checkCovered(c)
+	}
+
+	insertRange(mr, 2, 4)
+	// mr: [0,1), [2,4)
+	cases = []Case{
+		{0, 1, true},
+		{1, 2, false},
+		{2, 3, true},
+		{2, 4, true},
+		{1, 2, false},
+		{1, 3, false},
+		{1, 4, false},
+		{1, 5, false},
+		{2, 5, false},
+		{4, 5, false},
+		{0, 4, false},
+	}
+	for _, c := range cases {
+		checkCovered(c)
+	}
+}
+
+func TestMergeRangesInsert(t *testing.T) {
+	assert := require.New(t)
+
+	mr := NewMergeRanges()
+	assertEqual := func(keys [][2]int) {
+		expected := makeMergeRanges(keys)
+		assert.Equal(expected.ranges, mr.ranges)
+	}
+
+	insertRange(mr, 2, 3)
+	assertEqual([][2]int{{2, 3}})
+	insertRange(mr, 3, 4)
+	assertEqual([][2]int{{2, 4}})
+
+	insertRange(mr, 0, 1)
+	assertEqual([][2]int{{0, 1}, {2, 4}})
+	insertRange(mr, 1, 2)
+	assertEqual([][2]int{{0, 4}})
+
+	insertRange(mr, 5, 6)
+	assertEqual([][2]int{{0, 4}, {5, 6}})
+	insertRange(mr, 4, 5)
+	assertEqual([][2]int{{0, 6}})
+
+	insertRange(mr, 10, 11)
+	assertEqual([][2]int{{0, 6}, {10, 11}})
+	insertRange(mr, 3, 12)
+	assertEqual([][2]int{{0, 12}})
+
+	assert.True(mr.Covered(iToKey(0), iToKey(12)))
+}
+
+func TestMergeRangesOverlappingRanges(t *testing.T) {
+	assert := require.New(t)
+
+	// mr: [2,12)
+	mr := makeMergeRanges([][2]int{{2, 12}})
+	assertEqual := func(keys [][2]int) {
+		expected := makeMergeRanges(keys)
+		assert.Equal(expected.ranges, mr.ranges)
+	}
+
+	insertRange(mr, 2, 6)
+	assertEqual([][2]int{{2, 12}})
+
+	insertRange(mr, 3, 8)
+	assertEqual([][2]int{{2, 12}})
+
+	insertRange(mr, 6, 12)
+	assertEqual([][2]int{{2, 12}})
+
+	insertRange(mr, 1, 4)
+	assertEqual([][2]int{{1, 12}})
+
+	insertRange(mr, 4, 15)
+	assertEqual([][2]int{{1, 15}})
+
+	insertRange(mr, 20, 21)
+	insertRange(mr, 22, 23)
+	insertRange(mr, 24, 25)
+	insertRange(mr, 14, 30)
+	assertEqual([][2]int{{1, 30}})
+
+	insertRange(mr, 40, 41)
+	insertRange(mr, 42, 43)
+	insertRange(mr, 50, 51)
+	insertRange(mr, 53, 54)
+	insertRange(mr, 0, 42)
+	assertEqual([][2]int{{0, 43}, {50, 51}, {53, 54}})
+}
+
+func TestMergeRangesEmptyKey(t *testing.T) {
+	assert := require.New(t)
+
+	mr := NewMergeRanges()
+
+	assert.False(mr.Covered(nil, nil))
+
+	mr.Insert(nil, iToKey(10)) // (infinite, 10)
+	mr.Insert(iToKey(10), nil) // [10, infinite)
+	assert.True(mr.Covered(iToKey(0), iToKey(100)))
+	assert.True(mr.Covered(nil, nil))
+}
+
+func iToKey(i int) []byte {
+	return []byte(fmt.Sprintf("%04d", i))
+}
+
+func makeRange(start, end int) *keyRange {
+	return &keyRange{
+		start: iToKey(start),
+		end:   iToKey(end),
+	}
+}
+
+func insertRange(mr *MergeRanges, start, end int) {
+	mr.Insert(iToKey(start), iToKey(end))
+}
+
+func makeMergeRanges(keys [][2]int) *MergeRanges {
+	l := list.New()
+	for _, key := range keys {
+		l.PushBack(&keyRange{iToKey(key[0]), iToKey(key[1])})
+	}
+	return &MergeRanges{ranges: l}
+}