*: Add TruncateManager and TruncateWorker

cmd/tikv-ctl/src/executor.rs

@@ -633,6 +633,8 @@ pub trait DebugExecutor {
     fn dump_store_info(&self);
 
     fn dump_cluster_info(&self);
+
+    fn reset_to_version(&self, version: u64);
 }
 
 impl DebugExecutor for DebugClient {
@@ -840,6 +842,13 @@ impl DebugExecutor for DebugClient {
             .unwrap_or_else(|e| perror_and_exit("DebugClient::get_cluster_info", e));
         println!("{}", resp.get_cluster_id())
     }
+
+    fn reset_to_version(&self, version: u64) {
+        let mut req = ResetToVersionRequest::default();
+        req.set_ts(version);
+        DebugClient::reset_to_version(self, &req)
+            .unwrap_or_else(|e| perror_and_exit("DebugClient::get_cluster_info", e));
+    }
 }
 
 impl<ER: RaftEngine> DebugExecutor for Debugger<ER> {
@@ -1070,6 +1079,10 @@ impl<ER: RaftEngine> DebugExecutor for Debugger<ER> {
             println!("cluster id: {}", ident.get_cluster_id());
         }
     }
+
+    fn reset_to_version(&self, version: u64) {
+        Debugger::reset_to_version(self, version);
+    }
 }
 
 fn handle_engine_error(err: EngineError) -> ! {

src/server/debug.rs

@@ -34,6 +34,7 @@ use tikv_util::worker::Worker;
 use txn_types::Key;
 
 use crate::config::ConfigController;
+use crate::server::reset_to_version::ResetToVersionManager;
 use crate::storage::mvcc::{Lock, LockType, TimeStamp, Write, WriteRef, WriteType};
 
 pub use crate::storage::mvcc::MvccInfoIterator;
@@ -117,6 +118,7 @@ impl From<BottommostLevelCompaction> for debugpb::BottommostLevelCompaction {
 #[derive(Clone)]
 pub struct Debugger<ER: RaftEngine> {
     engines: Engines<RocksEngine, ER>,
+    reset_to_version_manager: ResetToVersionManager,
     cfg_controller: ConfigController,
 }
 
@@ -125,8 +127,10 @@ impl<ER: RaftEngine> Debugger<ER> {
         engines: Engines<RocksEngine, ER>,
         cfg_controller: ConfigController,
     ) -> Debugger<ER> {
+        let reset_to_version_manager = ResetToVersionManager::new(engines.kv.clone());
         Debugger {
             engines,
+            reset_to_version_manager,
             cfg_controller,
         }
     }
@@ -884,6 +888,10 @@ impl<ER: RaftEngine> Debugger<ER> {
         props.append(&mut props1);
         Ok(props)
     }
+
+    pub fn reset_to_version(&self, version: u64) {
+        self.reset_to_version_manager.start(version.into());
+    }
 }
 
 fn dump_default_cf_properties(
@@ -899,7 +907,6 @@ fn dump_default_cf_properties(
     for (_, v) in collection.iter() {
         num_entries += v.num_entries();
     }
-
     let sst_files = collection
         .iter()
         .map(|(k, _)| {

src/server/mod.rs

@@ -12,6 +12,7 @@ pub mod lock_manager;
 pub mod node;
 mod proxy;
 pub mod raftkv;
+mod reset_to_version;
 pub mod resolve;
 pub mod server;
 pub mod service;