Repository: tikv/client-go (go)

Issue:
Support 'flashback' request
Ref:
* https://github.com/pingcap/tidb/issues/37197
* https://github.com/tikv/tikv/issues/13303

We plan to support an MVCC-based flashback feature for TiDB, as described in the issues above. The proposed RPC request:

```
message FlashbackToVersionRequest {
    Context context = 1;
    // The TS version which the data should flashback to.
    uint64 version = 2;
    bytes start_key = 3;
    bytes end_key = 4;
}

message FlashbackToVersionResponse {
    errorpb.Error region_error = 1;
    string error = 2;
}
```

We would like to implement the new interface based on it:

```
func flashback(keyRanges []kv.KeyRange, version int64) (status Status, err error)
```

In which:
* `keyRanges`: a series of non-overlaped start/end key pairs that need to be flash-backed by TiKV.
* `version`: the timestamp of MVCC version that need to be flash-backed by TiKV.
* `status`: the status of flashback request: if TiKV is doing the flashback job, it returns the total number of regions and the number of done regions; if the job is not started by TiKV, it starts running the job.


Pull request description:
*: support the kv_flashback_to_version cmd
Signed-off-by: JmPotato <ghzpotato@gmail.com>

Close #569.

Ref: https://github.com/tikv/tikv/issues/13303, https://github.com/pingcap/tidb/issues/37197.

Patch:
diff --git a/go.mod b/go.mod
index ebbb8d2..5ada711 100644
--- a/go.mod
+++ b/go.mod
@@ -13,7 +13,7 @@ require (
 	github.com/opentracing/opentracing-go v1.2.0
 	github.com/pingcap/failpoint v0.0.0-20210918120811-547c13e3eb00
 	github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989
-	github.com/pingcap/kvproto v0.0.0-20220818023518-a0f02b6efcee
+	github.com/pingcap/kvproto v0.0.0-20220902075447-f95ac338b331
 	github.com/pingcap/log v0.0.0-20211215031037-e024ba4eb0ee
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.11.0
diff --git a/go.sum b/go.sum
index a045316..702e1e2 100644
--- a/go.sum
+++ b/go.sum
@@ -155,8 +155,8 @@ github.com/pingcap/failpoint v0.0.0-20210918120811-547c13e3eb00/go.mod h1:4qGtCB
 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-20220510035547-0e2f26c0a46a/go.mod h1:OYtxs0786qojVTmkVeufx93xe+jUgm56GUYRIKnmaGI=
-github.com/pingcap/kvproto v0.0.0-20220818023518-a0f02b6efcee h1:s1al2ci3MEj5VnNuUCGAfeqpbCxcMeZibOyxw8ClHLE=
-github.com/pingcap/kvproto v0.0.0-20220818023518-a0f02b6efcee/go.mod h1:OYtxs0786qojVTmkVeufx93xe+jUgm56GUYRIKnmaGI=
+github.com/pingcap/kvproto v0.0.0-20220902075447-f95ac338b331 h1:LLPNqaM0C0hHER4U+cl17pvVBFuSDBbVl2oJQ6f6bg4=
+github.com/pingcap/kvproto v0.0.0-20220902075447-f95ac338b331/go.mod h1:OYtxs0786qojVTmkVeufx93xe+jUgm56GUYRIKnmaGI=
 github.com/pingcap/log v0.0.0-20211215031037-e024ba4eb0ee h1:VO2t6IBpfvW34TdtD/G10VvnGqjLic1jzOuHjUb5VqM=
 github.com/pingcap/log v0.0.0-20211215031037-e024ba4eb0ee/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
diff --git a/tikvrpc/tikvrpc.go b/tikvrpc/tikvrpc.go
index d9c9d9c..eaad282 100644
--- a/tikvrpc/tikvrpc.go
+++ b/tikvrpc/tikvrpc.go
@@ -72,6 +72,7 @@ const (
 	CmdTxnHeartBeat
 	CmdCheckTxnStatus
 	CmdCheckSecondaryLocks
+	CmdFlashbackToVersion
 
 	CmdRawGet CmdType = 256 + iota
 	CmdRawBatchGet
@@ -203,6 +204,8 @@ func (t CmdType) String() string {
 		return "StoreSafeTS"
 	case CmdLockWaitInfo:
 		return "LockWaitInfo"
+	case CmdFlashbackToVersion:
+		return "FlashbackToVersion"
 	}
 	return "Unknown"
 }
@@ -511,6 +514,11 @@ func (req *Request) LockWaitInfo() *kvrpcpb.GetLockWaitInfoRequest {
 	return req.Req.(*kvrpcpb.GetLockWaitInfoRequest)
 }
 
+// FlashbackToVersion returns FlashbackToVersionRequest in request.
+func (req *Request) FlashbackToVersion() *kvrpcpb.FlashbackToVersionRequest {
+	return req.Req.(*kvrpcpb.FlashbackToVersionRequest)
+}
+
 // ToBatchCommandsRequest converts the request to an entry in BatchCommands request.
 func (req *Request) ToBatchCommandsRequest() *tikvpb.BatchCommandsRequest_Request {
 	switch req.Type {
@@ -566,6 +574,8 @@ func (req *Request) ToBatchCommandsRequest() *tikvpb.BatchCommandsRequest_Reques
 		return &tikvpb.BatchCommandsRequest_Request{Cmd: &tikvpb.BatchCommandsRequest_Request_CheckSecondaryLocks{CheckSecondaryLocks: req.CheckSecondaryLocks()}}
 	case CmdTxnHeartBeat:
 		return &tikvpb.BatchCommandsRequest_Request{Cmd: &tikvpb.BatchCommandsRequest_Request_TxnHeartBeat{TxnHeartBeat: req.TxnHeartBeat()}}
+	case CmdFlashbackToVersion:
+		return &tikvpb.BatchCommandsRequest_Request{Cmd: &tikvpb.BatchCommandsRequest_Request_FlashbackToVersion{FlashbackToVersion: req.FlashbackToVersion()}}
 	}
 	return nil
 }
@@ -603,6 +613,8 @@ func FromBatchCommandsResponse(res *tikvpb.BatchCommandsResponse_Response) (*Res
 		return &Response{Resp: res.GC}, nil
 	case *tikvpb.BatchCommandsResponse_Response_DeleteRange:
 		return &Response{Resp: res.DeleteRange}, nil
+	case *tikvpb.BatchCommandsResponse_Response_FlashbackToVersion:
+		return &Response{Resp: res.FlashbackToVersion}, nil
 	case *tikvpb.BatchCommandsResponse_Response_RawGet:
 		return &Response{Resp: res.RawGet}, nil
 	case *tikvpb.BatchCommandsResponse_Response_RawBatchGet:
@@ -753,6 +765,8 @@ func SetContext(req *Request, region *metapb.Region, peer *metapb.Peer) error {
 		req.CheckTxnStatus().Context = ctx
 	case CmdCheckSecondaryLocks:
 		req.CheckSecondaryLocks().Context = ctx
+	case CmdFlashbackToVersion:
+		req.FlashbackToVersion().Context = ctx
 	default:
 		return errors.Errorf("invalid request type %v", req.Type)
 	}
@@ -900,6 +914,10 @@ func GenRegionErrorResp(req *Request, e *errorpb.Error) (*Response, error) {
 		p = &kvrpcpb.CheckSecondaryLocksResponse{
 			RegionError: e,
 		}
+	case CmdFlashbackToVersion:
+		p = &kvrpcpb.FlashbackToVersionResponse{
+			RegionError: e,
+		}
 	default:
 		return nil, errors.Errorf("invalid request type %v", req.Type)
 	}
@@ -1054,6 +1072,8 @@ func CallRPC(ctx context.Context, client tikvpb.TikvClient, req *Request) (*Resp
 		resp.Resp, err = client.GetLockWaitInfo(ctx, req.LockWaitInfo())
 	case CmdCompact:
 		resp.Resp, err = client.Compact(ctx, req.Compact())
+	case CmdFlashbackToVersion:
+		resp.Resp, err = client.KvFlashbackToVersion(ctx, req.FlashbackToVersion())
 	default:
 		return nil, errors.Errorf("invalid request type: %v", req.Type)
 	}
@@ -1215,7 +1235,8 @@ func (req *Request) IsTxnWriteRequest() bool {
 		req.Type == CmdCheckSecondaryLocks ||
 		req.Type == CmdCleanup ||
 		req.Type == CmdTxnHeartBeat ||
-		req.Type == CmdResolveLock {
+		req.Type == CmdResolveLock ||
+		req.Type == CmdFlashbackToVersion {
 		return true
 	}
 	return false


Test patch:
diff --git a/integration_tests/go.mod b/integration_tests/go.mod
index 731295b..c7abc25 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.20211224045212-9687c2b0f87c
 	github.com/pingcap/failpoint v0.0.0-20220423142525-ae43b7f4e5c3
-	github.com/pingcap/kvproto v0.0.0-20220818023518-a0f02b6efcee
+	github.com/pingcap/kvproto v0.0.0-20220902075447-f95ac338b331
 	github.com/pingcap/tidb v1.1.0-beta.0.20220902042024-0482b2e83ed2
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.8.0
diff --git a/integration_tests/go.sum b/integration_tests/go.sum
index e36bedd..e9dbbd6 100644
--- a/integration_tests/go.sum
+++ b/integration_tests/go.sum
@@ -303,6 +303,8 @@ github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989/go.mod h1:O17Xtb
 github.com/pingcap/kvproto v0.0.0-20220510035547-0e2f26c0a46a/go.mod h1:OYtxs0786qojVTmkVeufx93xe+jUgm56GUYRIKnmaGI=
 github.com/pingcap/kvproto v0.0.0-20220818023518-a0f02b6efcee h1:s1al2ci3MEj5VnNuUCGAfeqpbCxcMeZibOyxw8ClHLE=
 github.com/pingcap/kvproto v0.0.0-20220818023518-a0f02b6efcee/go.mod h1:OYtxs0786qojVTmkVeufx93xe+jUgm56GUYRIKnmaGI=
+github.com/pingcap/kvproto v0.0.0-20220902075447-f95ac338b331 h1:LLPNqaM0C0hHER4U+cl17pvVBFuSDBbVl2oJQ6f6bg4=
+github.com/pingcap/kvproto v0.0.0-20220902075447-f95ac338b331/go.mod h1:OYtxs0786qojVTmkVeufx93xe+jUgm56GUYRIKnmaGI=
 github.com/pingcap/log v0.0.0-20200511115504-543df19646ad/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
 github.com/pingcap/log v0.0.0-20211215031037-e024ba4eb0ee/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
 github.com/pingcap/log v1.1.0 h1:ELiPxACz7vdo1qAvvaWJg1NrYFoY6gqAh/+Uo6aXdD8=
diff --git a/internal/locate/region_request_test.go b/internal/locate/region_request_test.go
index ebbbb46..61b9170 100644
--- a/internal/locate/region_request_test.go
+++ b/internal/locate/region_request_test.go
@@ -470,6 +470,10 @@ func (s *mockTikvGrpcServer) GetLockWaitHistory(ctx context.Context, request *kv
 	return nil, errors.New("unreachable")
 }
 
+func (s *mockTikvGrpcServer) KvFlashbackToVersion(context.Context, *kvrpcpb.FlashbackToVersionRequest) (*kvrpcpb.FlashbackToVersionResponse, error) {
+	return nil, errors.New("unreachable")
+}
+
 func (s *testRegionRequestToSingleStoreSuite) TestNoReloadRegionForGrpcWhenCtxCanceled() {
 	// prepare a mock tikv grpc server
 	addr := "localhost:56341"