debug: add a new API Compact.

src/raftstore/store/debug.rs

@@ -17,6 +17,7 @@ use kvproto::{eraftpb, raft_serverpb};
 
 use raftstore::store::{keys, Engines, Iterable, Peekable};
 use storage::{CF_DEFAULT, CF_LOCK, CF_RAFT, CF_WRITE};
+use util::rocksdb::{compact_range, get_cf_handle};
 
 pub type Result<T> = result::Result<T, Error>;
 
@@ -151,6 +152,19 @@ impl Debugger {
             Err(e) => Err(box_err!(e)),
         }
     }
+
+    /// Compact the cf[start..end) in the db **by manual**.
+    pub fn compact(&self, db: DB, cf: &str, start: &[u8], end: &[u8]) -> Result<()> {
+        try!(validate_db_and_cf(db, cf));
+        let db = match db {
+            DB::KV => &self.engines.kv_engine,
+            DB::RAFT => &self.engines.raft_engine,
+            _ => unreachable!(),
+        };
+        let handle = box_try!(get_cf_handle(db, cf));
+        compact_range(db, handle, Some(start), Some(end), true);
+        Ok(())
+    }
 }
 
 pub fn validate_db_and_cf(db: DB, cf: &str) -> Result<()> {

src/raftstore/store/worker/compact.rs

@@ -14,8 +14,9 @@
 use util::worker::Runnable;
 use util::rocksdb;
 use util::escape;
+use util::rocksdb::compact_range;
 
-use rocksdb::{CompactOptions, DB};
+use rocksdb::DB;
 use std::sync::Arc;
 use std::fmt::{self, Display, Formatter};
 use std::error;
@@ -41,7 +42,7 @@ impl Display for Task {
 
 quick_error! {
     #[derive(Debug)]
-    enum Error {
+    pub enum Error {
         Other(err: Box<error::Error + Sync + Send>) {
             from()
             cause(err.as_ref())
@@ -60,26 +61,19 @@ impl Runner {
         Runner { engine: engine }
     }
 
-    fn compact_range_cf(
+    pub fn compact_range_cf(
         &mut self,
         cf_name: String,
         start_key: Option<Vec<u8>>,
         end_key: Option<Vec<u8>>,
     ) -> Result<(), Error> {
-        let cf_handle = box_try!(rocksdb::get_cf_handle(&self.engine, &cf_name));
+        let handle = box_try!(rocksdb::get_cf_handle(&self.engine, &cf_name));
         let compact_range_timer = COMPACT_RANGE_CF
             .with_label_values(&[&cf_name])
             .start_coarse_timer();
-        let mut compact_opts = CompactOptions::new();
-        // manual compaction can concurrently run with background compaction threads.
-        compact_opts.set_exclusive_manual_compaction(false);
-        self.engine.compact_range_cf_opt(
-            cf_handle,
-            &compact_opts,
-            start_key.as_ref().map(Vec::as_slice),
-            end_key.as_ref().map(Vec::as_slice),
-        );
-
+        let start = start_key.as_ref().map(Vec::as_slice);
+        let end = end_key.as_ref().map(Vec::as_slice);
+        compact_range(&self.engine, handle, start, end, false);
         compact_range_timer.observe_duration();
         Ok(())
     }

src/server/service/debug.rs

@@ -180,4 +180,19 @@ impl debugpb_grpc::Debug for Service {
     ) {
         unimplemented!()
     }
+
+    fn compact(&self, ctx: RpcContext, req: CompactRequest, sink: UnarySink<CompactResponse>) {
+        let debugger = self.debugger.clone();
+        let f = self.pool.spawn_fn(move || {
+            debugger
+                .compact(
+                    req.get_db(),
+                    req.get_cf(),
+                    req.get_from_key(),
+                    req.get_to_key(),
+                )
+                .map(|_| CompactResponse::default())
+        });
+        self.handle_response(ctx, sink, f, "debug_compact");
+    }
 }

src/util/rocksdb/mod.rs

@@ -25,8 +25,8 @@ use std::sync::Arc;
 use std::str::FromStr;
 
 use storage::{ALL_CFS, CF_DEFAULT, CF_LOCK};
-use rocksdb::{ColumnFamilyOptions, DBCompressionType, DBOptions, ReadOptions, SliceTransform,
-              Writable, WriteBatch, DB};
+use rocksdb::{ColumnFamilyOptions, CompactOptions, DBCompressionType, DBOptions, ReadOptions,
+              SliceTransform, Writable, WriteBatch, DB};
 use rocksdb::rocksdb::supported_compression;
 use util::rocksdb::engine_metrics::{ROCKSDB_COMPRESSION_RATIO_AT_LEVEL,
                                     ROCKSDB_CUR_SIZE_ALL_MEM_TABLES, ROCKSDB_TOTAL_SST_FILES_SIZE};
@@ -348,6 +348,20 @@ pub fn roughly_cleanup_range(db: &DB, start_key: &[u8], end_key: &[u8]) -> Resul
     Ok(())
 }
 
+/// Compact the cf in the specified range by manual or not.
+pub fn compact_range(
+    db: &DB,
+    handle: &CFHandle,
+    start_key: Option<&[u8]>,
+    end_key: Option<&[u8]>,
+    manual: bool,
+) {
+    let mut compact_opts = CompactOptions::new();
+    // manual compaction can concurrently run with background compaction threads.
+    compact_opts.set_exclusive_manual_compaction(manual);
+    db.compact_range_cf_opt(handle, &compact_opts, start_key, end_key);
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;