Add compact on deletion support (#535)

* add compact on deletion support
Signed-off-by: Connor1996 <zbk602423539@gmail.com>

librocksdb_sys/crocksdb/c.cc

@@ -46,6 +46,7 @@
 #include "rocksdb/utilities/db_ttl.h"
 #include "rocksdb/utilities/debug.h"
 #include "rocksdb/utilities/options_util.h"
+#include "rocksdb/utilities/table_properties_collectors.h"
 #include "rocksdb/write_batch.h"
 #include "src/blob_format.h"
 #include "table/block_based/block_based_table_factory.h"
@@ -4903,6 +4904,14 @@ void crocksdb_options_add_table_properties_collector_factory(
       std::shared_ptr<TablePropertiesCollectorFactory>(f));
 }
 
+void crocksdb_options_set_compact_on_deletion(crocksdb_options_t* opt,
+                                              size_t sliding_window_size,
+                                              size_t deletion_trigger) {
+  opt->rep.table_properties_collector_factories.push_back(
+      rocksdb::NewCompactOnDeletionCollectorFactory(sliding_window_size,
+                                                    deletion_trigger));
+}
+
 /* Get Table Properties */
 
 crocksdb_table_properties_collection_t* crocksdb_get_properties_of_all_tables(

librocksdb_sys/crocksdb/crocksdb/c.h

@@ -1954,6 +1954,10 @@ extern C_ROCKSDB_LIBRARY_API void
 crocksdb_options_add_table_properties_collector_factory(
     crocksdb_options_t* opt, crocksdb_table_properties_collector_factory_t* f);
 
+extern C_ROCKSDB_LIBRARY_API void crocksdb_options_set_compact_on_deletion(
+    crocksdb_options_t* opt, size_t sliding_window_size,
+    size_t deletion_trigger);
+
 /* Get Table Properties */
 extern C_ROCKSDB_LIBRARY_API crocksdb_table_properties_collection_t*
 crocksdb_get_properties_of_all_tables(crocksdb_t* db, char** errptr);

librocksdb_sys/src/lib.rs

@@ -1991,6 +1991,12 @@ extern "C" {
         f: *mut DBTablePropertiesCollectorFactory,
     );
 
+    pub fn crocksdb_options_set_compact_on_deletion(
+        options: *mut Options,
+        sliding_window_size: size_t,
+        deletion_trigger: size_t,
+    );
+
     pub fn crocksdb_get_properties_of_all_tables(
         db: *mut DBInstance,
         errptr: *mut *mut c_char,

src/rocksdb_options.rs

@@ -1748,6 +1748,16 @@ impl ColumnFamilyOptions {
             crocksdb_ffi::crocksdb_options_set_sst_partitioner_factory(self.inner, f);
         }
     }
+
+    pub fn set_compact_on_deletion(&self, sliding_window_size: usize, deletion_trigger: usize) {
+        unsafe {
+            crocksdb_ffi::crocksdb_options_set_compact_on_deletion(
+                self.inner,
+                sliding_window_size,
+                deletion_trigger,
+            );
+        }
+    }
 }
 
 // ColumnFamilyDescriptor is a pair of column family's name and options.

tests/cases/test_rocksdb_options.rs

@@ -847,3 +847,49 @@ fn test_dboptions_set_env() {
     opts.set_env(Arc::new(Env::default()));
     let _db = DB::open(opts, path_str).unwrap();
 }
+
+#[test]
+fn test_compact_on_deletion() {
+    let num_keys = 1000;
+    let window_size = 100;
+    let dels_trigger = 90;
+
+    let mut opts = DBOptions::new();
+    let cf_opts = ColumnFamilyOptions::new();
+    opts.create_if_missing(true);
+    cf_opts.set_compact_on_deletion(window_size, dels_trigger);
+
+    let path = tempdir_with_prefix("_rust_rocksdb_compact_on_deletion_test");
+    let db = DB::open_cf(
+        opts,
+        path.path().to_str().unwrap(),
+        vec![("default", cf_opts)],
+    )
+    .unwrap();
+
+    let cf = db.cf_handle("default").unwrap();
+    db.put(b"key0", b"value").unwrap();
+    db.flush(true).unwrap();
+    let mut opt = CompactOptions::new();
+    opt.set_change_level(true);
+    opt.set_target_level(1);
+    db.compact_range_cf_opt(cf, &opt, None, None);
+
+    let name = format!("rocksdb.num-files-at-level{}", 1);
+    assert_eq!(db.get_property_int(&name).unwrap(), 1);
+
+    for i in 0..num_keys {
+        if i >= num_keys - window_size && i < num_keys - window_size + dels_trigger {
+            db.delete(format!("key{}", i).as_ref()).unwrap();
+        } else {
+            db.put(format!("key{}", i).as_ref(), b"value").unwrap();
+        }
+    }
+    db.flush(true).unwrap();
+    std::thread::sleep(std::time::Duration::from_millis(100));
+
+    let name = format!("rocksdb.num-files-at-level{}", 0);
+    assert_eq!(db.get_property_int(&name).unwrap(), 0);
+    let name = format!("rocksdb.num-files-at-level{}", 1);
+    assert_eq!(db.get_property_int(&name).unwrap(), 1);
+}