shunyadb 0.1.0

A correctness-first embedded storage engine written in Rust, built around WAL, immutable LSM pages, MVCC, and deterministic crash recovery.
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
111
112
113
114
115
116
117
118
119

use anyhow::Result;
use std::collections::BTreeMap;
use std::path::Path;

use crate::engine::seqno::allocate;
use crate::meta::PageMeta;
use crate::storage::memtable::MemTable;
use crate::storage::page::builder::{PageBuilder, Page};
use crate::storage::page::io::write_page;
use crate::storage::record::{FieldValue, Record};
use crate::storage::wal::{Wal, WalEntry, WalOp};

const MAX_RECORDS_PER_PAGE: usize = 1024;
const MAX_PER_PAGE_SIZE: usize = 32 * 1024; // 32 KB for L0

pub struct Writer;

impl Writer {
    pub fn new() -> Self {
        Self
    }

    pub fn put(
        &self,
        memtable: &mut MemTable,
        wal: &mut Wal,
        id: String,
        value: BTreeMap<String, FieldValue>,
    ) -> Result<()> {
        let seqno = allocate();
        let record = Record::new(id, seqno, value);
        let wal_entry = WalEntry::new(WalOp::Insert, String::new(), record.id.clone(), seqno, Some(record.clone()));
        wal.append(&wal_entry)?;
        memtable.put(record);

        Ok(())
    }

    pub fn delete(
        &self,
        memtable: &mut MemTable,
        wal: &mut Wal,
        id: String,
    ) -> Result<()> {
        let seqno = allocate();
        let record = Record::new_tombstone(id, seqno);
        let wal_entry = WalEntry::new(WalOp::Delete, String::new(), record.id.clone(), seqno, Some(record.clone()));
        wal.append(&wal_entry)?;
        memtable.put(record);

        Ok(())
    }

    pub fn flush(
        &self,
        memtable: &mut MemTable,
        dir: &Path,
        next_page_id: &u64,
    ) -> Result<(u64, Vec<PageMeta>)> {
        let mut next_page_id = next_page_id.clone();
        let mut builder = PageBuilder::new();
        let mut count = 0;
        let mut pagesmeta = Vec::new();

        for (_id, versions) in memtable.iter() {
            for record in versions {
                let estimated_size = builder.estimate_size_with(record);

                if estimated_size > MAX_PER_PAGE_SIZE {
                    let page = builder.build();
                    pagesmeta.push(self.flush_one(&page, dir, &next_page_id)?);
                    builder = PageBuilder::new();
                    count = 0;
                    next_page_id = next_page_id + 1;
                }

                builder.add(record.clone());
                builder.update_size(&record);
                count += 1;

                if count >= MAX_RECORDS_PER_PAGE {
                    let page = builder.build();
                    pagesmeta.push(self.flush_one(&page, dir, &next_page_id)?);
                    builder = PageBuilder::new();
                    count = 0;
                    next_page_id = next_page_id + 1;
                }
            }
        }

        if count > 0 {
            let page = builder.build();
            pagesmeta.push(self.flush_one(&page, dir, &mut &next_page_id)?);
            next_page_id = next_page_id + 1;
        }

        memtable.clear();
        Ok((next_page_id, pagesmeta))
    }

    fn flush_one(
        &self,
        page: &Page,
        dir: &Path,
        page_id: &u64
    ) -> Result<PageMeta> {
        let path = dir.join(format!("page_{}.db", page_id));
        let page_size = write_page(&path, &page)?;

        Ok(PageMeta::new(
            *page_id,
            page.header.min_id.clone(),
            page.header.max_id.clone(),
            page.header.num_records as usize,
            page_size,
            page.header.page_seqno,
        ))
    }
}