wip: integration tests with tikv-service

Signed-off-by: iosmanthus <[email protected]>

Author: iosmanthus

src/raw/requests.rs

@@ -1,6 +1,7 @@
 // Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
 
 use std::{any::Any, ops::Range, sync::Arc};
+use std::fmt::Formatter;
 
 use async_trait::async_trait;
 use futures::stream::BoxStream;
@@ -345,6 +346,12 @@ pub struct RawCoprocessorRequest {
     data_builder: RawCoprocessorRequestDataBuilder,
 }
 
+impl std::fmt::Debug for RawCoprocessorRequest {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        Ok(())
+    }
+}
+
 #[async_trait]
 impl Request for RawCoprocessorRequest {
     async fn dispatch(&self, client: &TikvClient, options: CallOption) -> Result<Box<dyn Any>> {

src/request/codec.rs

@@ -74,6 +74,10 @@ pub trait RequestCodecExt: RequestCodec {
         keys.into_iter().map(|key| self.encode_key(key)).collect()
     }
 
+    fn encode_secondaries(&self, secondaries: Vec<Vec<u8>>) -> Vec<Vec<u8>> {
+        self.encode_keys(secondaries)
+    }
+
     fn encode_pairs(&self, mut pairs: Vec<kvrpcpb::KvPair>) -> Vec<kvrpcpb::KvPair> {
         for pair in pairs.iter_mut() {
             *pair.mut_key() = self.encode_key(pair.take_key());
@@ -217,18 +221,18 @@ pub trait Mode: Clone + Copy + Sync + Send + 'static {
     const MAX_KEY: &'static [u8] = &[Self::PREFIX + 1, 0, 0, 0];
 }
 
-#[derive(Clone, Copy)]
+#[derive(Default, Clone, Copy)]
 pub struct RawMode;
 
-#[derive(Clone, Copy)]
+#[derive(Default, Clone, Copy)]
 pub struct TxnMode;
 
 impl Mode for RawMode {
     const PREFIX: u8 = b'r';
 }
 
 impl Mode for TxnMode {
-    const PREFIX: u8 = b't';
+    const PREFIX: u8 = b'x';
 }
 
 #[derive(Clone)]
@@ -263,12 +267,21 @@ impl RequestCodec for ApiV1<TxnMode> {
 
 impl TxnCodec for ApiV1<TxnMode> {}
 
-#[derive(Clone, Copy, Default)]
+#[derive(Clone, Copy)]
 pub struct KeySpace<M: Mode> {
     id: KeySpaceId,
     _phantom: PhantomData<M>,
 }
 
+impl<M: Mode> Default for KeySpace<M> {
+    fn default() -> Self {
+        KeySpace {
+            id: KeySpaceId::default(),
+            _phantom: PhantomData,
+        }
+    }
+}
+
 impl<M: Mode> From<KeySpace<M>> for Prefix {
     fn from(s: KeySpace<M>) -> Self {
         [M::PREFIX, s.id[0], s.id[1], s.id[2]]

src/request/mod.rs

@@ -28,9 +28,9 @@ pub mod codec;
 
 /// Abstracts any request sent to a TiKV server.
 #[async_trait]
-pub trait KvRequest<C>: Request + Sized + Clone + Sync + Send + 'static {
+pub trait KvRequest<C>: Request + Sized + Clone + Sync + Send + std::fmt::Debug + 'static {
     /// The expected response to the request.
-    type Response: HasKeyErrors + HasLocks + HasRegionError + Clone + Send + 'static;
+    type Response: HasKeyErrors + HasLocks + HasRegionError + Clone + Send + std::fmt::Debug +'static;
 
     fn encode_request(self, _codec: &C) -> Self {
         self

src/request/plan.rs

@@ -57,6 +57,10 @@ impl<C: RequestCodec, Req: KvRequest<C>> Plan for Dispatch<C, Req> {
     async fn execute(&self) -> Result<Self::Result> {
         let req = self.codec.encode_request(&self.request);
 
+        let time = std::time::Instant::now();
+
+        println!("{:?}: sending: {:?}", time ,req);
+
         let stats = tikv_stats(self.request.label());
         let result = self
             .kv_client
@@ -71,7 +75,9 @@ impl<C: RequestCodec, Req: KvRequest<C>> Plan for Dispatch<C, Req> {
                 .downcast()
                 .expect("Downcast failed: request and response type mismatch");
 
-            self.codec.decode_response(req.as_ref(), resp)
+            let resp = self.codec.decode_response(req.as_ref(), resp);
+            println!("{:?}: recv(decode): {:?}",time, resp);
+            resp
         })
     }
 }

src/transaction/requests.rs

@@ -233,7 +233,10 @@ pub fn new_pessimistic_prewrite_request(
     req
 }
 
-impl_kv_request!(kvrpcpb::PrewriteRequest, mutations; kvrpcpb::PrewriteResponse, errors);
+impl_kv_request!(
+    kvrpcpb::PrewriteRequest, mutations, primary_lock, secondaries;
+    kvrpcpb::PrewriteResponse, errors
+);
 
 impl Shardable for kvrpcpb::PrewriteRequest {
     type Shard = Vec<kvrpcpb::Mutation>;