[WIP] Support File Based Transaction

config/client.go

@@ -90,6 +90,9 @@ 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.
+	TxnChunkWriterAddr string `toml:"txn-chunk-writer-addr" json:"txn-chunk-writer-addr"`
 }
 
 // AsyncCommit is the config for the async commit feature. The switch to enable it is a system variable.

go.mod

@@ -59,3 +59,5 @@ require (
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
+
+replace github.com/pingcap/kvproto => github.com/tidbcloud/kvproto v0.0.0-20240117032603-369bbbf7b45e

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=

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
@@ -917,7 +922,7 @@ func (c *twoPhaseCommitter) preSplitRegion(ctx context.Context, group groupedMut
 // CommitSecondaryMaxBackoff is max sleep time of the 'commit' command
 const CommitSecondaryMaxBackoff = 41000
 
-// doActionOnGroupedMutations splits groups into batches (there is one group per region, and potentially many batches per group, but all mutations
+// doActionOnGroupedMutations splits groups into batches (there is one slice per region, and potentially many batches per slice, but all mutations
 // in a batch will belong to the same region).
 func (c *twoPhaseCommitter) doActionOnGroupMutations(bo *retry.Backoffer, action twoPhaseCommitAction, groups []groupedMutations) error {
 	action.tiKVTxnRegionsNumHistogram().Observe(float64(len(groups)))
@@ -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)

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.

txnkv/transaction/txn.go

@@ -154,7 +154,7 @@ type KVTxn struct {
 	interceptor    interceptor.RPCInterceptor
 	assertionLevel kvrpcpb.AssertionLevel
 	*util.RequestSource
-	// resourceGroupName is the name of tenant resource group.
+	// resourceGroupName is the name of tenant resource slice.
 	resourceGroupName string
 
 	aggressiveLockingContext *aggressiveLockingContext
@@ -296,7 +296,7 @@ func (txn *KVTxn) SetResourceGroupTagger(tagger tikvrpc.ResourceGroupTagger) {
 	txn.GetSnapshot().SetResourceGroupTagger(tagger)
 }
 
-// SetResourceGroupName set resource group name for both read and write.
+// SetResourceGroupName set resource slice name for both read and write.
 func (txn *KVTxn) SetResourceGroupName(name string) {
 	txn.resourceGroupName = name
 	txn.GetSnapshot().SetResourceGroupName(name)
@@ -502,6 +502,13 @@ func (txn *KVTxn) Commit(ctx context.Context) error {
 			}
 		}
 	}()
+	if committer.useTxnFile() {
+		err = committer.executeTxnFile(ctx)
+		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() {