lsm 0.4.1

An implementation of log-structured merge trees in pure Rust
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
147
148
149
150
151

use lsm::{Database, Params, StartMode, WriteOptions};
use tempfile::{Builder, TempDir};

use futures::stream::StreamExt;

#[cfg(feature = "async-io")]
use tokio_uring_executor::test as async_test;

#[cfg(not(feature = "async-io"))]
use tokio::test as async_test;

async fn test_init() -> (TempDir, Params, Database) {
    let tmp_dir = Builder::new()
        .prefix("lsm-async-test-reopen-")
        .tempdir()
        .unwrap();
    let _ = env_logger::builder().is_test(true).try_init();

    let mut db_path = tmp_dir.path().to_path_buf();
    db_path.push("storage.lsm");

    let params = Params {
        db_path,
        ..Default::default()
    };
    let database = Database::new_with_params(StartMode::CreateOrOverride, params.clone())
        .await
        .expect("Failed to create database instance");

    (tmp_dir, params, database)
}

#[async_test]
async fn get_put() {
    let (_tmpdir, params, database) = test_init().await;

    let key1 = String::from("Foo").into_bytes();
    let value1 = String::from("Bar").into_bytes();
    let value2 = String::from("Baz").into_bytes();

    assert!(database.get(&key1).await.unwrap().is_none());

    database.put(key1.clone(), value1.clone()).await.unwrap();
    drop(database);

    // Reopen
    let database = Database::new_with_params(StartMode::Open, params.clone())
        .await
        .expect("Failed to create database instance");

    assert_eq!(
        database.get(&key1).await.unwrap().unwrap().get_value(),
        value1
    );
    database.put(key1.clone(), value2.clone()).await.unwrap();

    drop(database);

    // Reopen again
    let database = Database::new_with_params(StartMode::Open, params)
        .await
        .expect("Failed to create database instance");

    assert_eq!(
        database.get(&key1).await.unwrap().unwrap().get_value(),
        value2
    );
}

#[async_test]
async fn get_put_many() {
    const COUNT: u64 = 100_000;

    let (_tmpdir, params, database) = test_init().await;

    // Write without fsync to speed up tests
    let options = WriteOptions { sync: false };

    for pos in 0..COUNT {
        let key = format!("key_{pos:05}").into_bytes();
        let value = format!("some_string_{pos}").into_bytes();
        database.put_opts(key, value, &options).await.unwrap();
    }

    database.synchronize().await.unwrap();
    drop(database);

    // Reopen
    let database = Database::new_with_params(StartMode::Open, params.clone())
        .await
        .expect("Failed to create database instance");

    for pos in 0..COUNT {
        let key = format!("key_{pos:05}").into_bytes();
        let value = format!("some_string_{pos}").into_bytes();

        assert_eq!(
            database.get(&key).await.unwrap().unwrap().get_value(),
            value,
        );
    }

    // Ensure iteration still works
    let mut iterator = database.iter().await;
    let mut pos = 0;
    while let Some((key, value)) = iterator.next().await {
        assert_eq!(format!("key_{pos:05}").into_bytes(), key);
        assert_eq!(format!("some_string_{pos}").into_bytes(), value.get_value());
        pos += 1;
    }
}

#[async_test]
async fn get_put_large() {
    const COUNT: usize = 100;
    const SIZE: usize = 100_000;

    let (_tmpdir, params, database) = test_init().await;

    // Write without fsync to speed up tests
    let options = WriteOptions { sync: false };

    for pos in 0..COUNT {
        let key = format!("key_{pos:05}").into_bytes();
        let value = format!("value_{pos}").repeat(SIZE).into_bytes();

        database.put_opts(key, value, &options).await.unwrap();
    }

    database.synchronize().await.unwrap();
    drop(database);

    // Reopen
    let database = Database::new_with_params(StartMode::Open, params.clone())
        .await
        .expect("Failed to create database instance");

    let mut iterator = database.iter().await;
    let mut pos = 0;

    while let Some((key, value)) = iterator.next().await {
        let expected_key = format!("key_{pos:05}").into_bytes();
        let expected_value = format!("value_{pos}").repeat(SIZE).into_bytes();

        assert_eq!(expected_key, key);
        // Value is very long, so don't print
        assert!(expected_value == value.get_value());

        pos += 1;
    }
}