menhirkv 0.4.5

// Copyright (C) 2024 Christian Mauduit <ufoot@ufoot.org>

use crate::error::{Error, Result};
use crate::iter::*;
use crate::serial::*;
use ofilter::SyncStream;
use rocksdb::compaction_filter::Decision;
use rocksdb::properties::STATS;
use rocksdb::IteratorMode;
use rocksdb::{ColumnFamily, Options, DB};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::fmt;
use std::hash::Hash;
use std::marker::PhantomData;
use std::path::Path;
use std::sync::Arc;
use tempfile::TempDir;

// https://blog.petitviolet.net/post/2021-03-25/use-rocksdb-from-rust

fn bloom_compaction_filter(filter: &SyncStream<Vec<u8>>, key: &[u8]) -> Decision {
    let k: Vec<u8> = Vec::from(key);
    if filter.check(&k) {
        Decision::Keep
    } else {
        Decision::Remove
    }
}

/// KV store based on RocksDB.
///
/// This is the main structure used for database access.
///
/// It is thread-safe.
///
/// # Examples
///
/// ```
/// use menhirkv::Store;
///
/// // Example with a key: usize, value: usize store,
/// // feel free to use your own types, obviously.
/// let store: Store<usize, usize> = Store::open_temporary(100).unwrap();
/// store.put(&123, &456).unwrap();
/// assert_eq!(Some(456), store.get(&123).unwrap());
/// ```
#[derive(Debug, Clone)]
pub struct Store<K, V>
where
    K: Serialize + DeserializeOwned,
    V: Serialize + DeserializeOwned,
{
    inner: Arc<Inner<K, V>>,
    cf: String,
}

struct Inner<K, V>
where
    K: Serialize + DeserializeOwned,
    V: Serialize + DeserializeOwned,
{
    path_str: String,
    tmp_dir: Option<TempDir>,
    db: DB,
    set_other_cfs: HashSet<String>,
    vec_other_cfs: Vec<String>,
    filter: SyncStream<Vec<u8>>,
    phantom_data: PhantomData<(K, V)>,
}

/// Complete dump of a KV store, easily (de)serializable.
///
/// This can be used to export data to cold storage.
///
///  # Examples
///
/// ```
/// use menhirkv::Store;
/// use serde_json::json;
///
/// let store = Store::<usize, usize>::open_temporary(100).unwrap();
/// store.put(&1, &2).unwrap();
/// store.put(&3, &4).unwrap();
/// let dump = store.dump().unwrap();
/// let js = json!(dump);
/// assert_eq!("{\"capacity\":100,\"data\":{\"\":[[1,2],[3,4]]}}", js.to_string());
/// ```
#[derive(Debug, Serialize, Deserialize)]
pub struct Dump<K, V> {
    /// Store capacity, which is a low limit of how many
    /// items the store should contain. Any item past that
    /// limit is likely to be dropped at some point in time.
    pub capacity: usize,
    /// Data exported, we use a vector of key-value pairs and
    /// not a hash, because by design, keys do not necessarily
    /// support the Hash trait.
    pub data: HashMap<String, Vec<(K, V)>>,
}

impl<K, V> fmt::Display for Store<K, V>
where
    K: Serialize + DeserializeOwned,
    V: Serialize + DeserializeOwned,
{
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(
            f,
            "{{ path: {}, capacity: {}, tmp: {} }}",
            self.inner.path_str,
            self.capacity(),
            !self.inner.tmp_dir.is_none()
        )
    }
}

impl<K, V> fmt::Debug for Inner<K, V>
where
    K: Serialize + DeserializeOwned,
    V: Serialize + DeserializeOwned,
{
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(
            f,
            "{{ path_str: {:?}, tmp_dir: {:?}, set_other_cfs: {:?}, vec_other_cfs: {:?}, filter: {:?}}}",
            self.path_str,
            self.tmp_dir,
            self.set_other_cfs,
            self.vec_other_cfs,
            self.filter,
        )
    }
}

const TEMP_DB: &str = "tmp.db";
const CF_META: &str = "menhir-kv-meta";
const CF_MAIN: &str = "menhir-kv-root";
const CF_OTHER: &str = "menhir-kv-other";
const KEY_CAPACITY: &str = "capacity";
const BLOOM_COMPACTION_FILTER: &str = "bloom-compaction-filter";

impl<K, V> Store<K, V>
where
    K: Serialize + DeserializeOwned,
    V: Serialize + DeserializeOwned,
{
    /// Open a new store with the given RocksDB options.
    ///
    /// By default, the store is not created, if it does exist, this will fail.
    /// Hint: there is a `create_if_missing` option in RocksDB API.
    ///
    /// On top of these options, internally, MenhirKV will install
    /// a custom compaction filter to make sure data expires at some point.
    ///
    /// # Examples
    ///
    /// ```
    /// use menhirkv::{Store, rocksdb::Options, Result};
    /// use std::path::Path;
    ///
    /// let mut opts = Options::default();
    /// opts.create_if_missing(true);
    /// opts.set_max_write_buffer_number(32); // let's pretend we need this
    ///
    /// let store: Result<Store<String, String>> = Store::open_with_options(&opts, Path::new("/tmp/test-with-options.db"), 10_000);
    /// match store {
    ///     Ok(s) => println!("strings store: {}", s),
    ///     Err(e) => println!("ooops: {}", e),
    /// }
    /// ```
    pub fn open_with_options<P>(options: &Options, path: P, capacity: usize) -> Result<Self>
    where
        P: AsRef<Path>,
    {
        Self::open_cf_with_options::<P>(options, path, &[], capacity)
    }

    /// Open a new store with a given path.
    ///
    /// By default, the store is created if it does not exist.
    /// Otherwise it re-uses the existing store.
    ///
    /// # Examples
    ///
    /// ```
    /// use menhirkv::Store;
    /// use std::path::Path;
    ///
    /// match Store::<String, String>::open_with_path(Path::new("/tmp/test-with-path.db"), 10_000) {
    ///     Ok(s) => println!("strings store: {}", s),
    ///     Err(e) => println!("ooops: {}", e),
    /// }
    /// ```
    pub fn open_with_path<P>(path: P, capacity: usize) -> Result<Self>
    where
        P: AsRef<Path>,
    {
        Self::open_cf_with_path::<P>(path, &[], capacity)
    }

    /// Open a temporary store.
    ///
    /// This ensures a new store is created, and it is destroyed and removed
    /// from filesystem when the store is dropped.
    ///
    /// This is convenient mostly for testing, as it still creates a real
    /// RocksDB database on filesystem, so any performance issue and/or side effect
    /// should be exposed. However, since the store is completely removed when
    /// program exits, it is not suitable for true persistent storage.
    ///
    /// # Examples
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let s = Store::<String, String>::open_temporary(10_000).unwrap();
    /// println!("strings store: {}", s);
    /// ```
    pub fn open_temporary(capacity: usize) -> Result<Self> {
        Self::open_cf_temporary(&[], capacity)
    }

    /// Open a new store with the given RocksDB options, and some column families.
    ///
    /// By default, the store is not created, if it does exist, this will fail.
    /// Hint: there is a `create_if_missing` option in RocksDB API.
    ///
    /// On top of these options, internally, MenhirKV will install
    /// a custom compaction filter to make sure data expires at some point.
    ///
    /// # Examples
    ///
    /// ```
    /// use menhirkv::{Store, rocksdb::Options, Result};
    /// use std::path::Path;
    ///
    /// let mut opts = Options::default();
    /// opts.create_if_missing(true);
    /// opts.set_max_write_buffer_number(32); // let's pretend we need this
    ///
    /// let store: Result<Store<String, String>> = Store::open_cf_with_options(&opts, Path::new("/tmp/test-with-options.db"), &["foo", "bar"], 10_000);
    /// match store {
    ///     Ok(s) => println!("strings store: {}", s),
    ///     Err(e) => println!("ooops: {}", e),
    /// }
    /// ```
    pub fn open_cf_with_options<P>(
        options: &Options,
        path: P,
        cfs: &[&str],
        capacity: usize,
    ) -> Result<Self>
    where
        P: AsRef<Path>,
    {
        Self::open(options, Some(path), cfs, None, capacity)
    }

    /// Open a new store with a given path, and some column families.
    ///
    /// By default, the store is created if it does not exist.
    /// Otherwise it re-uses the existing store.
    ///
    /// # Examples
    ///
    /// ```
    /// use menhirkv::Store;
    /// use std::path::Path;
    ///
    /// match Store::<String, String>::open_cf_with_path(Path::new("/tmp/test-with-path.db"), &["foo", "bar"], 10_000) {
    ///     Ok(s) => println!("strings store: {}", s),
    ///     Err(e) => println!("ooops: {}", e),
    /// }
    /// ```
    pub fn open_cf_with_path<P>(path: P, cfs: &[&str], capacity: usize) -> Result<Self>
    where
        P: AsRef<Path>,
    {
        let mut options = Options::default();
        options.create_if_missing(true);
        Self::open(&options, Some(path), cfs, None, capacity)
    }

    /// Open a temporary store with some column families.
    ///
    /// This ensures a new store is created, and it is destroyed and removed
    /// from filesystem when the store is dropped.
    ///
    /// This is convenient mostly for testing, as it still creates a real
    /// RocksDB database on filesystem, so any performance issue and/or side effect
    /// should be exposed. However, since the store is completely removed when
    /// program exits, it is not suitable for true persistent storage.
    ///
    /// # Examples
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let s = Store::<String, String>::open_cf_temporary(&["foo", "bar"], 10_000).unwrap();
    /// println!("strings store: {}", s);
    /// ```
    pub fn open_cf_temporary(cfs: &[&str], capacity: usize) -> Result<Self> {
        let tmp_dir = match TempDir::new() {
            Ok(d) => d,
            Err(e) => {
                return Err(Error::db(&format!(
                    "unable to create temporary directory for db: {}",
                    e
                )))
            }
        };

        let mut options = Options::default();
        options.create_if_missing(true);

        Self::open(&options, None::<&Path>, cfs, Some(tmp_dir), capacity)
    }

    fn open<P>(
        options: &Options,
        path: Option<P>,
        cfs: &[&str],
        tmp_dir: Option<TempDir>,
        capacity: usize,
    ) -> Result<Self>
    where
        P: AsRef<Path>,
    {
        for cf in cfs.iter() {
            if cf == &"" {
                // fail fast
                return Err(Error::invalid_data(
                    "can not create column family with empty name",
                ));
            }
        }

        let mut set_other_cfs: HashSet<String> = HashSet::with_capacity(cfs.len());
        let mut vec_other_cfs: Vec<String> = Vec::with_capacity(cfs.len());
        cfs.iter()
            .map(|cf| {
                set_other_cfs.insert(String::from(*cf));
                vec_other_cfs.push(String::from(*cf));
            })
            .count();
        if set_other_cfs.len() != vec_other_cfs.len() {
            // fail fast
            return Err(Error::invalid_data(&format!(
                "column family repeated in \"[{}]\"",
                vec_other_cfs.join(", ")
            )));
        }

        let (db, path_str) = match path {
            Some(p) => match &tmp_dir {
                Some(_) => {
                    return Err(Error::report_bug(
                        "either path or tmp_dir must be set, but not both",
                    ))
                }
                None => match DB::open(options, p.as_ref()) {
                    Ok(db) => (db, format!("{}", p.as_ref().display())),
                    Err(e) => return Err(Error::from(e)),
                },
            },
            None => match &tmp_dir {
                Some(d) => {
                    let p = d.path().join(TEMP_DB);
                    match DB::open(options, &p) {
                        Ok(db) => (db, format!("{}", &p.display())),
                        Err(e) => return Err(Error::from(e)),
                    }
                }
                None => {
                    return Err(Error::report_bug(
                        " path or tmp_dir must be set, but none is defined",
                    ))
                }
            },
        };

        let mut filter_options: Options = options.clone();
        let filter = SyncStream::new(capacity);

        let mut inner = Inner {
            path_str,
            tmp_dir, // it's important to keep track of this, when defined
            db,
            set_other_cfs,
            vec_other_cfs,
            filter,
            phantom_data: PhantomData,
        };
        let compaction_filter: SyncStream<Vec<u8>> = inner.filter.clone();
        // The following code, setting the compaction filter through set_compaction_filter,
        // does *NOT* work on rust < 1.67. Message follows:
        //
        // ----8<-------------------------------------------------
        // 	error: implementation of `FnOnce` is not general enough
        //    --> src/store.rs:147:9
        //     |
        // 147 | /         filter_options
        // 148 | |             .set_compaction_filter(BLOOM_COMPACTION_FILTER, move |_level, key, _value| {
        // 149 | |                 bloom_compaction_filter(&f, key)
        // 150 | |             });
        //     | |______________^ implementation of `FnOnce` is not general enough
        //     |
        //     = note: closure with signature `fn(u32, &'2 [u8], &[u8]) -> CompactionDecision` must implement `FnOnce<(u32, &'1 [u8], &[u8])>`, for any lifetime `'1`...
        //     = note: ...but it actually implements `FnOnce<(u32, &'2 [u8], &[u8])>`, for some specific lifetime `'2`
        // ----8<-------------------------------------------------
        //
        // Possibly related to this:
        // https://www.reddit.com/r/rust/comments/ntqu68/implementation_of_fnonce_is_not_general_enough/
        // Suggests it could be a type inference issue.
        //
        // [TODO] -> get to the bottom of it, be more explicit with closure type?
        filter_options
            .set_compaction_filter(BLOOM_COMPACTION_FILTER, move |_level, key, _value| {
                bloom_compaction_filter(&compaction_filter, key)
            });

        let mut meta_created_now = false;
        if (&inner.db).cf_handle(CF_META).is_none() {
            (&mut inner.db).create_cf(CF_META, options)?;
            if (&inner.db).cf_handle(CF_META).is_none() {
                return Err(Error::report_bug("unable to create meta column family"));
            }
            meta_created_now = true;
        };
        if (&inner.db).cf_handle(CF_MAIN).is_none() {
            (&mut inner.db).create_cf(CF_MAIN, &filter_options)?;
            if (&inner.db).cf_handle(CF_MAIN).is_none() {
                return Err(Error::report_bug(
                    "unable to create main data column family",
                ));
            }
        };
        for cf in cfs.iter() {
            let cf_real_name = Self::cf_real_name(cf);
            if (&inner.db).cf_handle(&cf_real_name).is_none() {
                (&mut inner.db).create_cf(&cf_real_name, &filter_options)?;
                if (&inner.db).cf_handle(&cf_real_name).is_none() {
                    return Err(Error::report_bug(&format!(
                        "unable to create custom column family \"{}\" with real name \"{}\"",
                        cf, cf_real_name,
                    )));
                }
            };
        }
        let store = Store {
            inner: Arc::new(inner),
            cf: String::from(CF_MAIN),
        };

        if meta_created_now {
            store.set_capacity(capacity)?;
        }

        // Do some (useless) read/write to fail fast if there's anything
        // wrong with that DB.
        let current_capacity = store.get_capacity()?;
        store.set_capacity(current_capacity)?;

        Ok(store)
    }

    /// Return the RocksDB object powering the store.
    ///
    /// This is provided for debugging purposes, and may also serve
    /// to gather statistics or properties from the db. Obviously,
    /// do not attempt to write data into it, that could interfere
    /// with the logic of this package.
    ///
    /// # Examples
    ///
    /// ```
    /// use menhirkv::Store;
    /// use rocksdb::properties::CUR_SIZE_ACTIVE_MEM_TABLE;
    ///
    /// let s = Store::<String, String>::open_temporary(10_000).unwrap();
    /// let db = s.db();
    /// let n = db.live_files().unwrap().into_iter().map(|f| println!("* {:?}", f)).count();
    /// println!("live files count: {}", n);
    /// let active_mt = db.property_int_value(CUR_SIZE_ACTIVE_MEM_TABLE).unwrap().unwrap();
    /// println!("active mem tables: {}", active_mt);
    /// ```
    pub fn db(&self) -> &DB {
        &self.inner.db
    }

    /// Return RocksDB stats.
    ///
    /// This is mostly for general information, if you want detailed info,
    /// call `property_value` on the db object itself.
    ///
    /// # Examples
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// println!("stats: {}", store.db_stats().unwrap());
    /// store.put(&1, &10).unwrap();
    /// println!("stats: {}", store.db_stats().unwrap());
    /// ```
    pub fn db_stats(&self) -> Result<String> {
        let stats = self.inner.db.property_value(STATS)?;
        match stats {
            Some(stats) => match stats.len() {
                0 => Err(Error::report_bug("stats are empty")),
                _ => Ok(stats),
            },
            None => Err(Error::report_bug("there are no stats")),
        }
    }

    fn cf_meta_handle(&self) -> Result<&ColumnFamily> {
        match self.inner.db.cf_handle(CF_META) {
            Some(handle) => Ok(handle),
            // This should *ALWAYS* be here.
            None => Err(Error::report_bug("unable to open meta column family")),
        }
    }

    fn cf_data_handle(&self) -> Result<&ColumnFamily> {
        match self.inner.db.cf_handle(&self.cf) {
            Some(handle) => Ok(handle),
            // At this stage if it does not exist, it's a bug, we should
            // have checked for existence long before.
            None => Err(Error::report_bug("unable to open data column family")),
        }
    }

    fn cf_real_name(pub_name: &str) -> String {
        match pub_name {
            "" => String::from(CF_MAIN),
            other => format!("{}-{}", CF_OTHER, other),
        }
    }

    fn cf_real_name_handle(&self, real_name: &str) -> Result<&ColumnFamily> {
        match self.inner.db.cf_handle(real_name) {
            Some(handle) => Ok(handle),
            // At this stage if it does not exist, it's a bug, we should
            // have checked for existence long before.
            None => Err(Error::report_bug(&format!(
                "unable to open data column family with real name \"{}\"",
                real_name
            ))),
        }
    }

    /// Iterate over the column family names of a KV store.
    ///
    /// Use this to list the name of all column family names.
    /// It will also yield the empty string, corresponding to
    /// the main/root column family you open by default.
    ///
    /// These are not necessarily exactly the names used by RocksDB
    /// internally, there can be some marshalling. Typically,
    /// a prefix is added. Unless you inspect the content of the
    /// inner database yourself, do not worry about this.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_cf_temporary(&["a", "b"], 100).unwrap();
    /// let cf_names = store.iter_cf_names().collect::<Vec<&str>>();
    /// assert_eq!(vec!["", "a", "b"], cf_names);
    /// ```
    pub fn iter_cf_names(&self) -> IterCfNames<'_> {
        IterCfNames {
            idx: -1,
            other_cfs: &self.inner.vec_other_cfs,
        }
    }

    /// Return all the column family names of a KV store.
    ///
    /// Use this to list the name of all column family names.
    /// Similar to `iter_cf_names`, but returns an array of owned values.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_cf_temporary(&["a", "b"], 100).unwrap();
    /// let cf_names = store.cf_names();
    /// assert_eq!(", a, b", cf_names.join(", "));
    /// ```
    pub fn cf_names(&self) -> Vec<String> {
        self.iter_cf_names().map(|cf| String::from(cf)).collect()
    }

    /// Give access to a column family.
    ///
    /// This returns an object which will by default work on the
    /// given column family. Think of column families as namespaces.
    /// The capacity is shared between all column families.
    ///
    /// Opening the empty str `""` will return a handle on the
    /// default column family. This may not necessarily be
    /// the default column family from RocksDB point of view.
    ///
    /// # Examples
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_cf_temporary(&["foo", "bar"], 10_000).unwrap();
    /// store.put(&1, &10);
    /// let store_foo = store.cf("foo").unwrap();
    /// store_foo.put(&2, &20);
    /// let store_bar = store.cf("bar").unwrap();
    /// store_bar.put(&3, &30);
    /// let store_root = store_bar.cf("").unwrap();
    /// assert_eq!(Some(10), store_root.get(&1).unwrap());
    /// assert_eq!(None, store_root.get(&2).unwrap());
    /// assert_eq!(None, store_root.get(&3).unwrap());
    /// assert_eq!(None, store_foo.get(&1).unwrap());
    /// assert_eq!(Some(20), store_foo.get(&2).unwrap());
    /// assert_eq!(None, store_foo.get(&3).unwrap());
    /// assert_eq!(None, store_bar.get(&1).unwrap());
    /// assert_eq!(None, store_bar.get(&2).unwrap());
    /// assert_eq!(Some(30), store_bar.get(&3).unwrap());
    /// ```
    pub fn cf(&self, name: &str) -> Result<Store<K, V>> {
        let real_name = match name {
            "" => String::from(CF_MAIN),
            other => {
                if !self.inner.set_other_cfs.contains(&String::from(other)) {
                    return Err(Error::invalid_data(&format!(
                        "custom column family \"{}\" does not exist",
                        other
                    )));
                }
                Self::cf_real_name(other)
            }
        };
        if self.inner.db.cf_handle(&real_name).is_none() {
            return Err(Error::report_bug(&format!(
                "column family \"{}\" with real name \"{}\" not found",
                name, real_name,
            )));
        }
        Ok(Store {
            inner: self.inner.clone(),
            cf: real_name,
        })
    }

    fn set_meta_size(&self, s: &str, capacity: usize) -> Result<()> {
        let meta = self.cf_meta_handle()?;
        let buf = serialize_size(capacity)?;
        self.inner.db.put_cf(meta, s, buf)?;
        Ok(())
    }

    fn get_meta_size(&self, s: &str) -> Result<usize> {
        let meta = self.cf_meta_handle()?;
        let v = self.inner.db.get_cf(meta, s)?;
        match v {
            Some(buf) => {
                let size = deserialize_size(buf)?;
                Ok(size)
            }
            None => Err(Error::report_bug(&format!("no \"{}\" size in db", s))),
        }
    }

    fn set_capacity(&self, capacity: usize) -> Result<()> {
        self.set_meta_size(KEY_CAPACITY, capacity)
    }

    fn get_capacity(&self) -> Result<usize> {
        self.get_meta_size(KEY_CAPACITY)
    }

    /// Readable path of the DB.
    ///
    /// This is not a real std::path::Path and you can not (should not) use this
    /// to open the database. It is here for debugging purposes, logging, to get
    /// clues on what's happening.
    ///
    /// # Examples
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<String, String>::open_temporary(100).unwrap();
    /// println!("{}", store.path());
    /// ```
    pub fn path(&self) -> String {
        self.inner.path_str.clone()
    }

    /// Capacity of the store.
    ///
    /// Number of items the store should keep. In practice it is going to
    /// keep much more than this, because:
    ///
    /// * the streaming Bloom filter, be design, keeps more than expected
    /// * RocksDB compaction needs to happen, and this is not continous
    ///
    /// However, one can reasonably expect the total number of items stored
    /// will be between this number and 10X this number.
    ///
    /// # Examples
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<String, String>::open_temporary(100).unwrap();
    /// assert_eq!(100, store.capacity());
    /// ```
    pub fn capacity(&self) -> usize {
        self.inner.filter.capacity()
    }

    /// Resize the store.
    ///
    /// This updates the capacity. When shrinking the store, extra entries
    /// are not necessarily dropped immediately. The change will happen
    /// when the streaming Bloom filter "ages".
    /// The entries removed in prority will be the old ones.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(1_000).unwrap();
    /// for i in 0..10_000 {
    ///     store.put(&i, &i).unwrap();
    /// }
    /// store.resize(100).unwrap();
    /// assert_eq!(100, store.capacity());
    /// assert_eq!(Some(9_999), store.get(&9_999).unwrap());
    /// ```
    pub fn resize(&self, capacity: usize) -> Result<()> {
        self.set_capacity(capacity)?;
        self.inner.filter.resize(capacity);
        Ok(())
    }

    /// Put item into store.
    ///
    /// When data is put in the store, it is marked as "recently used"
    /// so it will stay in the store for as long as possible.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(1_000).unwrap();
    /// store.put(&1, &10).unwrap();
    /// assert_eq!(Some(10), store.get(&1).unwrap());
    /// store.put(&1, &100).unwrap();
    /// assert_eq!(Some(100), store.get(&1).unwrap());
    /// ```
    pub fn put(&self, key: &K, value: &V) -> Result<()> {
        let data = self.cf_data_handle()?;
        let k = serialize_key(key)?;
        self.inner.filter.set(&k); // put in the bloom filter
        let v = serialize_value(value)?;
        self.inner.db.put_cf(data, k, v)?;
        Ok(())
    }

    /// Get item, mark it as recently used.
    ///
    /// This is a bit more than "getting data". Not only this returns
    /// the value, if it exists, but it also marks the key as "used",
    /// so that the key/value pair will not be removed during the next compaction,
    /// if the number of stored items exceeds capacity.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// let mut last_777: Option<usize> = None;
    /// for i in 1..10_000 {
    ///     store.put(&i, &(i*10)).unwrap();
    ///     last_777 = store.get(&777).unwrap();
    /// }
    /// assert_eq!(Some(7_770), last_777);
    /// assert_eq!(Some(99_990), store.get(&9_999).unwrap());
    /// ```
    pub fn get(&self, key: &K) -> Result<Option<V>> {
        let data = self.cf_data_handle()?;
        let k = serialize_key(key)?;
        self.inner.filter.set(&k); // put in the bloom filter
        let resp = self.inner.db.get_cf(data, k)?;
        match resp {
            Some(buf) => {
                let v = deserialize_value(buf)?;
                Ok(Some(v))
            }
            None => Ok(None),
        }
    }

    /// Peek item, does not mark it as recently used.
    ///
    /// This is different from `get()`, it returns the value,
    /// but will not mark it as recently used. So it may be removed
    /// soon, as if it had not been accessed.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&1, &10).unwrap();
    /// assert_eq!(Some(10), store.peek(&1).unwrap());
    /// ```
    pub fn peek(&self, key: &K) -> Result<Option<V>> {
        let data = self.cf_data_handle()?;
        let k = serialize_key(key)?;
        let resp = self.inner.db.get_cf(data, k)?;
        match resp {
            Some(buf) => {
                let v = deserialize_value(buf)?;
                Ok(Some(v))
            }
            None => Ok(None),
        }
    }

    /// Bump item, mark it as recently used.
    ///
    /// This marks the item as recently used, preventing it
    /// from being removed, but does not return the value,
    /// avoiding the costs of deserialization.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// for i in 1..10_000 {
    ///     store.put(&i, &(i*10)).unwrap();
    ///     store.bump(&777).unwrap();
    /// }
    /// assert_eq!(Some(7_770), store.get(&777).unwrap());
    /// assert_eq!(Some(99_990), store.get(&9_999).unwrap());
    /// ```
    pub fn bump(&self, key: &K) -> Result<()> {
        let k = serialize_key(key)?;
        self.inner.filter.set(&k); // put in the bloom filter
        Ok(())
    }

    /// Delete item, remove it from store.
    ///
    /// This removes the item from the database, but does
    /// not remove it from the Bloom filter. As a side effect,
    /// it still counts for an entry in that filter. So if you
    /// keep removing entries actively, the store may appear
    /// to retain less than capacity.
    ///
    /// In general, the way to use MenhirKV is to let the store
    /// take care of removal, so while it is totally OK to
    /// pro-actively remove items, it is interesting to find
    /// out why you need to remove them.
    ///
    /// If it is just to get rid of useless junk... the store
    /// does it for you, and possibly faster. Very likely, the
    /// physical removal would happen during compaction, and this
    /// is exactly what natural expiration would do.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&1, &10).unwrap();
    /// assert_eq!(Some(10), store.get(&1).unwrap());
    /// store.delete(&1).unwrap();
    /// assert_eq!(None, store.get(&1).unwrap());
    /// ```
    pub fn delete(&self, key: &K) -> Result<()> {
        let data = self.cf_data_handle()?;
        let k = serialize_key(key)?;
        self.inner.db.delete_cf(data, k)?;
        Ok(())
    }

    /// Returns true if item exists.
    ///
    /// This will return true even if the item would be removed
    /// during the next compaction. This is to be consistent with
    /// the results of `get()` and `peek()`.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// assert!(!store.has(&1).unwrap());
    /// store.put(&1, &10).unwrap();
    /// assert!(store.has(&1).unwrap());
    /// ```
    pub fn has(&self, key: &K) -> Result<bool> {
        let data = self.cf_data_handle()?;
        let k = serialize_key(key)?;
        let resp = self.inner.db.get_cf(data, k)?;
        Ok(!resp.is_none())
    }

    /// Manual compaction.
    ///
    /// This forces a RocksDB compaction. Under normal circumstances,
    /// you should never need to do this, unless doing some testing.
    /// RocksDB should trigger this automatically when needed.
    ///
    /// This compacts all column families.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// for i in 1..10_000 {
    ///     store.put(&i, &(i*10)).unwrap();
    /// }
    /// store.compact().unwrap();
    /// assert_eq!(store.len().unwrap(), store.db_len().unwrap()); // slow, don't do this
    /// ```
    pub fn compact(&self) -> Result<()> {
        for pub_name in self.iter_cf_names() {
            let real_name = Self::cf_real_name(pub_name);
            let handle = self.cf_real_name_handle(&real_name)?;
            self.inner
                .db
                .compact_range_cf(handle, Some(MIN_KEY), Some(MAX_KEY));
        }
        let meta_handle = self.cf_meta_handle()?;
        self.inner
            .db
            .compact_range_cf(meta_handle, Some(MIN_KEY), Some(MAX_KEY));
        self.inner.db.compact_range(Some(MIN_KEY), Some(MAX_KEY));
        Ok(())
    }

    /// Manual flush.
    ///
    /// This forces a RocksDB flush. Under normal circumstances,
    /// you should never need to do this. This is called automatically
    /// when the store is dropped. However, when called during a
    /// drop, it will not raise a panic() or signal anything if
    /// something goes wrong. So if you want to be double sure
    /// to store your data, call this.
    ///
    /// This flushes all column families.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// match store.flush() {
    ///     Ok(_) => println!("all good"),
    ///     Err(e) => {
    ///         println!("some data possibly lost");
    ///         panic!("flush failed {}", e);
    ///     }
    /// }
    /// ```
    pub fn flush(&self) -> Result<()> {
        for pub_name in self.iter_cf_names() {
            let real_name = Self::cf_real_name(pub_name);
            let handle = self.cf_real_name_handle(&real_name)?;
            self.inner.db.flush_cf(handle)?;
        }
        let meta_handle = self.cf_meta_handle()?;
        self.inner.db.flush_cf(meta_handle)?;
        self.inner.db.flush()?;
        Ok(())
    }

    /// Number of items in the store, after a possible compaction.
    ///
    /// This returns the number of items in the store that would be
    /// left after the next compaction. It is also the number of items
    /// that would be yielded by iterators. This is a per-column-family value.
    ///
    /// WARNING: this is very slow!
    ///
    /// It iterates on all items, even those that would be compacted,
    /// filters them... Possibly the slowest thing you could do on the
    /// store. Use it for debugging, on at rare occasions to get
    /// point-in-time stats.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&3, &4).unwrap();
    /// assert_eq!(1, store.len().unwrap()); // slow, don't do this
    /// ```
    pub fn len(&self) -> Result<usize> {
        let mut len = 0;
        let data = self.cf_data_handle()?;
        let mut iter = self.inner.db.full_iterator_cf(data, IteratorMode::Start);
        loop {
            if let Some(res) = iter.next() {
                match res {
                    Ok(item) => {
                        // We filter out keys which would be
                        // dropped anyway by the next compaction.
                        let kvec: &Vec<u8> = &item.0.to_vec();
                        if !self.inner.filter.check(kvec) {
                            continue;
                        }
                        len += 1;
                    }
                    Err(e) => return Err(Error::from(e)),
                }
            } else {
                break;
            }
        }
        Ok(len)
    }

    /// Total number of items in RocksDB, for the current column family.
    ///
    /// This returns the number of items in the store, even those which
    /// which would be removed by the next compaction. It is typically
    /// greater than what would be yielded by iterators.
    ///
    /// WARNING: this is very slow!
    ///
    /// It iterates on all items, there is no easy way to reliably know
    /// the number of items, you can get estimations, but not the precise
    /// number. Use this with caution, only for debugging.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&3, &4).unwrap();
    /// assert_eq!(1, store.db_len().unwrap()); // slow, don't do this
    /// ```
    pub fn db_len(&self) -> Result<usize> {
        let data = self.cf_data_handle()?;
        let iter = self.inner.db.full_iterator_cf(data, IteratorMode::Start);
        Ok(iter.count())
    }

    /// Removes all entries from store.
    ///
    /// WARNING: this is very slow!
    ///
    /// It iterates on all items. There would be a quick way to do this,
    /// which is dropping the column family, then recreate it. But that
    /// would require some sort of locking strategy. Since clearing data
    /// is not really required in an auto-expiring model, this is just
    /// provided as a possible debugging feature, but keep in mind it is
    /// very likely going to be quite slow.
    ///
    /// Only clears the current column family.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&3, &4).unwrap();
    /// assert_eq!(1, store.db_len().unwrap()); // slow, don't do this
    /// store.clear().unwrap();
    /// assert_eq!(0, store.db_len().unwrap()); // slow, don't do this
    /// ```
    pub fn clear(&self) -> Result<usize> {
        let mut len = 0;
        let data = self.cf_data_handle()?;
        let iter = self.inner.db.full_iterator_cf(data, IteratorMode::Start);
        for item in iter {
            let kv = item?;
            self.inner.db.delete_cf(data, kv.0)?;
            len += 1;
        }
        self.compact()?;
        Ok(len)
    }

    /// Iterates on all items.
    ///
    /// IMPORTANT: the iterator will only iterate on entries which would
    /// be removed by the next compaction. This means you could have
    /// an item which is retrievable by `get()` or `peek()`, yet will
    /// never be yielded by the iterator.
    ///
    /// The idea is that when you
    /// iterate, you care about what is "recent" and "important".
    /// Old items which are still in the database are still available
    /// if you know their keys, but they are not discoverable through
    /// iterators.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&1, &2).unwrap();
    /// store.put(&3, &4).unwrap();
    /// store.put(&5, &6).unwrap();
    /// for item in store.iter().unwrap() {
    ///     let item = item.unwrap();
    ///     println!("key: {}, value: {}", item.0, item.1);
    /// }
    /// ```
    pub fn iter(&self) -> Result<Iter<'_, K, V>> {
        let data = self.cf_data_handle()?;
        let inner_iter = self.inner.db.full_iterator_cf(data, IteratorMode::Start);
        let iter: Iter<'_, K, V> = Iter {
            inner: inner_iter,
            filter: self.inner.filter.clone(),
            done: 0,
            phantom_data: PhantomData,
        };
        Ok(iter)
    }

    /// Iterates on all keys.
    ///
    /// Only yield keys that would still be here after a compaction.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&1, &2).unwrap();
    /// store.put(&3, &4).unwrap();
    /// store.put(&5, &6).unwrap();
    /// for key in store.keys().unwrap() {
    ///     let key = key.unwrap();
    ///     println!("key: {}", key);
    /// }
    /// ```
    pub fn keys(&self) -> Result<Keys<'_, K, V>> {
        let data = self.cf_data_handle()?;
        let inner_iter = self.inner.db.full_iterator_cf(data, IteratorMode::Start);
        let iter: Keys<'_, K, V> = Keys {
            inner: inner_iter,
            filter: self.inner.filter.clone(),
            done: 0,
            phantom_data: PhantomData,
        };
        Ok(iter)
    }

    /// Iterates on all values.
    ///
    /// Only yield values that would still be here after a compaction.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&1, &2).unwrap();
    /// store.put(&3, &4).unwrap();
    /// store.put(&5, &6).unwrap();
    /// for value in store.values().unwrap() {
    ///     let value = value.unwrap();
    ///     println!("value: {}", value);
    /// }
    /// ```
    pub fn values(&self) -> Result<Values<'_, K, V>> {
        let data = self.cf_data_handle()?;
        let inner_iter = self.inner.db.full_iterator_cf(data, IteratorMode::Start);
        let iter: Values<'_, K, V> = Values {
            inner: inner_iter,
            filter: self.inner.filter.clone(),
            done: 0,
            phantom_data: PhantomData,
        };
        Ok(iter)
    }

    /// Export all items with an iterator.
    ///
    /// The difference between this and a standard iterator is that
    /// this one returns `(K, V)` instead of `Result<(K, V)>`.
    /// A consequence of this is that it may `panic()` at any iteration,
    /// as there can always be an error, and it will not catch it.
    ///
    /// If you want something failsafe, use the standard iterator.
    /// This one is easier to use, but can crash your program.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&1, &2).unwrap();
    /// store.put(&3, &4).unwrap();
    /// store.put(&5, &6).unwrap();
    /// let export = store.export().unwrap().collect::<Vec<(usize, usize)>>(); // may panic()
    /// assert_eq!(3, export.len());
    /// ```
    pub fn export(&self) -> Result<Export<'_, K, V>> {
        let data = self.cf_data_handle()?;
        let inner_iter = self.inner.db.full_iterator_cf(data, IteratorMode::Start);
        let iter: Export<'_, K, V> = Export {
            inner: inner_iter,
            filter: self.inner.filter.clone(),
            done: 0,
            phantom_data: PhantomData,
        };
        Ok(iter)
    }

    /// Export all keys with an iterator.
    ///
    /// The difference between this and a standard iterator is that
    /// this one returns `K` instead of `Result<K>`.
    /// A consequence of this is that it may `panic()` at any iteration,
    /// as there can always be an error, and it will not catch it.
    ///
    /// If you want something failsafe, use the standard iterator.
    /// This one is easier to use, but can crash your program.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&1, &2).unwrap();
    /// store.put(&3, &4).unwrap();
    /// store.put(&5, &6).unwrap();
    /// let mut keys = store.export_keys().unwrap().collect::<Vec<usize>>(); // may panic()
    /// keys.sort();
    /// assert_eq!(vec![1, 3, 5], keys);
    /// ```
    pub fn export_keys(&self) -> Result<ExportKeys<'_, K, V>> {
        let data = self.cf_data_handle()?;
        let inner_iter = self.inner.db.full_iterator_cf(data, IteratorMode::Start);
        let iter: ExportKeys<'_, K, V> = ExportKeys {
            inner: inner_iter,
            filter: self.inner.filter.clone(),
            done: 0,
            phantom_data: PhantomData,
        };
        Ok(iter)
    }

    /// Export all values with an iterator.
    ///
    /// The difference between this and a standard iterator is that
    /// this one returns `V` instead of `Result<V>`.
    /// A consequence of this is that it may `panic()` at any iteration,
    /// as there can always be an error, and it will not catch it.
    ///
    /// If you want something failsafe, use the standard iterator.
    /// This one is easier to use, but can crash your program.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&1, &2).unwrap();
    /// store.put(&3, &4).unwrap();
    /// store.put(&5, &6).unwrap();
    /// let mut values = store.export_values().unwrap().collect::<Vec<usize>>(); // may panic()
    /// values.sort();
    /// assert_eq!(vec![2, 4, 6], values);
    /// ```
    pub fn export_values(&self) -> Result<ExportValues<'_, K, V>> {
        let data = self.cf_data_handle()?;
        let inner_iter = self.inner.db.full_iterator_cf(data, IteratorMode::Start);
        let iter: ExportValues<'_, K, V> = ExportValues {
            inner: inner_iter,
            filter: self.inner.filter.clone(),
            done: 0,
            phantom_data: PhantomData,
        };
        Ok(iter)
    }

    /// Export all items in a vec.
    ///
    /// You could do that with an iterator, but either check for
    /// errors at each iteration, or take the risk of a `panic()`.
    ///
    /// This utility just dumps everything, the cost being that
    /// all items must fit in memory.
    ///
    /// If you want something failsafe, use the standard iterator.
    /// This one is easier to use, but can crash your program.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&1, &2).unwrap();
    /// store.put(&3, &4).unwrap();
    /// store.put(&5, &6).unwrap();
    /// let vec = store.to_vec().unwrap();
    /// assert_eq!(3, vec.len());
    /// ```
    pub fn to_vec(&self) -> Result<Vec<(K, V)>> {
        let mut iter = self.export()?;
        let mut ret: Vec<(K, V)> = Vec::with_capacity(self.capacity());
        loop {
            let next_item = iter.try_next()?;
            match next_item {
                Some(kv) => ret.push((kv.0, kv.1)),
                None => break,
            }
        }
        Ok(ret)
    }

    /// Import data from an export iterator.
    ///
    /// This imports data from an iterator returned by `export()`.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store1 = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store1.put(&1, &2).unwrap();
    /// store1.put(&3, &4).unwrap();
    /// store1.put(&5, &6).unwrap();
    /// let store2 = Store::<usize, usize>::open_temporary(1000).unwrap();
    ///
    /// let export = store1.export().unwrap();
    /// store2.import(export).unwrap();
    ///
    /// assert_eq!(store1.to_map().unwrap(), store2.to_map().unwrap());
    /// ```
    pub fn import<I>(&self, iter: I) -> Result<usize>
    where
        I: Iterator<Item = (K, V)>,
    {
        let mut imported = 0;
        for kv in iter {
            self.put(&kv.0, &kv.1)?;
            imported += 1;
        }
        Ok(imported)
    }

    /// Import data from an iterator.
    ///
    /// This imports data from an iterator returned by `iter()`
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store1 = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store1.put(&1, &2).unwrap();
    /// store1.put(&3, &4).unwrap();
    /// store1.put(&5, &6).unwrap();
    /// let store2 = Store::<usize, usize>::open_temporary(1000).unwrap();
    ///
    /// let iter = store1.iter().unwrap();
    /// store2.import_iter(iter).unwrap();
    ///
    /// assert_eq!(store1.to_map().unwrap(), store2.to_map().unwrap());
    /// ```
    pub fn import_iter<I>(&self, iter: I) -> Result<usize>
    where
        I: Iterator<Item = Result<(K, V)>>,
    {
        let mut imported = 0;
        for item in iter {
            let kv = item?;
            self.put(&kv.0, &kv.1)?;
            imported += 1;
        }
        Ok(imported)
    }

    /// Import data from a map iterator.
    ///
    /// This imports data from an iterator that would typically
    /// be returned by a map.
    ///
    /// ```
    /// use menhirkv::Store;
    /// use std::collections::HashMap;
    ///
    /// let mut map: HashMap<usize, usize> = HashMap::new();
    /// map.insert(1, 2);
    /// map.insert(3, 4);
    /// map.insert(5, 6);
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    ///
    /// store.import_map_iter(map.iter()).unwrap();
    ///
    /// assert_eq!(map, store.to_map().unwrap());
    /// ```
    pub fn import_map_iter<'a, I>(&self, iter: I) -> Result<usize>
    where
        K: Serialize + DeserializeOwned + Eq + 'a,
        V: Serialize + DeserializeOwned + 'a,
        I: Iterator<Item = (&'a K, &'a V)>,
    {
        let mut imported = 0;
        for kv in iter {
            self.put(kv.0, kv.1)?;
            imported += 1;
        }
        Ok(imported)
    }

    /// Import data from a vec iterator.
    ///
    /// This imports data from an iterator that would typically
    /// be returned by a vec.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let mut vec: Vec<(usize, usize)> = Vec::new();
    /// vec.push((1, 2));
    /// vec.push((3, 4));
    /// vec.push((5, 6));
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    ///
    /// store.import_vec_iter(vec.iter()).unwrap();
    ///
    /// assert_eq!(3, store.len().unwrap());
    /// ```
    pub fn import_vec_iter<'a, I>(&self, iter: I) -> Result<usize>
    where
        K: Serialize + DeserializeOwned + 'a,
        V: Serialize + DeserializeOwned + 'a,
        I: Iterator<Item = &'a (K, V)>,
    {
        let mut imported = 0;
        for kv in iter {
            self.put(&kv.0, &kv.1)?;
            imported += 1;
        }
        Ok(imported)
    }

    /// Dumps all content into a (de)serializable object.
    ///
    /// This is a bit different from a vec export, as it also:
    /// * exports the capacity
    /// * makes it easy to (de)serialize
    ///
    /// This can be useful for some cold storage of the store.
    ///
    ///  # Examples
    ///
    /// ```
    /// use menhirkv::Store;
    /// use serde_json::json;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&1, &2).unwrap();
    /// store.put(&3, &4).unwrap();
    /// let dump = store.dump().unwrap();
    /// let js = json!(dump);
    /// assert_eq!("{\"capacity\":100,\"data\":{\"\":[[1,2],[3,4]]}}", js.to_string());
    /// ```
    pub fn dump(&self) -> Result<Dump<K, V>> {
        let mut data = HashMap::new();
        for name in self.iter_cf_names() {
            let cf = self.cf(name)?;
            let vec = cf.to_vec()?;
            data.insert(String::from(name), vec);
        }

        Ok(Dump {
            capacity: self.capacity(),
            data,
        })
    }

    /// Restores all contents.
    ///
    /// This is a bit different from other imports:
    ///
    /// * it imports the capacity
    /// * it clears the store before importing
    ///
    /// This can be useful for complete restore from a cold storage.
    /// Column families must be the same. Note that the main, default
    /// column family is exported with a name of `""` (empty string).
    ///
    ///  # Examples
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store1 = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store1.put(&1, &2).unwrap();
    /// store1.put(&3, &4).unwrap();
    /// let dump = store1.dump().unwrap();
    /// let store2 = Store::<usize, usize>::open_temporary(1_000).unwrap();
    ///
    /// store2.restore(&dump).unwrap();
    ///
    /// assert_eq!(100, store2.capacity());
    /// assert_eq!(2, store2.len().unwrap());
    /// ```
    pub fn restore(&self, dump: &Dump<K, V>) -> Result<usize> {
        let len_cfs = self.iter_cf_names().count();
        if dump.data.len() != len_cfs {
            return Err(Error::invalid_data("column families do not match"));
        }
        self.resize(dump.capacity)?;

        let mut count: usize = 0;
        for name in self.iter_cf_names() {
            let cf = self.cf(name)?;
            cf.clear()?;
            match dump.data.get(name) {
                Some(data) => {
                    cf.import_vec_iter(data.iter())?;
                    count += data.len();
                }
                None => {
                    return Err(Error::invalid_data(&format!(
                        "missing column family \"{}\"",
                        name,
                    )))
                }
            }
        }
        Ok(count)
    }
}

impl<K, V> Store<K, V>
where
    K: Serialize + DeserializeOwned + Hash + Eq,
    V: Serialize + DeserializeOwned,
{
    /// Export all items in a map.
    ///
    /// You could do that with an iterator, but either check for
    /// errors at each iteration, or take the risk of a `panic()`.
    ///
    /// This utility just dumps everything, the cost being that
    /// all items must fit in memory.
    ///
    /// If you want something failsafe, use the standard iterator.
    /// This one is easier to use, but can crash your program.
    ///
    /// ```
    /// use menhirkv::Store;
    ///
    /// let store = Store::<usize, usize>::open_temporary(100).unwrap();
    /// store.put(&1, &2).unwrap();
    /// store.put(&3, &4).unwrap();
    /// store.put(&5, &6).unwrap();
    /// let map = store.to_map().unwrap();
    /// assert_eq!(3, map.len());
    /// ```
    pub fn to_map(&self) -> Result<HashMap<K, V>> {
        let mut iter = self.export()?;
        let mut ret: HashMap<K, V> = HashMap::with_capacity(self.capacity());
        loop {
            let next_item = iter.try_next()?;
            match next_item {
                Some(kv) => {
                    ret.insert(kv.0, kv.1);
                }
                None => break,
            }
        }
        Ok(ret)
    }
}

impl<K, V> Drop for Store<K, V>
where
    K: Serialize + DeserializeOwned,
    V: Serialize + DeserializeOwned,
{
    fn drop(&mut self) {
        self.flush().unwrap_or({
            // Nothing we can really do here, apart from
            // saying explicitly in docs that you will not
            // get clues of this. We could output on stdout
            // but that package tends to be silent.
            //
            // It's really a best-effort, if it works, great.
            // If not, well, no luck.
            //
            // Users wanting to fully control that should
            // call flush() pro-actively at the time they
            // can control the result.
        });
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde::{Deserialize, Serialize};
    use std::collections::HashMap;

    const BLOB_SIZE: usize = 10_000;
    #[derive(Serialize, Deserialize)]
    struct Blob {
        bytes: Vec<u8>,
    }

    impl Blob {
        fn new() -> Self {
            Blob {
                bytes: [42; BLOB_SIZE].to_vec(),
            }
        }
    }

    #[derive(Serialize, Deserialize)]
    struct Person {
        name: String,
        age: usize,
    }

    #[test]
    fn test_open_temporary() {
        let store: Store<String, String> = Store::open_temporary(100).unwrap();
        assert_eq!(100, store.capacity());
        assert_eq!(0, store.len().unwrap());
    }

    #[test]
    fn test_basic_put_get() {
        let store: Store<String, usize> = Store::open_temporary(100).unwrap();
        let k = String::from("key");
        let v1 = 421;
        store.put(&k, &v1).expect("data should be put");
        let v2 = store.get(&k).expect("data should be get");
        assert_eq!(Some(v1), v2);
    }

    #[test]
    fn test_usize() {
        let store: Store<usize, usize> = Store::open_temporary(100).unwrap();
        store.put(&123, &456).unwrap();
        assert_eq!(Some(456), store.get(&123).unwrap());
    }

    #[test]
    fn test_person() {
        let store: Store<usize, Person> = Store::open_temporary(100).unwrap();
        store
            .put(
                &42,
                &Person {
                    name: String::from("Ford Escort"),
                    age: 33,
                },
            )
            .expect("put should work");
        let who = store.get(&42).expect("get should work").unwrap();
        assert_eq!(33, who.age);
    }

    #[test]
    fn test_basic_len() {
        let store: Store<usize, String> = Store::open_temporary(100).unwrap();
        assert_eq!(0, store.len().unwrap());
        store.put(&1, &String::from("Alice")).unwrap();
        store.put(&2, &String::from("Bob")).unwrap();
        store.put(&3, &String::from("Oscar")).unwrap();
        assert_eq!(3, store.len().unwrap());
    }

    #[test]
    fn test_basic_iter() {
        let store: Store<String, usize> = Store::open_temporary(100).unwrap();
        let mut source: HashMap<String, usize> = HashMap::new();
        source.insert(String::from("a"), 10);
        source.insert(String::from("b"), 20);
        source.insert(String::from("c"), 30);
        for kv in &source {
            store.put(kv.0, kv.1).expect("data should be put");
        }
        let mut check: HashMap<String, usize> = HashMap::new();
        for item in store.iter().unwrap() {
            let kv = item.unwrap();
            check.insert(kv.0, kv.1);
        }
        assert_eq!(source, check);
    }

    #[test]
    fn test_compact() {
        const N: usize = 100;
        const M: usize = 100;
        let store: Store<usize, Blob> = Store::open_temporary(N).unwrap();
        store.compact().unwrap();

        for i in 0..M * N {
            store.put(&i, &Blob::new()).unwrap();
        }

        let len = store.len().unwrap();
        let db_len = store.db_len().unwrap();
        println!(
            "before compaction len: {}, db_len: {}, capacity: {}",
            len,
            db_len,
            store.capacity()
        );
        assert!(len < db_len);

        // Check all the various iterators are capped to len
        // and do not report db_len.
        assert_eq!(len, store.iter().unwrap().count());
        assert_eq!(len, store.keys().unwrap().count());
        assert_eq!(len, store.values().unwrap().count());
        assert_eq!(len, store.export().unwrap().count());
        assert_eq!(len, store.export_keys().unwrap().count());
        assert_eq!(len, store.export_values().unwrap().count());
        let vec = store.to_vec().unwrap();
        assert_eq!(len, vec.len());
        let map = store.to_map().unwrap();
        assert_eq!(len, map.len());

        // Normally RocksDB does this for us in the background,
        // but we trigger it for the sake of testing.
        store.compact().unwrap();

        let len = store.len().unwrap();
        let db_len = store.db_len().unwrap();
        println!(
            "after compaction len: {}, db_len: {}, capacity: {}",
            len,
            db_len,
            store.capacity()
        );

        assert_eq!(len, db_len);
        assert!(len < M * N);
    }

    #[test]
    fn test_threads() {
        use std::thread;

        let store: Store<usize, usize> = Store::open_temporary(30_000).unwrap();

        let s = store.clone();
        let handle1 = thread::spawn(move || {
            for i in 0..10_000 {
                let j: usize = i * 2;
                s.put(&j, &j).unwrap();
            }
        });

        let s = store.clone();
        let handle2 = thread::spawn(move || {
            for i in 0..10_000 {
                let j: usize = i * 2 + 1;
                s.put(&j, &j).unwrap();
            }
        });

        handle1.join().unwrap();
        handle2.join().unwrap();

        assert_eq!(Some(19_998), store.get(&19_998).unwrap());
        assert_eq!(Some(19_999), store.get(&19_999).unwrap());
    }

    #[test]
    fn get_bumps_entry() {
        let store = Store::<usize, usize>::open_temporary(100).unwrap();
        let mut last_777: Option<usize> = None;
        for i in 1..10_000 {
            store.put(&i, &(i * 10)).unwrap();
            last_777 = store.get(&777).unwrap();
        }
        assert_eq!(Some(7_770), last_777);
        assert_eq!(Some(99_990), store.get(&9_999).unwrap());

        // peek/get always return the value until *COMPACTION* has run,
        // which never happens on those small keys. To double-verify,
        // let's just iterate on keys, because iterators respect
        // the bloom filter.
        let map = store.to_map().unwrap();
        assert!(map.contains_key(&777));
    }

    #[test]
    fn peek_does_not_bump_entry() {
        let store = Store::<usize, usize>::open_temporary(100).unwrap();
        for i in 1..1_000_000 {
            store.put(&i, &(i * 10)).unwrap();
            store.peek(&777).unwrap();
            store.peek(&888).unwrap();
        }
        assert_eq!(Some(9_999_990), store.peek(&999_999).unwrap());

        let map = store.to_map().unwrap();
        assert!(!(map.contains_key(&777) && map.contains_key(&888)));
    }

    #[test]
    fn dump_restore_multiple_cf() {
        let store =
            Store::<usize, usize>::open_cf_temporary(&["red", "green", "blue"], 100).unwrap();
        store.put(&1, &10).unwrap();
        store.put(&2, &20).unwrap();
        let store_red = store.cf("red").unwrap();
        store_red.put(&3, &30).unwrap();
        store_red.put(&4, &40).unwrap();
        let store_green = store.cf("green").unwrap();
        store_green.put(&5, &50).unwrap();
        let store_blue = store.cf("blue").unwrap();
        let dump = store_blue.dump().unwrap();

        assert_eq!(100, dump.capacity);
        assert!(dump.data.contains_key(&String::from("red")));
        assert!(dump.data.contains_key(&String::from("green")));
        assert!(dump.data.contains_key(&String::from("blue")));

        let restored =
            Store::<usize, usize>::open_cf_temporary(&["red", "blue", "green"], 1000).unwrap();
        restored.put(&3, &300).unwrap();

        assert_eq!(5, restored.restore(&dump).unwrap());
        assert_eq!(100, restored.capacity());
        assert_eq!(Some(10), restored.get(&1).unwrap());
        assert_eq!(Some(20), restored.get(&2).unwrap());
        assert_eq!(None, restored.get(&3).unwrap());
        assert_eq!(2, restored.db_len().unwrap());

        let restored_red = restored.cf("red").unwrap();
        assert_eq!(100, restored_red.capacity());
        assert_eq!(Some(30), restored_red.get(&3).unwrap());
        assert_eq!(Some(40), restored_red.get(&4).unwrap());
        assert_eq!(2, restored_red.db_len().unwrap());

        let restored_green = restored.cf("green").unwrap();
        assert_eq!(100, restored_green.capacity());
        assert_eq!(Some(50), restored_green.get(&5).unwrap());
        assert_eq!(1, restored_green.db_len().unwrap());

        let restored_blue = restored.cf("blue").unwrap();
        assert_eq!(100, restored_blue.capacity());
        assert_eq!(0, restored_blue.db_len().unwrap());

        let bad_columns =
            Store::<usize, usize>::open_cf_temporary(&["red", "green", "yellow"], 1000).unwrap();
        bad_columns.restore(&dump).unwrap_err();
    }
}