solistore 0.2.1

Embedded LSM key-value store for SoliDB: pure Rust, durable on every commit by default
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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

//! `Db::ingest`: a whole column family loaded as bottom-level tables.

use solistore::{Db, Error, IteratorMode, Options, WriteBatch};

fn key(i: u32) -> Vec<u8> {
    format!("doc:{i:08}").into_bytes()
}

fn value(i: u32) -> Vec<u8> {
    format!("value {i} ")
        .repeat(1 + (i % 7) as usize)
        .into_bytes()
}

fn entries(n: u32) -> impl Iterator<Item = Result<(Vec<u8>, Vec<u8>), String>> {
    (0..n).map(|i| Ok((key(i), value(i))))
}

fn small() -> Options {
    Options {
        write_buffer_size: 64 << 10,
        target_file_size: 256 << 10,
        ..Options::default()
    }
}

/// Everything ingested reads back, by key and in order, before and after a
/// reopen, and nothing went through the WAL or a flush.
#[test]
fn an_ingested_column_family_reads_back_whole() {
    let dir = tempfile::tempdir().unwrap();
    let n = 50_000;
    {
        let db = Db::open(dir.path(), small()).unwrap();
        let cf = db.create_cf("bulk").unwrap();
        let report = db.ingest(&cf, entries(n), 4).unwrap();
        assert_eq!(report.entries, n as u64);
        assert!(report.files > 1, "{report:?}");
        assert_eq!(db.stats().flushes, 0, "no flush");
        assert_eq!(db.memtable_bytes(), 0, "no memtable");
        for i in (0..n).step_by(997) {
            assert_eq!(db.get(&cf, key(i)).unwrap().as_deref(), Some(&value(i)[..]));
        }
        assert_eq!(
            db.iterator(&cf, IteratorMode::Start).unwrap().count(),
            n as usize
        );
    }
    let db = Db::open(dir.path(), small()).unwrap();
    let cf = db.cf_handle("bulk").unwrap();
    let mut expected = 0u32;
    for item in db.iterator(&cf, IteratorMode::Start).unwrap() {
        let (k, v) = item.unwrap();
        assert_eq!(&k[..], &key(expected)[..]);
        assert_eq!(&v[..], &value(expected)[..]);
        expected += 1;
    }
    assert_eq!(expected, n);
}

/// Writes after an ingest win over the ingested entries (sequence 0), and
/// deletes hide them, through flushes and compactions.
#[test]
fn later_writes_and_deletes_win_over_ingested_entries() {
    let dir = tempfile::tempdir().unwrap();
    let db = Db::open(dir.path(), small()).unwrap();
    let cf = db.create_cf("bulk").unwrap();
    db.ingest(&cf, entries(10_000), 2).unwrap();
    for i in (0..10_000).step_by(3) {
        db.put(&cf, key(i), b"new").unwrap();
    }
    for i in (1..10_000).step_by(3) {
        db.delete(&cf, key(i)).unwrap();
    }
    let mut batch = WriteBatch::new();
    batch.delete_range(&cf, key(9_000), key(9_500));
    db.write(batch).unwrap();
    let check = |db: &Db, when: &str| {
        let cf = db.cf_handle("bulk").unwrap();
        for i in 0..10_000u32 {
            let got = db.get(&cf, key(i)).unwrap();
            let want: Option<Vec<u8>> = if (9_000..9_500).contains(&i) || i % 3 == 1 {
                None
            } else if i % 3 == 0 {
                Some(b"new".to_vec())
            } else {
                Some(value(i))
            };
            assert_eq!(got.map(|v| v.to_vec()), want, "{when}: key {i}");
        }
    };
    check(&db, "before compaction");
    db.compact_range(&cf, None, None).unwrap();
    check(&db, "after compaction");
    drop(db);
    let db = Db::open(dir.path(), small()).unwrap();
    check(&db, "after reopening");
}

/// An ingest into a column family that holds anything, or with keys out of
/// order, is refused and leaves nothing behind.
#[test]
fn an_ingest_is_refused_unless_the_family_is_untouched_and_keys_ascend() {
    let dir = tempfile::tempdir().unwrap();
    let db = Db::open(dir.path(), small()).unwrap();

    let used = db.create_cf("used").unwrap();
    db.put(&used, b"x", b"1").unwrap();
    assert!(matches!(
        db.ingest(&used, entries(10), 1),
        Err(Error::InvalidArgument(_))
    ));

    // A delete is enough: it would shadow entries at sequence 0.
    let deleted = db.create_cf("deleted").unwrap();
    db.delete(&deleted, b"x").unwrap();
    assert!(db.ingest(&deleted, entries(10), 1).is_err());

    let unordered = db.create_cf("unordered").unwrap();
    let bad = vec![
        Ok::<_, String>((b"b".to_vec(), b"1".to_vec())),
        Ok((b"a".to_vec(), b"2".to_vec())),
    ];
    assert!(db.ingest(&unordered, bad, 1).is_err());
    assert_eq!(
        db.iterator(&unordered, IteratorMode::Start)
            .unwrap()
            .count(),
        0
    );

    let failing = db.create_cf("failing").unwrap();
    let source = (0..10u32).map(|i| {
        if i == 5 {
            Err("source broke".to_string())
        } else {
            Ok((key(i), value(i)))
        }
    });
    let err = db.ingest(&failing, source, 2).unwrap_err().to_string();
    assert!(err.contains("source broke"), "{err}");
    assert_eq!(
        db.iterator(&failing, IteratorMode::Start).unwrap().count(),
        0
    );
}