rocksdb/
compaction_filter_factory.rs

1use std::ffi::CStr;
2
3use libc::{self, c_char, c_void};
4
5use crate::{
6    compaction_filter::{self, CompactionFilter},
7    ffi,
8};
9
10/// Each compaction will create a new CompactionFilter allowing the
11/// application to know about different compactions.
12///
13///  See [compaction_filter::CompactionFilter][CompactionFilter] and
14///  [Options::set_compaction_filter_factory][set_compaction_filter_factory]
15///  for more details
16///
17///  [CompactionFilter]: ../compaction_filter/trait.CompactionFilter.html
18///  [set_compaction_filter_factory]: ../struct.Options.html#method.set_compaction_filter_factory
19pub trait CompactionFilterFactory {
20    type Filter: CompactionFilter;
21
22    /// Returns a CompactionFilter for the compaction process
23    fn create(&mut self, context: CompactionFilterContext) -> Self::Filter;
24
25    /// Returns a name that identifies this compaction filter factory.
26    fn name(&self) -> &CStr;
27}
28
29pub unsafe extern "C" fn destructor_callback<F>(raw_self: *mut c_void)
30where
31    F: CompactionFilterFactory,
32{
33    drop(unsafe { Box::from_raw(raw_self as *mut F) });
34}
35
36pub unsafe extern "C" fn name_callback<F>(raw_self: *mut c_void) -> *const c_char
37where
38    F: CompactionFilterFactory,
39{
40    let self_ = unsafe { &*(raw_self.cast_const() as *const F) };
41    self_.name().as_ptr()
42}
43
44/// Context information of a compaction run
45pub struct CompactionFilterContext {
46    /// Does this compaction run include all data files
47    pub is_full_compaction: bool,
48    /// Is this compaction requested by the client (true),
49    /// or is it occurring as an automatic compaction process
50    pub is_manual_compaction: bool,
51}
52
53impl CompactionFilterContext {
54    unsafe fn from_raw(ptr: *mut ffi::rocksdb_compactionfiltercontext_t) -> Self {
55        let is_full_compaction =
56            unsafe { ffi::rocksdb_compactionfiltercontext_is_full_compaction(ptr) } != 0;
57        let is_manual_compaction =
58            unsafe { ffi::rocksdb_compactionfiltercontext_is_manual_compaction(ptr) } != 0;
59
60        Self {
61            is_full_compaction,
62            is_manual_compaction,
63        }
64    }
65}
66
67pub unsafe extern "C" fn create_compaction_filter_callback<F>(
68    raw_self: *mut c_void,
69    context: *mut ffi::rocksdb_compactionfiltercontext_t,
70) -> *mut ffi::rocksdb_compactionfilter_t
71where
72    F: CompactionFilterFactory,
73{
74    let self_ = unsafe { &mut *(raw_self as *mut F) };
75    let context = unsafe { CompactionFilterContext::from_raw(context) };
76    let filter = Box::new(self_.create(context));
77
78    let filter_ptr = Box::into_raw(filter);
79
80    unsafe {
81        ffi::rocksdb_compactionfilter_create(
82            filter_ptr as *mut c_void,
83            Some(compaction_filter::destructor_callback::<F::Filter>),
84            Some(compaction_filter::filter_callback::<F::Filter>),
85            Some(compaction_filter::name_callback::<F::Filter>),
86        )
87    }
88}
89
90#[cfg(test)]
91mod tests {
92    use super::*;
93    use crate::compaction_filter::Decision;
94    use crate::{Options, DB};
95    use std::ffi::CString;
96
97    struct CountFilter(u16, CString);
98    impl CompactionFilter for CountFilter {
99        fn filter(&mut self, _level: u32, _key: &[u8], _value: &[u8]) -> crate::CompactionDecision {
100            self.0 += 1;
101            if self.0 > 2 {
102                Decision::Remove
103            } else {
104                Decision::Keep
105            }
106        }
107
108        fn name(&self) -> &CStr {
109            &self.1
110        }
111    }
112
113    struct TestFactory(CString);
114    impl CompactionFilterFactory for TestFactory {
115        type Filter = CountFilter;
116
117        fn create(&mut self, _context: CompactionFilterContext) -> Self::Filter {
118            CountFilter(0, CString::new("CountFilter").unwrap())
119        }
120
121        fn name(&self) -> &CStr {
122            &self.0
123        }
124    }
125
126    #[test]
127    fn compaction_filter_factory_test() {
128        let tempdir = tempfile::Builder::new()
129            .prefix("_rust_rocksdb_filter_factory_test")
130            .tempdir()
131            .expect("Failed to create temporary path for the _rust_rocksdb_filter_factory_test.");
132        let path = tempdir.path();
133        let mut opts = Options::default();
134        opts.create_if_missing(true);
135        opts.set_compaction_filter_factory(TestFactory(CString::new("TestFactory").unwrap()));
136        {
137            let db = DB::open(&opts, path).unwrap();
138            let _r = db.put(b"k1", b"a");
139            let _r = db.put(b"_rk", b"b");
140            let _r = db.put(b"%k", b"c");
141            db.compact_range(None::<&[u8]>, None::<&[u8]>);
142            assert_eq!(db.get(b"%k1").unwrap(), None);
143        }
144        let result = DB::destroy(&opts, path);
145        assert!(result.is_ok());
146    }
147}
rocksdb/compaction_filter_factory.rs

rocksdb/
compaction_filter_factory.rs
