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(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_ = &*(raw_self as *const c_void 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 = ffi::rocksdb_compactionfiltercontext_is_full_compaction(ptr) != 0;
56        let is_manual_compaction =
57            ffi::rocksdb_compactionfiltercontext_is_manual_compaction(ptr) != 0;
58
59        Self {
60            is_full_compaction,
61            is_manual_compaction,
62        }
63    }
64}
65
66pub unsafe extern "C" fn create_compaction_filter_callback<F>(
67    raw_self: *mut c_void,
68    context: *mut ffi::rocksdb_compactionfiltercontext_t,
69) -> *mut ffi::rocksdb_compactionfilter_t
70where
71    F: CompactionFilterFactory,
72{
73    let self_ = &mut *(raw_self as *mut F);
74    let context = CompactionFilterContext::from_raw(context);
75    let filter = Box::new(self_.create(context));
76
77    let filter_ptr = Box::into_raw(filter);
78
79    ffi::rocksdb_compactionfilter_create(
80        filter_ptr as *mut c_void,
81        Some(compaction_filter::destructor_callback::<F::Filter>),
82        Some(compaction_filter::filter_callback::<F::Filter>),
83        Some(compaction_filter::name_callback::<F::Filter>),
84    )
85}
86
87#[cfg(test)]
88mod tests {
89    use super::*;
90    use crate::compaction_filter::Decision;
91    use crate::{Options, DB};
92    use std::ffi::CString;
93
94    struct CountFilter(u16, CString);
95    impl CompactionFilter for CountFilter {
96        fn filter(&mut self, _level: u32, _key: &[u8], _value: &[u8]) -> crate::CompactionDecision {
97            self.0 += 1;
98            if self.0 > 2 {
99                Decision::Remove
100            } else {
101                Decision::Keep
102            }
103        }
104
105        fn name(&self) -> &CStr {
106            &self.1
107        }
108    }
109
110    struct TestFactory(CString);
111    impl CompactionFilterFactory for TestFactory {
112        type Filter = CountFilter;
113
114        fn create(&mut self, _context: CompactionFilterContext) -> Self::Filter {
115            CountFilter(0, CString::new("CountFilter").unwrap())
116        }
117
118        fn name(&self) -> &CStr {
119            &self.0
120        }
121    }
122
123    #[test]
124    fn compaction_filter_factory_test() {
125        let path = "_rust_rocksdb_filter_factory_test";
126        let mut opts = Options::default();
127        opts.create_if_missing(true);
128        opts.set_compaction_filter_factory(TestFactory(CString::new("TestFactory").unwrap()));
129        {
130            let db = DB::open(&opts, path).unwrap();
131            let _r = db.put(b"k1", b"a");
132            let _r = db.put(b"_rk", b"b");
133            let _r = db.put(b"%k", b"c");
134            db.compact_range(None::<&[u8]>, None::<&[u8]>);
135            assert_eq!(db.get(b"%k1").unwrap(), None);
136        }
137        let result = DB::destroy(&opts, path);
138        assert!(result.is_ok());
139    }
140}
rocksdb/compaction_filter_factory.rs

rocksdb/
compaction_filter_factory.rs
