From 2ed4d80cf312cbb84d7c7953365649fdf9b8b92a Mon Sep 17 00:00:00 2001
From: Roman Zeyde <me@romanzey.de>
Date: Mon, 23 Nov 2020 10:43:50 +0200
Subject: [PATCH] Allow setting options on a ColumnFamily (#478)

---
 src/db.rs        | 53 ++++++++++++++++++++++++++++++++++--------------
 tests/test_db.rs | 41 +++++++++++++++++++++++++++++++++++++
 2 files changed, 79 insertions(+), 15 deletions(-)

diff --git a/src/db.rs b/src/db.rs
index 9443146..754459a 100644
--- a/src/db.rs
+++ b/src/db.rs
@@ -1005,21 +1005,7 @@ impl DB {
     }
 
     pub fn set_options(&self, opts: &[(&str, &str)]) -> Result<(), Error> {
-        let copts = opts
-            .iter()
-            .map(|(name, value)| {
-                let cname = match CString::new(name.as_bytes()) {
-                    Ok(cname) => cname,
-                    Err(e) => return Err(Error::new(format!("Invalid option name `{}`", e))),
-                };
-                let cvalue = match CString::new(value.as_bytes()) {
-                    Ok(cvalue) => cvalue,
-                    Err(e) => return Err(Error::new(format!("Invalid option value: `{}`", e))),
-                };
-                Ok((cname, cvalue))
-            })
-            .collect::<Result<Vec<(CString, CString)>, Error>>()?;
-
+        let copts = convert_options(opts)?;
         let cnames: Vec<*const c_char> = copts.iter().map(|opt| opt.0.as_ptr()).collect();
         let cvalues: Vec<*const c_char> = copts.iter().map(|opt| opt.1.as_ptr()).collect();
         let count = opts.len() as i32;
@@ -1034,6 +1020,27 @@ impl DB {
         Ok(())
     }
 
+    pub fn set_options_cf(
+        &self,
+        cf_handle: &ColumnFamily,
+        opts: &[(&str, &str)],
+    ) -> Result<(), Error> {
+        let copts = convert_options(opts)?;
+        let cnames: Vec<*const c_char> = copts.iter().map(|opt| opt.0.as_ptr()).collect();
+        let cvalues: Vec<*const c_char> = copts.iter().map(|opt| opt.1.as_ptr()).collect();
+        let count = opts.len() as i32;
+        unsafe {
+            ffi_try!(ffi::rocksdb_set_options_cf(
+                self.inner,
+                cf_handle.inner,
+                count,
+                cnames.as_ptr(),
+                cvalues.as_ptr(),
+            ));
+        }
+        Ok(())
+    }
+
     /// Retrieves a RocksDB property by name.
     ///
     /// Full list of properties could be find
@@ -1405,3 +1412,19 @@ pub struct LiveFile {
     /// Number of deletions/tomb key(s) in the file
     pub num_deletions: u64,
 }
+
+fn convert_options(opts: &[(&str, &str)]) -> Result<Vec<(CString, CString)>, Error> {
+    opts.iter()
+        .map(|(name, value)| {
+            let cname = match CString::new(name.as_bytes()) {
+                Ok(cname) => cname,
+                Err(e) => return Err(Error::new(format!("Invalid option name `{}`", e))),
+            };
+            let cvalue = match CString::new(value.as_bytes()) {
+                Ok(cvalue) => cvalue,
+                Err(e) => return Err(Error::new(format!("Invalid option value: `{}`", e))),
+            };
+            Ok((cname, cvalue))
+        })
+        .collect()
+}
diff --git a/tests/test_db.rs b/tests/test_db.rs
index e74c18b..8006341 100644
--- a/tests/test_db.rs
+++ b/tests/test_db.rs
@@ -336,6 +336,47 @@ fn set_option_test() {
     }
 }
 
+#[test]
+fn set_option_cf_test() {
+    let path = DBPath::new("_rust_rocksdb_set_options_cftest");
+    {
+        let mut opts = Options::default();
+        opts.create_if_missing(true);
+        opts.create_missing_column_families(true);
+        let db = DB::open_cf(&opts, &path, vec!["cf1"]).unwrap();
+        let cf = db.cf_handle("cf1").unwrap();
+        // set an option to valid values
+        assert!(db
+            .set_options_cf(cf, &[("disable_auto_compactions", "true")])
+            .is_ok());
+        assert!(db
+            .set_options_cf(cf, &[("disable_auto_compactions", "false")])
+            .is_ok());
+        // invalid names/values should result in an error
+        assert!(db
+            .set_options_cf(cf, &[("disable_auto_compactions", "INVALID_VALUE")])
+            .is_err());
+        assert!(db
+            .set_options_cf(cf, &[("INVALID_NAME", "INVALID_VALUE")])
+            .is_err());
+        // option names/values must not contain NULLs
+        assert!(db
+            .set_options_cf(cf, &[("disable_auto_compactions", "true\0")])
+            .is_err());
+        assert!(db
+            .set_options_cf(cf, &[("disable_auto_compactions\0", "true")])
+            .is_err());
+        // empty options are not allowed
+        assert!(db.set_options_cf(cf, &[]).is_err());
+        // multiple options can be set in a single API call
+        let multiple_options = [
+            ("paranoid_file_checks", "true"),
+            ("report_bg_io_stats", "true"),
+        ];
+        db.set_options(&multiple_options).unwrap();
+    }
+}
+
 #[test]
 fn test_sequence_number() {
     let path = DBPath::new("_rust_rocksdb_test_sequence_number");