armdb 0.4.1

sharded bitcask key-value storage optimized for NVMe
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

use std::marker::PhantomData;
use std::sync::Arc;

use super::access::IndexTarget;
use crate::TypedWriteHook;
use crate::key::Key;

/// Hook that keeps a secondary index in sync via a projection function.
///
/// Mutation matrix (spec §3): insert -> put; update with the same projection
/// key -> put (overwrite); update with a changed key -> delete old + put new;
/// delete -> delete. Index write errors are logged (the hook cannot return an
/// error) — the validator catches drift.
pub struct IndexHook<SK, SV, I: IndexTarget, P> {
    pub(super) index: Arc<I>,
    pub(super) project: P,
    pub(super) _pd: PhantomData<fn(&SK, &SV)>,
}

impl<SK, SV, I, P> TypedWriteHook<SK, SV> for IndexHook<SK, SV, I, P>
where
    SK: Key,
    SV: Send + Sync,
    I: IndexTarget,
    P: Fn(&SK, &SV) -> (I::K, I::V) + Send + Sync,
{
    const NEEDS_OLD_VALUE: bool = true;
    const NEEDS_WRITE: bool = true;
    const NEEDS_INIT: bool = false;

    fn on_write(&self, key: &SK, old: Option<&SV>, new: Option<&SV>) {
        let old_p = old.map(|o| (self.project)(key, o));
        let new_p = new.map(|n| (self.project)(key, n));
        match (old_p, new_p) {
            (Some((ok, _)), Some((nk, nv))) => {
                if ok.as_bytes() != nk.as_bytes()
                    && let Err(e) = self.index.delete_entry(&ok)
                {
                    tracing::error!(error = %e, "index hook: delete old projection failed");
                }
                if let Err(e) = self.index.put_entry(&nk, &nv) {
                    tracing::error!(error = %e, "index hook: put projection failed");
                }
            }
            (None, Some((nk, nv))) => {
                if let Err(e) = self.index.put_entry(&nk, &nv) {
                    tracing::error!(error = %e, "index hook: put projection failed");
                }
            }
            (Some((ok, _)), None) => {
                if let Err(e) = self.index.delete_entry(&ok) {
                    tracing::error!(error = %e, "index hook: delete projection failed");
                }
            }
            (None, None) => {}
        }
    }
}