rocksdb/
compaction_filter.rs

1// Copyright 2020 Tyler Neely
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14//
15
16use libc::{c_char, c_int, c_uchar, c_void, size_t};
17use std::ffi::{CStr, CString};
18use std::slice;
19
20/// Decision about how to handle compacting an object
21///
22/// This is returned by a compaction filter callback. Depending
23/// on the value, the object may be kept, removed, or changed
24/// in the database during a compaction.
25pub enum Decision {
26    /// Keep the old value
27    Keep,
28    /// Remove the object from the database
29    Remove,
30    /// Change the value for the key
31    Change(&'static [u8]),
32}
33
34/// CompactionFilter allows an application to modify/delete a key-value at
35/// the time of compaction.
36pub trait CompactionFilter {
37    /// The compaction process invokes this
38    /// method for kv that is being compacted. The application can inspect
39    /// the existing value of the key and make decision based on it.
40    ///
41    /// Key-Values that are results of merge operation during compaction are not
42    /// passed into this function. Currently, when you have a mix of Put()s and
43    /// Merge()s on a same key, we only guarantee to process the merge operands
44    /// through the compaction filters. Put()s might be processed, or might not.
45    ///
46    /// When the value is to be preserved, the application has the option
47    /// to modify the existing_value and pass it back through new_value.
48    /// value_changed needs to be set to true in this case.
49    ///
50    /// Note that RocksDB snapshots (i.e. call GetSnapshot() API on a
51    /// DB* object) will not guarantee to preserve the state of the DB with
52    /// CompactionFilter. Data seen from a snapshot might disappear after a
53    /// compaction finishes. If you use snapshots, think twice about whether you
54    /// want to use compaction filter and whether you are using it in a safe way.
55    ///
56    /// If the CompactionFilter was created by a factory, then it will only ever
57    /// be used by a single thread that is doing the compaction run, and this
58    /// call does not need to be thread-safe.  However, multiple filters may be
59    /// in existence and operating concurrently.
60    fn filter(&mut self, level: u32, key: &[u8], value: &[u8]) -> Decision;
61
62    /// Returns a name that identifies this compaction filter.
63    /// The name will be printed to LOG file on start up for diagnosis.
64    fn name(&self) -> &CStr;
65}
66
67/// Function to filter compaction with.
68///
69/// This function takes the level of compaction, the key, and the existing value
70/// and returns the decision about how to handle the Key-Value pair.
71///
72///  See [Options::set_compaction_filter][set_compaction_filter] for more details
73///
74///  [set_compaction_filter]: ../struct.Options.html#method.set_compaction_filter
75pub trait CompactionFilterFn: FnMut(u32, &[u8], &[u8]) -> Decision {}
76impl<F> CompactionFilterFn for F where F: FnMut(u32, &[u8], &[u8]) -> Decision + Send + 'static {}
77
78pub struct CompactionFilterCallback<F>
79where
80    F: CompactionFilterFn,
81{
82    pub name: CString,
83    pub filter_fn: F,
84}
85
86impl<F> CompactionFilter for CompactionFilterCallback<F>
87where
88    F: CompactionFilterFn,
89{
90    fn name(&self) -> &CStr {
91        self.name.as_c_str()
92    }
93
94    fn filter(&mut self, level: u32, key: &[u8], value: &[u8]) -> Decision {
95        (self.filter_fn)(level, key, value)
96    }
97}
98
99pub unsafe extern "C" fn destructor_callback<F>(raw_cb: *mut c_void)
100where
101    F: CompactionFilter,
102{
103    drop(unsafe { Box::from_raw(raw_cb as *mut F) });
104}
105
106pub unsafe extern "C" fn name_callback<F>(raw_cb: *mut c_void) -> *const c_char
107where
108    F: CompactionFilter,
109{
110    let cb = unsafe { &*(raw_cb as *mut F) };
111    cb.name().as_ptr()
112}
113
114pub unsafe extern "C" fn filter_callback<F>(
115    raw_cb: *mut c_void,
116    level: c_int,
117    raw_key: *const c_char,
118    key_length: size_t,
119    existing_value: *const c_char,
120    value_length: size_t,
121    new_value: *mut *mut c_char,
122    new_value_length: *mut size_t,
123    value_changed: *mut c_uchar,
124) -> c_uchar
125where
126    F: CompactionFilter,
127{
128    use self::Decision::{Change, Keep, Remove};
129
130    let cb = unsafe { &mut *(raw_cb as *mut F) };
131    let key = unsafe { slice::from_raw_parts(raw_key as *const u8, key_length) };
132    let oldval = unsafe { slice::from_raw_parts(existing_value as *const u8, value_length) };
133    let result = cb.filter(level as u32, key, oldval);
134    match result {
135        Keep => 0,
136        Remove => 1,
137        Change(newval) => {
138            unsafe { *new_value = newval.as_ptr() as *mut c_char };
139            unsafe { *new_value_length = newval.len() as size_t };
140            unsafe { *value_changed = 1_u8 };
141            0
142        }
143    }
144}
145
146#[cfg(test)]
147#[allow(unused_variables)]
148fn test_filter(level: u32, key: &[u8], value: &[u8]) -> Decision {
149    use self::Decision::{Change, Keep, Remove};
150    match key.first() {
151        Some(&b'_') => Remove,
152        Some(&b'%') => Change(b"secret"),
153        _ => Keep,
154    }
155}
156
157#[test]
158fn compaction_filter_test() {
159    use crate::{Options, DB};
160
161    let tempdir = tempfile::Builder::new()
162        .prefix("_rust_rocksdb_filter_test")
163        .tempdir()
164        .expect("Failed to create temporary path for the _rust_rocksdb_filter_test");
165    let path = tempdir.path();
166    let mut opts = Options::default();
167    opts.create_if_missing(true);
168    opts.set_compaction_filter("test", test_filter);
169    {
170        let db = DB::open(&opts, path).unwrap();
171        let _r = db.put(b"k1", b"a");
172        let _r = db.put(b"_k", b"b");
173        let _r = db.put(b"%k", b"c");
174        db.compact_range(None::<&[u8]>, None::<&[u8]>);
175        assert_eq!(&*db.get(b"k1").unwrap().unwrap(), b"a");
176        assert!(db.get(b"_k").unwrap().is_none());
177        assert_eq!(&*db.get(b"%k").unwrap().unwrap(), b"secret");
178    }
179    let result = DB::destroy(&opts, path);
180    assert!(result.is_ok());
181}
rocksdb/compaction_filter.rs

rocksdb/
compaction_filter.rs
