fjall 3.1.4

Log-structured, embeddable key-value storage engine
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

// Copyright (c) 2024-present, fjall-rs
// This source code is licensed under both the Apache 2.0 and MIT License
// (found in the LICENSE-* files in the repository)

use crate::{worker_pool::WorkerMessage, Keyspace};
use lsm_tree::{AnyIngestion, UserKey, UserValue};

pub struct Ingestion<'a> {
    keyspace: &'a Keyspace,
    inner: AnyIngestion<'a>,
}

impl<'a> Ingestion<'a> {
    pub fn new(keyspace: &'a Keyspace) -> crate::Result<Self> {
        let inner = keyspace.tree.ingestion()?;
        Ok(Self { keyspace, inner })
    }

    pub fn write<K: Into<UserKey>, V: Into<UserValue>>(
        &mut self,
        key: K,
        value: V,
    ) -> crate::Result<()> {
        self.inner.write(key, value).map_err(Into::into)
    }

    pub fn write_tombstone<K: Into<UserKey>>(&mut self, key: K) -> crate::Result<()> {
        self.inner.write_tombstone(key).map_err(Into::into)
    }

    #[doc(hidden)]
    pub fn write_weak_tombstone<K: Into<UserKey>>(&mut self, key: K) -> crate::Result<()> {
        self.inner.write_weak_tombstone(key).map_err(Into::into)
    }

    pub fn finish(self) -> crate::Result<()> {
        // NOTE: We hold to avoid a race condition with concurrent writes:
        //
        // write         ingest
        // lock journal
        // |
        // next seqno=1
        // |
        // --------------finish
        //                 flush
        //                 seqno=2
        //                 register
        //                 |
        // -----------------
        // |
        // insert seqno=1
        let _journal_lock = self.keyspace.supervisor.journal.get_writer();

        self.inner
            .finish()
            .inspect(|()| {
                self.keyspace
                    .worker_messager
                    .try_send(WorkerMessage::Compact(self.keyspace.clone()))
                    .ok();

                self.keyspace.supervisor.snapshot_tracker.gc();
            })
            .map_err(Into::into)
    }
}