raftstore-v2: implement local read for raftstore-v2

components/raftstore-v2/src/operation/query/local.rs

@@ -157,7 +157,7 @@ mod tests {
     use engine_traits::{OpenOptions, Peekable, SyncMutable, ALL_CFS, CF_DEFAULT};
     use kvproto::{metapb::Region, raft_cmdpb::*};
     use raftstore::store::{
-        util::Lease, Callback, CasualMessage, CasualRouter, LocalReader, ProposalRouter,
+        util::Lease, Callback, CasualMessage, CasualRouter, LocalReaderCore, ProposalRouter,
         RaftCommand,
     };
     use tempfile::{Builder, TempDir};

components/raftstore/src/router.rs

@@ -13,8 +13,8 @@ use crate::{
     store::{
         fsm::RaftRouter,
         transport::{CasualRouter, ProposalRouter, SignificantRouter},
-        CachedReadDelegate, Callback, CasualMessage, LocalReader, PeerMsg, RaftCmdExtraOpts,
-        RaftCommand, SignificantMsg, StoreMetaDelegate, StoreMsg, StoreRouter,
+        CachedReadDelegate, Callback, CasualMessage, LocalReaderCore, LocalReader, PeerMsg,
+        RaftCmdExtraOpts, RaftCommand, SignificantMsg, StoreMetaDelegate, StoreMsg, StoreRouter,
     },
     DiscardReason, Error as RaftStoreError, Result as RaftStoreResult,
 };
@@ -174,8 +174,9 @@ where
     ER: RaftEngine,
 {
     router: RaftRouter<EK, ER>,
-    local_reader:
-        RefCell<LocalReader<RaftRouter<EK, ER>, EK, CachedReadDelegate<EK>, StoreMetaDelegate<EK>>>,
+    local_reader: RefCell<
+        LocalReader<RaftRouter<EK, ER>, EK, CachedReadDelegate<EK>, StoreMetaDelegate<EK>>,
+    >,
 }
 
 impl<EK, ER> Clone for ServerRaftStoreRouter<EK, ER>
@@ -195,7 +196,12 @@ impl<EK: KvEngine, ER: RaftEngine> ServerRaftStoreRouter<EK, ER> {
     /// Creates a new router.
     pub fn new(
         router: RaftRouter<EK, ER>,
-        reader: LocalReader<RaftRouter<EK, ER>, EK, CachedReadDelegate<EK>, StoreMetaDelegate<EK>>,
+        reader: LocalReader<
+            RaftRouter<EK, ER>,
+            EK,
+            CachedReadDelegate<EK>,
+            StoreMetaDelegate<EK>,
+        >,
     ) -> ServerRaftStoreRouter<EK, ER> {
         let local_reader = RefCell::new(reader);
         ServerRaftStoreRouter {

components/raftstore/src/store/mod.rs

@@ -69,9 +69,9 @@ pub use self::{
     worker::{
         AutoSplitController, Bucket, BucketRange, CachedReadDelegate, CheckLeaderRunner,
         CheckLeaderTask, FetchedLogs, FlowStatistics, FlowStatsReporter, KeyEntry,
-        LocalReadContext, LocalReader, LogFetchedNotifier, PdTask, QueryStats, RaftlogFetchRunner,
-        RaftlogFetchTask, ReadDelegate, ReadExecutor, ReadExecutorProvider, ReadProgress,
-        ReadStats, RefreshConfigTask, RegionTask, SplitCheckRunner, SplitCheckTask, SplitConfig,
-        SplitConfigManager, StoreMetaDelegate, TrackVer, WriteStats,
+        LocalReadContext, LocalReaderCore, LocalReader, LogFetchedNotifier, PdTask, QueryStats,
+        RaftlogFetchRunner, RaftlogFetchTask, ReadDelegate, ReadExecutor, ReadExecutorProvider,
+        ReadProgress, ReadStats, RefreshConfigTask, RegionTask, SplitCheckRunner, SplitCheckTask,
+        SplitConfig, SplitConfigManager, StoreMetaDelegate, TrackVer, WriteStats,
     },
 };

components/raftstore/src/store/msg.rs

@@ -13,12 +13,13 @@ use kvproto::{
     metapb,
     metapb::RegionEpoch,
     pdpb::{self, CheckPolicy},
-    raft_cmdpb::{RaftCmdRequest, RaftCmdResponse},
+    raft_cmdpb::{RaftCmdRequest, RaftCmdResponse, Response},
     raft_serverpb::RaftMessage,
     replication_modepb::ReplicationStatus,
 };
 #[cfg(any(test, feature = "testexport"))]
 use pd_client::BucketMeta;
+use protobuf::RepeatedField;
 use raft::SnapshotStatus;
 use smallvec::{smallvec, SmallVec};
 use tikv_util::{deadline::Deadline, escape, memory::HeapSize, time::Instant};
@@ -37,6 +38,70 @@ use crate::store::{
     SnapKey,
 };
 
+pub trait ReadResponseTrait<S: Snapshot>: Default {
+    type Response;
+
+    fn set_term(&mut self, _term: u64) {
+        unimplemented!()
+    }
+
+    fn set_response(&mut self, response: Self::Response);
+
+    fn set_responses(&mut self, _response: RepeatedField<Response>) {
+        unimplemented!()
+    }
+
+    fn mut_snapshot(&mut self) -> Option<&mut RegionSnapshot<S>> {
+        unimplemented!()
+    }
+
+    fn set_snapshot(&mut self, _snapshot: RegionSnapshot<S>) {
+        unimplemented!()
+    }
+
+    fn set_txn_extra_op(&mut self, _txn_extra_op: TxnExtraOp) {
+        unimplemented!()
+    }
+
+    fn set_error(&mut self, error: RaftCmdResponse);
+}
+
+impl<S: Snapshot> ReadResponseTrait<S> for ReadResponse<S> {
+    type Response = RaftCmdResponse;
+
+    fn set_term(&mut self, term: u64) {
+        if term == 0 {
+            return;
+        }
+
+        self.response.mut_header().set_current_term(term);
+    }
+
+    fn set_response(&mut self, response: Self::Response) {
+        self.response = response;
+    }
+
+    fn set_responses(&mut self, response: RepeatedField<Response>) {
+        self.response.set_responses(response);
+    }
+
+    fn mut_snapshot(&mut self) -> Option<&mut RegionSnapshot<S>> {
+        self.snapshot.as_mut()
+    }
+
+    fn set_snapshot(&mut self, snapshot: RegionSnapshot<S>) {
+        self.snapshot = Some(snapshot);
+    }
+
+    fn set_txn_extra_op(&mut self, txn_extra_op: TxnExtraOp) {
+        self.txn_extra_op = txn_extra_op;
+    }
+
+    fn set_error(&mut self, error: RaftCmdResponse) {
+        self.response = error;
+    }
+}
+
 #[derive(Debug)]
 pub struct ReadResponse<S: Snapshot> {
     pub response: RaftCmdResponse,
@@ -73,6 +138,19 @@ where
     }
 }
 
+impl<S> Default for ReadResponse<S>
+where
+    S: Snapshot,
+{
+    fn default() -> Self {
+        ReadResponse {
+            response: RaftCmdResponse::default(),
+            snapshot: None,
+            txn_extra_op: TxnExtraOp::Noop,
+        }
+    }
+}
+
 pub type BoxReadCallback<S> = Box<dyn FnOnce(ReadResponse<S>) + Send>;
 pub type BoxWriteCallback = Box<dyn FnOnce(WriteResponse) + Send>;
 pub type ExtCallback = Box<dyn FnOnce() + Send>;
@@ -201,8 +279,8 @@ where
     }
 }
 
-pub trait ReadCallback: ErrorCallback {
-    type Response;
+pub trait ReadCallback<S: Snapshot>: ErrorCallback {
+    type Response: ReadResponseTrait<S>;
 
     fn set_result(self, result: Self::Response);
 }
@@ -222,7 +300,7 @@ pub trait ErrorCallback: Send {
     fn is_none(&self) -> bool;
 }
 
-impl<S: Snapshot> ReadCallback for Callback<S> {
+impl<S: Snapshot> ReadCallback<S> for Callback<S> {
     type Response = ReadResponse<S>;
 
     #[inline]

components/raftstore/src/store/peer.rs

@@ -5476,6 +5476,8 @@ where
     EK: KvEngine,
     ER: RaftEngine,
 {
+    type Response = ReadResponse<EK::Snapshot>;
+
     fn get_tablet(&mut self) -> &EK {
         &self.engines.kv
     }

components/raftstore/src/store/worker/mod.rs

@@ -35,8 +35,9 @@ pub use self::{
     },
     raftlog_gc::{Runner as RaftlogGcRunner, Task as RaftlogGcTask},
     read::{
-        CachedReadDelegate, LocalReadContext, LocalReader, Progress as ReadProgress, ReadDelegate,
-        ReadExecutor, ReadExecutorProvider, StoreMetaDelegate, TrackVer,
+        CachedReadDelegate, LocalReadContext, LocalReaderCore, LocalReader,
+        Progress as ReadProgress, ReadDelegate, ReadExecutor, ReadExecutorProvider,
+        StoreMetaDelegate, TrackVer,
     },
     refresh_config::{
         BatchComponent as RaftStoreBatchComponent, Runner as RefreshConfigRunner,