sakuhiki-core 0.0.1-alpha.0

Indexing overlay for any key-value store
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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110

use futures_util::StreamExt as _;

use crate::{
    Backend, CfError, Datum, IndexError,
    backend::{BackendCf as _, Transaction as _},
};

pub trait Indexer<B: Backend>: Send + Sync {
    type Datum: Datum;

    fn cfs(&self) -> &'static [&'static str];

    fn index<'fut, 't>(
        &'fut self,
        object_key: &'fut [u8],
        datum: &'fut Self::Datum,
        transaction: &'fut B::Transaction<'t>,
        cfs: &'fut [B::TransactionCf<'t>],
    ) -> waaa::BoxFuture<'fut, Result<(), CfError<B::Error>>>;

    fn unindex<'fut, 't>(
        &'fut self,
        object_key: &'fut [u8],
        datum: &'fut Self::Datum,
        transaction: &'fut B::Transaction<'t>,
        cfs: &'fut [B::TransactionCf<'t>],
    ) -> waaa::BoxFuture<'fut, Result<(), CfError<B::Error>>>;

    #[allow(clippy::type_complexity)] // No good way to cut this smaller
    fn index_from_slice<'fut, 't>(
        &'fut self,
        object_key: &'fut [u8],
        slice: &'fut [u8],
        transaction: &'fut B::Transaction<'t>,
        cfs: &'fut [B::TransactionCf<'t>],
    ) -> waaa::BoxFuture<'fut, Result<(), IndexError<B::Error, <Self::Datum as Datum>::Error>>>
    {
        Box::pin(async move {
            let datum = Self::Datum::from_slice(slice).map_err(IndexError::Parsing)?;
            self.index(object_key, &datum, transaction, cfs)
                .await
                .map_err(IndexError::Backend)
        })
    }

    #[allow(clippy::type_complexity)] // No good way to cut this smaller
    fn unindex_from_slice<'fut, 't>(
        &'fut self,
        object_key: &'fut [u8],
        slice: &'fut [u8],
        transaction: &'fut B::Transaction<'t>,
        cfs: &'fut [B::TransactionCf<'t>],
    ) -> waaa::BoxFuture<'fut, Result<(), IndexError<B::Error, <Self::Datum as Datum>::Error>>>
    {
        Box::pin(async move {
            let datum = Self::Datum::from_slice(slice).map_err(IndexError::Parsing)?;
            self.unindex(object_key, &datum, transaction, cfs)
                .await
                .map_err(IndexError::Backend)
        })
    }

    #[allow(clippy::type_complexity)]
    fn rebuild<'fut, 't>(
        &'fut self,
        transaction: &'fut B::Transaction<'t>,
        index_cfs: &'fut [B::TransactionCf<'t>],
        datum_cf: &'fut B::TransactionCf<'t>,
    ) -> waaa::BoxFuture<'fut, Result<(), IndexError<B::Error, <Self::Datum as Datum>::Error>>>
    {
        Box::pin(async move { default_rebuild(self, transaction, index_cfs, datum_cf).await })
    }
}

pub async fn default_rebuild<'fut, 't, B, I>(
    this: &'fut I,
    transaction: &'fut B::Transaction<'t>,
    index_cfs: &'fut [B::TransactionCf<'t>],
    datum_cf: &'fut B::TransactionCf<'t>,
) -> Result<(), IndexError<B::Error, <I::Datum as Datum>::Error>>
where
    B: Backend,
    I: ?Sized + Indexer<B>,
{
    // Dark magic to work around https://github.com/rust-lang/rust/issues/100013
    // See https://github.com/rust-lang/rust/issues/100013#issuecomment-2210995259 for the inspiration
    // This should be removed someday when rustc gets fixed
    fn assert_send<T: Send>(v: T) -> impl Send {
        v
    }
    let _lock = assert_send(
        transaction
            .take_exclusive_lock(datum_cf)
            .await
            .map_err(|e| IndexError::cf(datum_cf.name(), e))?,
    );
    for cf in index_cfs {
        transaction
            .clear(cf)
            .await
            .map_err(|e| IndexError::cf(cf.name(), e))?;
    }
    let mut all_data = transaction.scan::<[u8]>(datum_cf, ..);
    while let Some(d) = all_data.next().await {
        let (key, datum) = d.map_err(|e| IndexError::cf(datum_cf.name(), e))?;
        this.index_from_slice(key.as_ref(), datum.as_ref(), transaction, index_cfs)
            .await?;
    }
    Ok(())
}