tessarikv 0.1.0

A pure-Rust, lock-free-first LSM 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
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

//! Every keyspace compacts on its own, by its own options, and a range request compacts the
//! keyspace it names.

use std::sync::Arc;

use tessarikv_constants::LEVELS;

use super::Db;
use super::flush_tests::{PATH, key, until, value};
use crate::log::fault::FaultStorage;
use crate::{DbOptions, KeyspaceId, KeyspaceOptions, WriteBatch, WriteOptions};

/// A database whose default keyspace compacts level 0 at `default_trigger` tables and whose
/// users keyspace at `users_trigger`.
async fn open(default_trigger: u32, users_trigger: u32) -> (Db, KeyspaceId) {
    let trigger = |tables| {
        KeyspaceOptions::builder()
            .level0_compaction_trigger(tables)
            .build()
            .expect("valid")
    };
    let options = DbOptions::builder()
        .default_keyspace(trigger(default_trigger))
        .keyspace("users", trigger(users_trigger))
        .create_missing_keyspaces(true)
        .build()
        .expect("valid");
    let db = Db::open_with_storage(PATH, options, Arc::new(FaultStorage::default()))
        .await
        .expect("opened");
    let users = db.keyspace("users").expect("users");
    (db, users)
}

/// Writes key `index` into both keyspaces and flushes: one level-0 table each.
async fn flush_both(db: &Db, users: KeyspaceId, index: u32) {
    let mut batch = WriteBatch::new();
    for keyspace in [KeyspaceId::DEFAULT, users] {
        batch
            .put(keyspace, &key(index), &value(index))
            .expect("entry");
    }
    db.write(&batch, &WriteOptions::default())
        .await
        .expect("written");
    db.flush().await.expect("flushed");
}

/// The levels of `keyspace`'s tables.
fn levels(db: &Db, keyspace: KeyspaceId) -> Vec<u8> {
    db.view
        .load()
        .files(keyspace.get())
        .map(|file| file.meta.level)
        .collect()
}

/// A keyspace whose level 0 reaches its trigger compacts, while one under its own trigger
/// keeps its tables at level 0.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn each_keyspace_compacts_by_its_own_trigger() {
    let (db, users) = open(8, 2).await;
    for index in 0..4 {
        flush_both(&db, users, index).await;
    }
    until("a users table below level 0", || {
        levels(&db, users).iter().any(|level| *level > 0)
    })
    .await;
    assert_eq!(levels(&db, KeyspaceId::DEFAULT), [0, 0, 0, 0]);
}

/// A range request compacts the keyspace it names down to the bottom level, and no other.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_range_request_compacts_only_the_keyspace_it_names() {
    let (db, users) = open(8, 8).await;
    for index in 0..2 {
        flush_both(&db, users, index).await;
    }
    db.compact_range(users, b"", None).await.expect("compacted");
    let bottom = LEVELS
        .checked_sub(1)
        .and_then(|bottom| u8::try_from(bottom).ok())
        .expect("a bottom level");
    let compacted = levels(&db, users);
    assert!(!compacted.is_empty(), "the users keyspace keeps its keys");
    assert!(
        compacted.iter().all(|level| *level == bottom),
        "{compacted:?}"
    );
    assert_eq!(levels(&db, KeyspaceId::DEFAULT), [0, 0]);
}