Modularize beacon node backend (#4718)

#4669


  Modularize the beacon node backend to make it easier to add new database implementations

beacon_node/store/src/database/interface.rs

@@ -0,0 +1,220 @@
+#[cfg(feature = "leveldb")]
+use crate::database::leveldb_impl;
+#[cfg(feature = "redb")]
+use crate::database::redb_impl;
+use crate::{config::DatabaseBackend, KeyValueStoreOp, StoreConfig};
+use crate::{metrics, ColumnIter, ColumnKeyIter, DBColumn, Error, ItemStore, Key, KeyValueStore};
+use std::collections::HashSet;
+use std::path::Path;
+use types::EthSpec;
+
+pub enum BeaconNodeBackend<E: EthSpec> {
+    #[cfg(feature = "leveldb")]
+    LevelDb(leveldb_impl::LevelDB<E>),
+    #[cfg(feature = "redb")]
+    Redb(redb_impl::Redb<E>),
+}
+
+impl<E: EthSpec> ItemStore<E> for BeaconNodeBackend<E> {}
+
+impl<E: EthSpec> KeyValueStore<E> for BeaconNodeBackend<E> {
+    fn get_bytes(&self, column: DBColumn, key: &[u8]) -> Result<Option<Vec<u8>>, Error> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::get_bytes(txn, column, key),
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::get_bytes(txn, column, key),
+        }
+    }
+
+    fn put_bytes(&self, column: DBColumn, key: &[u8], value: &[u8]) -> Result<(), Error> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::put_bytes_with_options(
+                txn,
+                column,
+                key,
+                value,
+                txn.write_options(),
+            ),
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::put_bytes_with_options(
+                txn,
+                column,
+                key,
+                value,
+                txn.write_options(),
+            ),
+        }
+    }
+
+    fn put_bytes_sync(&self, column: DBColumn, key: &[u8], value: &[u8]) -> Result<(), Error> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::put_bytes_with_options(
+                txn,
+                column,
+                key,
+                value,
+                txn.write_options_sync(),
+            ),
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::put_bytes_with_options(
+                txn,
+                column,
+                key,
+                value,
+                txn.write_options_sync(),
+            ),
+        }
+    }
+
+    fn sync(&self) -> Result<(), Error> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::sync(txn),
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::sync(txn),
+        }
+    }
+
+    fn key_exists(&self, column: DBColumn, key: &[u8]) -> Result<bool, Error> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::key_exists(txn, column, key),
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::key_exists(txn, column, key),
+        }
+    }
+
+    fn key_delete(&self, column: DBColumn, key: &[u8]) -> Result<(), Error> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::key_delete(txn, column, key),
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::key_delete(txn, column, key),
+        }
+    }
+
+    fn do_atomically(&self, batch: Vec<KeyValueStoreOp>) -> Result<(), Error> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::do_atomically(txn, batch),
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::do_atomically(txn, batch),
+        }
+    }
+
+    fn begin_rw_transaction(&self) -> parking_lot::MutexGuard<()> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::begin_rw_transaction(txn),
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::begin_rw_transaction(txn),
+        }
+    }
+
+    fn compact(&self) -> Result<(), Error> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::compact(txn),
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::compact(txn),
+        }
+    }
+
+    fn iter_column_keys_from<K: Key>(&self, _column: DBColumn, from: &[u8]) -> ColumnKeyIter<K> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => {
+                leveldb_impl::LevelDB::iter_column_keys_from(txn, _column, from)
+            }
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => {
+                redb_impl::Redb::iter_column_keys_from(txn, _column, from)
+            }
+        }
+    }
+
+    fn iter_column_keys<K: Key>(&self, column: DBColumn) -> ColumnKeyIter<K> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::iter_column_keys(txn, column),
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::iter_column_keys(txn, column),
+        }
+    }
+
+    fn iter_column_from<K: Key>(&self, column: DBColumn, from: &[u8]) -> ColumnIter<K> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => {
+                leveldb_impl::LevelDB::iter_column_from(txn, column, from)
+            }
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::iter_column_from(txn, column, from),
+        }
+    }
+
+    fn compact_column(&self, _column: DBColumn) -> Result<(), Error> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::compact_column(txn, _column),
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::compact(txn),
+        }
+    }
+
+    fn delete_batch(&self, col: DBColumn, ops: HashSet<&[u8]>) -> Result<(), Error> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::delete_batch(txn, col, ops),
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::delete_batch(txn, col, ops),
+        }
+    }
+
+    fn delete_if(
+        &self,
+        column: DBColumn,
+        f: impl FnMut(&[u8]) -> Result<bool, Error>,
+    ) -> Result<(), Error> {
+        match self {
+            #[cfg(feature = "leveldb")]
+            BeaconNodeBackend::LevelDb(txn) => leveldb_impl::LevelDB::delete_if(txn, column, f),
+            #[cfg(feature = "redb")]
+            BeaconNodeBackend::Redb(txn) => redb_impl::Redb::delete_if(txn, column, f),
+        }
+    }
+}
+
+impl<E: EthSpec> BeaconNodeBackend<E> {
+    pub fn open(config: &StoreConfig, path: &Path) -> Result<Self, Error> {
+        metrics::inc_counter_vec(&metrics::DISK_DB_TYPE, &[&config.backend.to_string()]);
+        match config.backend {
+            #[cfg(feature = "leveldb")]
+            DatabaseBackend::LevelDb => {
+                leveldb_impl::LevelDB::open(path).map(BeaconNodeBackend::LevelDb)
+            }
+            #[cfg(feature = "redb")]
+            DatabaseBackend::Redb => redb_impl::Redb::open(path).map(BeaconNodeBackend::Redb),
+        }
+    }
+}
+
+pub struct WriteOptions {
+    /// fsync before acknowledging a write operation.
+    pub sync: bool,
+}
+
+impl WriteOptions {
+    pub fn new() -> Self {
+        WriteOptions { sync: false }
+    }
+}
+
+impl Default for WriteOptions {
+    fn default() -> Self {
+        Self::new()
+    }
+}

beacon_node/store/src/database/leveldb_impl.rs

@@ -0,0 +1,304 @@
+use crate::hot_cold_store::{BytesKey, HotColdDBError};
+use crate::Key;
+use crate::{
+    get_key_for_col, metrics, ColumnIter, ColumnKeyIter, DBColumn, Error, KeyValueStoreOp,
+};
+use leveldb::{
+    compaction::Compaction,
+    database::{
+        batch::{Batch, Writebatch},
+        kv::KV,
+        Database,
+    },
+    iterator::{Iterable, LevelDBIterator},
+    options::{Options, ReadOptions},
+};
+use parking_lot::{Mutex, MutexGuard};
+use std::collections::HashSet;
+use std::marker::PhantomData;
+use std::path::Path;
+use types::{EthSpec, FixedBytesExtended, Hash256};
+
+use super::interface::WriteOptions;
+
+pub struct LevelDB<E: EthSpec> {
+    db: Database<BytesKey>,
+    /// A mutex to synchronise sensitive read-write transactions.
+    transaction_mutex: Mutex<()>,
+    _phantom: PhantomData<E>,
+}
+
+impl From<WriteOptions> for leveldb::options::WriteOptions {
+    fn from(options: WriteOptions) -> Self {
+        let mut opts = leveldb::options::WriteOptions::new();
+        opts.sync = options.sync;
+        opts
+    }
+}
+
+impl<E: EthSpec> LevelDB<E> {
+    pub fn open(path: &Path) -> Result<Self, Error> {
+        let mut options = Options::new();
+
+        options.create_if_missing = true;
+
+        let db = Database::open(path, options)?;
+        let transaction_mutex = Mutex::new(());
+
+        Ok(Self {
+            db,
+            transaction_mutex,
+            _phantom: PhantomData,
+        })
+    }
+
+    pub fn read_options(&self) -> ReadOptions<BytesKey> {
+        ReadOptions::new()
+    }
+
+    pub fn write_options(&self) -> WriteOptions {
+        WriteOptions::new()
+    }
+
+    pub fn write_options_sync(&self) -> WriteOptions {
+        let mut opts = WriteOptions::new();
+        opts.sync = true;
+        opts
+    }
+
+    pub fn put_bytes_with_options(
+        &self,
+        col: DBColumn,
+        key: &[u8],
+        val: &[u8],
+        opts: WriteOptions,
+    ) -> Result<(), Error> {
+        let column_key = get_key_for_col(col, key);
+
+        metrics::inc_counter_vec(&metrics::DISK_DB_WRITE_COUNT, &[col.into()]);
+        metrics::inc_counter_vec_by(
+            &metrics::DISK_DB_WRITE_BYTES,
+            &[col.into()],
+            val.len() as u64,
+        );
+        let timer = metrics::start_timer(&metrics::DISK_DB_WRITE_TIMES);
+
+        self.db
+            .put(opts.into(), BytesKey::from_vec(column_key), val)
+            .map_err(Into::into)
+            .map(|()| {
+                metrics::stop_timer(timer);
+            })
+    }
+
+    /// Store some `value` in `column`, indexed with `key`.
+    pub fn put_bytes(&self, col: DBColumn, key: &[u8], val: &[u8]) -> Result<(), Error> {
+        self.put_bytes_with_options(col, key, val, self.write_options())
+    }
+
+    pub fn put_bytes_sync(&self, col: DBColumn, key: &[u8], val: &[u8]) -> Result<(), Error> {
+        self.put_bytes_with_options(col, key, val, self.write_options_sync())
+    }
+
+    pub fn sync(&self) -> Result<(), Error> {
+        self.put_bytes_sync(DBColumn::Dummy, b"sync", b"sync")
+    }
+
+    // Retrieve some bytes in `column` with `key`.
+    pub fn get_bytes(&self, col: DBColumn, key: &[u8]) -> Result<Option<Vec<u8>>, Error> {
+        let column_key = get_key_for_col(col, key);
+
+        metrics::inc_counter_vec(&metrics::DISK_DB_READ_COUNT, &[col.into()]);
+        let timer = metrics::start_timer(&metrics::DISK_DB_READ_TIMES);
+
+        self.db
+            .get(self.read_options(), BytesKey::from_vec(column_key))
+            .map_err(Into::into)
+            .map(|opt| {
+                opt.inspect(|bytes| {
+                    metrics::inc_counter_vec_by(
+                        &metrics::DISK_DB_READ_BYTES,
+                        &[col.into()],
+                        bytes.len() as u64,
+                    );
+                    metrics::stop_timer(timer);
+                })
+            })
+    }
+
+    /// Return `true` if `key` exists in `column`.
+    pub fn key_exists(&self, col: DBColumn, key: &[u8]) -> Result<bool, Error> {
+        let column_key = get_key_for_col(col, key);
+
+        metrics::inc_counter_vec(&metrics::DISK_DB_EXISTS_COUNT, &[col.into()]);
+
+        self.db
+            .get(self.read_options(), BytesKey::from_vec(column_key))
+            .map_err(Into::into)
+            .map(|val| val.is_some())
+    }
+
+    /// Removes `key` from `column`.
+    pub fn key_delete(&self, col: DBColumn, key: &[u8]) -> Result<(), Error> {
+        let column_key = get_key_for_col(col, key);
+
+        metrics::inc_counter_vec(&metrics::DISK_DB_DELETE_COUNT, &[col.into()]);
+
+        self.db
+            .delete(self.write_options().into(), BytesKey::from_vec(column_key))
+            .map_err(Into::into)
+    }
+
+    pub fn do_atomically(&self, ops_batch: Vec<KeyValueStoreOp>) -> Result<(), Error> {
+        let mut leveldb_batch = Writebatch::new();
+        for op in ops_batch {
+            match op {
+                KeyValueStoreOp::PutKeyValue(col, key, value) => {
+                    let _timer = metrics::start_timer(&metrics::DISK_DB_WRITE_TIMES);
+                    metrics::inc_counter_vec_by(
+                        &metrics::DISK_DB_WRITE_BYTES,
+                        &[col.into()],
+                        value.len() as u64,
+                    );
+                    metrics::inc_counter_vec(&metrics::DISK_DB_WRITE_COUNT, &[col.into()]);
+                    let column_key = get_key_for_col(col, &key);
+                    leveldb_batch.put(BytesKey::from_vec(column_key), &value);
+                }
+
+                KeyValueStoreOp::DeleteKey(col, key) => {
+                    let _timer = metrics::start_timer(&metrics::DISK_DB_DELETE_TIMES);
+                    metrics::inc_counter_vec(&metrics::DISK_DB_DELETE_COUNT, &[col.into()]);
+                    let column_key = get_key_for_col(col, &key);
+                    leveldb_batch.delete(BytesKey::from_vec(column_key));
+                }
+            }
+        }
+        self.db.write(self.write_options().into(), &leveldb_batch)?;
+        Ok(())
+    }
+
+    pub fn begin_rw_transaction(&self) -> MutexGuard<()> {
+        self.transaction_mutex.lock()
+    }
+
+    /// Compact all values in the states and states flag columns.
+    pub fn compact(&self) -> Result<(), Error> {
+        let _timer = metrics::start_timer(&metrics::DISK_DB_COMPACT_TIMES);
+        let endpoints = |column: DBColumn| {
+            (
+                BytesKey::from_vec(get_key_for_col(column, Hash256::zero().as_slice())),
+                BytesKey::from_vec(get_key_for_col(
+                    column,
+                    Hash256::repeat_byte(0xff).as_slice(),
+                )),
+            )
+        };
+
+        for (start_key, end_key) in [
+            endpoints(DBColumn::BeaconStateTemporary),
+            endpoints(DBColumn::BeaconState),
+            endpoints(DBColumn::BeaconStateSummary),
+        ] {
+            self.db.compact(&start_key, &end_key);
+        }
+
+        Ok(())
+    }
+
+    pub fn compact_column(&self, column: DBColumn) -> Result<(), Error> {
+        // Use key-size-agnostic keys [] and 0xff..ff with a minimum of 32 bytes to account for
+        // columns that may change size between sub-databases or schema versions.
+        let start_key = BytesKey::from_vec(get_key_for_col(column, &[]));
+        let end_key = BytesKey::from_vec(get_key_for_col(
+            column,
+            &vec![0xff; std::cmp::max(column.key_size(), 32)],
+        ));
+        self.db.compact(&start_key, &end_key);
+        Ok(())
+    }
+
+    pub fn iter_column_from<K: Key>(&self, column: DBColumn, from: &[u8]) -> ColumnIter<K> {
+        let start_key = BytesKey::from_vec(get_key_for_col(column, from));
+        let iter = self.db.iter(self.read_options());
+        iter.seek(&start_key);
+
+        Box::new(
+            iter.take_while(move |(key, _)| key.matches_column(column))
+                .map(move |(bytes_key, value)| {
+                    metrics::inc_counter_vec(&metrics::DISK_DB_READ_COUNT, &[column.into()]);
+                    metrics::inc_counter_vec_by(
+                        &metrics::DISK_DB_READ_BYTES,
+                        &[column.into()],
+                        value.len() as u64,
+                    );
+                    let key = bytes_key.remove_column_variable(column).ok_or_else(|| {
+                        HotColdDBError::IterationError {
+                            unexpected_key: bytes_key.clone(),
+                        }
+                    })?;
+                    Ok((K::from_bytes(key)?, value))
+                }),
+        )
+    }
+
+    pub fn iter_column_keys_from<K: Key>(&self, column: DBColumn, from: &[u8]) -> ColumnKeyIter<K> {
+        let start_key = BytesKey::from_vec(get_key_for_col(column, from));
+
+        let iter = self.db.keys_iter(self.read_options());
+        iter.seek(&start_key);
+
+        Box::new(
+            iter.take_while(move |key| key.matches_column(column))
+                .map(move |bytes_key| {
+                    metrics::inc_counter_vec(&metrics::DISK_DB_KEY_READ_COUNT, &[column.into()]);
+                    metrics::inc_counter_vec_by(
+                        &metrics::DISK_DB_KEY_READ_BYTES,
+                        &[column.into()],
+                        bytes_key.key.len() as u64,
+                    );
+                    let key = &bytes_key.key[column.as_bytes().len()..];
+                    K::from_bytes(key)
+                }),
+        )
+    }
+
+    /// Iterate through all keys and values in a particular column.
+    pub fn iter_column_keys<K: Key>(&self, column: DBColumn) -> ColumnKeyIter<K> {
+        self.iter_column_keys_from(column, &vec![0; column.key_size()])
+    }
+
+    pub fn iter_column<K: Key>(&self, column: DBColumn) -> ColumnIter<K> {
+        self.iter_column_from(column, &vec![0; column.key_size()])
+    }
+
+    pub fn delete_batch(&self, col: DBColumn, ops: HashSet<&[u8]>) -> Result<(), Error> {
+        let mut leveldb_batch = Writebatch::new();
+        for op in ops {
+            let column_key = get_key_for_col(col, op);
+            leveldb_batch.delete(BytesKey::from_vec(column_key));
+        }
+        self.db.write(self.write_options().into(), &leveldb_batch)?;
+        Ok(())
+    }
+
+    pub fn delete_if(
+        &self,
+        column: DBColumn,
+        mut f: impl FnMut(&[u8]) -> Result<bool, Error>,
+    ) -> Result<(), Error> {
+        let mut leveldb_batch = Writebatch::new();
+        let iter = self.db.iter(self.read_options());
+
+        iter.take_while(move |(key, _)| key.matches_column(column))
+            .for_each(|(key, value)| {
+                if f(&value).unwrap_or(false) {
+                    let _timer = metrics::start_timer(&metrics::DISK_DB_DELETE_TIMES);
+                    metrics::inc_counter_vec(&metrics::DISK_DB_DELETE_COUNT, &[column.into()]);
+                    leveldb_batch.delete(key);
+                }
+            });
+
+        self.db.write(self.write_options().into(), &leveldb_batch)?;
+        Ok(())
+    }
+}

beacon_node/store/src/database/redb_impl.rs

@@ -0,0 +1,314 @@
+use crate::{metrics, ColumnIter, ColumnKeyIter, Key};
+use crate::{DBColumn, Error, KeyValueStoreOp};
+use parking_lot::{Mutex, MutexGuard, RwLock};
+use redb::TableDefinition;
+use std::collections::HashSet;
+use std::{borrow::BorrowMut, marker::PhantomData, path::Path};
+use strum::IntoEnumIterator;
+use types::EthSpec;
+
+use super::interface::WriteOptions;
+
+pub const DB_FILE_NAME: &str = "database.redb";
+
+pub struct Redb<E: EthSpec> {
+    db: RwLock<redb::Database>,
+    transaction_mutex: Mutex<()>,
+    _phantom: PhantomData<E>,
+}
+
+impl From<WriteOptions> for redb::Durability {
+    fn from(options: WriteOptions) -> Self {
+        if options.sync {
+            redb::Durability::Immediate
+        } else {
+            redb::Durability::Eventual
+        }
+    }
+}
+
+impl<E: EthSpec> Redb<E> {
+    pub fn open(path: &Path) -> Result<Self, Error> {
+        let db_file = path.join(DB_FILE_NAME);
+        let db = redb::Database::create(db_file)?;
+        let transaction_mutex = Mutex::new(());
+
+        for column in DBColumn::iter() {
+            Redb::<E>::create_table(&db, column.into())?;
+        }
+
+        Ok(Self {
+            db: db.into(),
+            transaction_mutex,
+            _phantom: PhantomData,
+        })
+    }
+
+    fn create_table(db: &redb::Database, table_name: &str) -> Result<(), Error> {
+        let table_definition: TableDefinition<'_, &[u8], &[u8]> = TableDefinition::new(table_name);
+        let tx = db.begin_write()?;
+        tx.open_table(table_definition)?;
+        tx.commit().map_err(Into::into)
+    }
+
+    pub fn write_options(&self) -> WriteOptions {
+        WriteOptions::new()
+    }
+
+    pub fn write_options_sync(&self) -> WriteOptions {
+        let mut opts = WriteOptions::new();
+        opts.sync = true;
+        opts
+    }
+
+    pub fn begin_rw_transaction(&self) -> MutexGuard<()> {
+        self.transaction_mutex.lock()
+    }
+
+    pub fn put_bytes_with_options(
+        &self,
+        col: DBColumn,
+        key: &[u8],
+        val: &[u8],
+        opts: WriteOptions,
+    ) -> Result<(), Error> {
+        metrics::inc_counter_vec(&metrics::DISK_DB_WRITE_COUNT, &[col.into()]);
+        metrics::inc_counter_vec_by(
+            &metrics::DISK_DB_WRITE_BYTES,
+            &[col.into()],
+            val.len() as u64,
+        );
+        let timer = metrics::start_timer(&metrics::DISK_DB_WRITE_TIMES);
+
+        let table_definition: TableDefinition<'_, &[u8], &[u8]> = TableDefinition::new(col.into());
+        let open_db = self.db.read();
+        let mut tx = open_db.begin_write()?;
+        tx.set_durability(opts.into());
+        let mut table = tx.open_table(table_definition)?;
+
+        table.insert(key, val).map(|_| {
+            metrics::stop_timer(timer);
+        })?;
+        drop(table);
+        tx.commit().map_err(Into::into)
+    }
+
+    /// Store some `value` in `column`, indexed with `key`.
+    pub fn put_bytes(&self, col: DBColumn, key: &[u8], val: &[u8]) -> Result<(), Error> {
+        self.put_bytes_with_options(col, key, val, self.write_options())
+    }
+
+    pub fn put_bytes_sync(&self, col: DBColumn, key: &[u8], val: &[u8]) -> Result<(), Error> {
+        self.put_bytes_with_options(col, key, val, self.write_options_sync())
+    }
+
+    pub fn sync(&self) -> Result<(), Error> {
+        self.put_bytes_sync(DBColumn::Dummy, b"sync", b"sync")
+    }
+
+    // Retrieve some bytes in `column` with `key`.
+    pub fn get_bytes(&self, col: DBColumn, key: &[u8]) -> Result<Option<Vec<u8>>, Error> {
+        metrics::inc_counter_vec(&metrics::DISK_DB_READ_COUNT, &[col.into()]);
+        let timer = metrics::start_timer(&metrics::DISK_DB_READ_TIMES);
+
+        let table_definition: TableDefinition<'_, &[u8], &[u8]> = TableDefinition::new(col.into());
+        let open_db = self.db.read();
+        let tx = open_db.begin_read()?;
+        let table = tx.open_table(table_definition)?;
+
+        let result = table.get(key)?;
+
+        match result {
+            Some(access_guard) => {
+                let value = access_guard.value().to_vec();
+                metrics::inc_counter_vec_by(
+                    &metrics::DISK_DB_READ_BYTES,
+                    &[col.into()],
+                    value.len() as u64,
+                );
+                metrics::stop_timer(timer);
+                Ok(Some(value))
+            }
+            None => {
+                metrics::stop_timer(timer);
+                Ok(None)
+            }
+        }
+    }
+
+    /// Return `true` if `key` exists in `column`.
+    pub fn key_exists(&self, col: DBColumn, key: &[u8]) -> Result<bool, Error> {
+        metrics::inc_counter_vec(&metrics::DISK_DB_EXISTS_COUNT, &[col.into()]);
+
+        let table_definition: TableDefinition<'_, &[u8], &[u8]> = TableDefinition::new(col.into());
+        let open_db = self.db.read();
+        let tx = open_db.begin_read()?;
+        let table = tx.open_table(table_definition)?;
+
+        table
+            .get(key)
+            .map_err(Into::into)
+            .map(|access_guard| access_guard.is_some())
+    }
+
+    /// Removes `key` from `column`.
+    pub fn key_delete(&self, col: DBColumn, key: &[u8]) -> Result<(), Error> {
+        let table_definition: TableDefinition<'_, &[u8], &[u8]> = TableDefinition::new(col.into());
+        let open_db = self.db.read();
+        let tx = open_db.begin_write()?;
+        let mut table = tx.open_table(table_definition)?;
+        metrics::inc_counter_vec(&metrics::DISK_DB_DELETE_COUNT, &[col.into()]);
+
+        table.remove(key).map(|_| ())?;
+        drop(table);
+        tx.commit().map_err(Into::into)
+    }
+
+    pub fn do_atomically(&self, ops_batch: Vec<KeyValueStoreOp>) -> Result<(), Error> {
+        let open_db = self.db.read();
+        let mut tx = open_db.begin_write()?;
+        tx.set_durability(self.write_options().into());
+        for op in ops_batch {
+            match op {
+                KeyValueStoreOp::PutKeyValue(column, key, value) => {
+                    let _timer = metrics::start_timer(&metrics::DISK_DB_WRITE_TIMES);
+                    metrics::inc_counter_vec_by(
+                        &metrics::DISK_DB_WRITE_BYTES,
+                        &[column.into()],
+                        value.len() as u64,
+                    );
+                    metrics::inc_counter_vec(&metrics::DISK_DB_WRITE_COUNT, &[column.into()]);
+                    let table_definition: TableDefinition<'_, &[u8], &[u8]> =
+                        TableDefinition::new(column.into());
+
+                    let mut table = tx.open_table(table_definition)?;
+                    table.insert(key.as_slice(), value.as_slice())?;
+                    drop(table);
+                }
+
+                KeyValueStoreOp::DeleteKey(column, key) => {
+                    metrics::inc_counter_vec(&metrics::DISK_DB_DELETE_COUNT, &[column.into()]);
+                    let _timer = metrics::start_timer(&metrics::DISK_DB_DELETE_TIMES);
+                    let table_definition: TableDefinition<'_, &[u8], &[u8]> =
+                        TableDefinition::new(column.into());
+
+                    let mut table = tx.open_table(table_definition)?;
+                    table.remove(key.as_slice())?;
+                    drop(table);
+                }
+            }
+        }
+
+        tx.commit()?;
+        Ok(())
+    }
+
+    /// Compact all values in the states and states flag columns.
+    pub fn compact(&self) -> Result<(), Error> {
+        let _timer = metrics::start_timer(&metrics::DISK_DB_COMPACT_TIMES);
+        let mut open_db = self.db.write();
+        let mut_db = open_db.borrow_mut();
+        mut_db.compact().map_err(Into::into).map(|_| ())
+    }
+
+    pub fn iter_column_keys_from<K: Key>(&self, column: DBColumn, from: &[u8]) -> ColumnKeyIter<K> {
+        let table_definition: TableDefinition<'_, &[u8], &[u8]> =
+            TableDefinition::new(column.into());
+
+        let iter = {
+            let open_db = self.db.read();
+            let read_txn = open_db.begin_read()?;
+            let table = read_txn.open_table(table_definition)?;
+            table.range(from..)?.map(move |res| {
+                let (key, _) = res?;
+                metrics::inc_counter_vec(&metrics::DISK_DB_KEY_READ_COUNT, &[column.into()]);
+                metrics::inc_counter_vec_by(
+                    &metrics::DISK_DB_KEY_READ_BYTES,
+                    &[column.into()],
+                    key.value().len() as u64,
+                );
+                K::from_bytes(key.value())
+            })
+        };
+
+        Box::new(iter)
+    }
+
+    /// Iterate through all keys and values in a particular column.
+    pub fn iter_column_keys<K: Key>(&self, column: DBColumn) -> ColumnKeyIter<K> {
+        self.iter_column_keys_from(column, &vec![0; column.key_size()])
+    }
+
+    pub fn iter_column_from<K: Key>(&self, column: DBColumn, from: &[u8]) -> ColumnIter<K> {
+        let table_definition: TableDefinition<'_, &[u8], &[u8]> =
+            TableDefinition::new(column.into());
+
+        let prefix = from.to_vec();
+
+        let iter = {
+            let open_db = self.db.read();
+            let read_txn = open_db.begin_read()?;
+            let table = read_txn.open_table(table_definition)?;
+
+            table
+                .range(from..)?
+                .take_while(move |res| match res.as_ref() {
+                    Ok((_, _)) => true,
+                    Err(_) => false,
+                })
+                .map(move |res| {
+                    let (key, value) = res?;
+                    metrics::inc_counter_vec(&metrics::DISK_DB_READ_COUNT, &[column.into()]);
+                    metrics::inc_counter_vec_by(
+                        &metrics::DISK_DB_READ_BYTES,
+                        &[column.into()],
+                        value.value().len() as u64,
+                    );
+                    Ok((K::from_bytes(key.value())?, value.value().to_vec()))
+                })
+        };
+
+        Ok(Box::new(iter))
+    }
+
+    pub fn iter_column<K: Key>(&self, column: DBColumn) -> ColumnIter<K> {
+        self.iter_column_from(column, &vec![0; column.key_size()], |_, _| true)
+    }
+
+    pub fn delete_batch(&self, col: DBColumn, ops: HashSet<&[u8]>) -> Result<(), Error> {
+        let open_db = self.db.read();
+        let mut tx = open_db.begin_write()?;
+
+        tx.set_durability(redb::Durability::None);
+
+        let table_definition: TableDefinition<'_, &[u8], &[u8]> = TableDefinition::new(col.into());
+
+        let mut table = tx.open_table(table_definition)?;
+        table.retain(|key, _| !ops.contains(key))?;
+
+        drop(table);
+        tx.commit()?;
+        Ok(())
+    }
+
+    pub fn delete_if(
+        &self,
+        column: DBColumn,
+        mut f: impl FnMut(&[u8]) -> Result<bool, Error>,
+    ) -> Result<(), Error> {
+        let open_db = self.db.read();
+        let mut tx = open_db.begin_write()?;
+
+        tx.set_durability(redb::Durability::None);
+
+        let table_definition: TableDefinition<'_, &[u8], &[u8]> =
+            TableDefinition::new(column.into());
+
+        let mut table = tx.open_table(table_definition)?;
+        table.retain(|_, value| !f(value).unwrap_or(false))?;
+
+        drop(table);
+        tx.commit()?;
+        Ok(())
+    }
+}