Add delete_range to OptimisticTransactionDB (#879)

2 files changed, 71 insertions, 5 deletions

diff --git a/src/transactions/optimistic_transaction_db.rs b/src/transactions/optimistic_transaction_db.rs
index 5642594..5ce3bff 100644
--- a/src/transactions/optimistic_transaction_db.rs
+++ b/src/transactions/optimistic_transaction_db.rs
@@ -15,12 +15,15 @@
 
 use std::{collections::BTreeMap, ffi::CString, fs, iter, marker::PhantomData, path::Path, ptr};
 
-use libc::{c_char, c_int};
+use libc::{c_char, c_int, size_t};
 
 use crate::{
-    db::DBCommon, db::DBInner, ffi, ffi_util::to_cpath, write_batch::WriteBatchWithTransaction,
-    ColumnFamilyDescriptor, Error, OptimisticTransactionOptions, Options, ThreadMode, Transaction,
-    WriteOptions, DEFAULT_COLUMN_FAMILY_NAME,
+    db::{DBCommon, DBInner},
+    ffi,
+    ffi_util::to_cpath,
+    write_batch::WriteBatchWithTransaction,
+    AsColumnFamilyRef, ColumnFamilyDescriptor, Error, OptimisticTransactionOptions, Options,
+    ThreadMode, Transaction, WriteOptions, DEFAULT_COLUMN_FAMILY_NAME,
 };
 
 /// A type alias to RocksDB Optimistic Transaction DB.
@@ -42,7 +45,7 @@ use crate::{
 /// {
 ///     let db: OptimisticTransactionDB = OptimisticTransactionDB::open_default(path).unwrap();
 ///     db.put(b"my key", b"my value").unwrap();
-///     
+///
 ///     // create transaction
 ///     let txn = db.transaction();
 ///     txn.put(b"key2", b"value2");
@@ -290,4 +293,39 @@ impl<T: ThreadMode> OptimisticTransactionDB<T> {
         wo.disable_wal(true);
         self.write_opt(batch, &wo)
     }
+
+    /// Removes the database entries in the range `["from", "to")` using given write options.
+    pub fn delete_range_cf_opt<K: AsRef<[u8]>>(
+        &self,
+        cf: &impl AsColumnFamilyRef,
+        from: K,
+        to: K,
+        writeopts: &WriteOptions,
+    ) -> Result<(), Error> {
+        let from = from.as_ref();
+        let to = to.as_ref();
+
+        unsafe {
+            ffi_try!(ffi::rocksdb_delete_range_cf(
+                self.inner.inner(),
+                writeopts.inner,
+                cf.inner(),
+                from.as_ptr() as *const c_char,
+                from.len() as size_t,
+                to.as_ptr() as *const c_char,
+                to.len() as size_t,
+            ));
+            Ok(())
+        }
+    }
+
+    /// Removes the database entries in the range `["from", "to")` using default write options.
+    pub fn delete_range_cf<K: AsRef<[u8]>>(
+        &self,
+        cf: &impl AsColumnFamilyRef,
+        from: K,
+        to: K,
+    ) -> Result<(), Error> {
+        self.delete_range_cf_opt(cf, from, to, &WriteOptions::default())
+    }
 }
diff --git a/tests/test_optimistic_transaction_db.rs b/tests/test_optimistic_transaction_db.rs
index 61c86de..036ab92 100644
--- a/tests/test_optimistic_transaction_db.rs
+++ b/tests/test_optimistic_transaction_db.rs
@@ -581,3 +581,31 @@ fn transaction_snapshot() {
         assert_eq!(snapshot.get(b"k3").unwrap().unwrap(), b"v3");
     }
 }
+
+#[test]
+fn delete_range_test() {
+    let path = DBPath::new("_rust_rocksdb_optimistic_transaction_db_delete_range_test");
+    {
+        let mut opts = Options::default();
+        opts.create_if_missing(true);
+        opts.create_missing_column_families(true);
+
+        let cfs = vec!["cf1"];
+        let db: OptimisticTransactionDB =
+            OptimisticTransactionDB::open_cf(&opts, &path, cfs).unwrap();
+
+        let cf1 = db.cf_handle("cf1").unwrap();
+        db.put_cf(&cf1, b"k1", b"v1").unwrap();
+        db.put_cf(&cf1, b"k2", b"v2").unwrap();
+        db.put_cf(&cf1, b"k3", b"v3").unwrap();
+        db.put_cf(&cf1, b"k4", b"v4").unwrap();
+        db.put_cf(&cf1, b"k5", b"v5").unwrap();
+
+        db.delete_range_cf(&cf1, b"k2", b"k4").unwrap();
+        assert_eq!(db.get_cf(&cf1, b"k1").unwrap().unwrap(), b"v1");
+        assert_eq!(db.get_cf(&cf1, b"k4").unwrap().unwrap(), b"v4");
+        assert_eq!(db.get_cf(&cf1, b"k5").unwrap().unwrap(), b"v5");
+        assert!(db.get_cf(&cf1, b"k2").unwrap().is_none());
+        assert!(db.get_cf(&cf1, b"k3").unwrap().is_none());
+    }
+}