bplus_store 0.4.0

Copy-on-write B+ tree with page-aligned storage, split/merge, and crash-safety primitives.
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
152
153
154
155
156
157

#![cfg(test)]

use crate::bplustree::transaction::WriteTransaction;
use crate::tests::common;
use rand::seq::SliceRandom;
use rand::thread_rng;
use std::thread;
use tempfile::TempDir;

// ---------------------------------------------------------------------------
// Key/value encoding helpers
// ---------------------------------------------------------------------------

/// Encodes a `u64` as a big-endian byte array for use as a tree key.
fn k(i: u64) -> [u8; 8] {
    i.to_be_bytes()
}

/// Returns the value string for key `i`.
fn v_str(i: u64) -> String {
    format!("value_{}", i)
}

/// Returns the value for key `i` as raw bytes.
fn v_bytes(i: u64) -> Vec<u8> {
    v_str(i).into_bytes()
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[test]
fn commit_happy_path() {
    let dir = TempDir::new().unwrap();
    let order = 16;
    let tree = common::make_tree(&dir, order).expect("create tree");
    let mut trx = WriteTransaction::new(tree.clone());

    for i in 0..100u64 {
        trx.insert(k(i), v_bytes(i));
    }

    trx.commit(&tree).expect("commit");

    for i in 0..100u64 {
        assert_eq!(tree.search(k(i)).expect("get"), Some(v_bytes(i)));
    }
}

#[test]
fn commit_with_random_inserts() {
    let dir = TempDir::new().unwrap();
    let order = 16;
    let tree = common::make_tree(&dir, order).expect("create tree");
    let mut trx = WriteTransaction::new(tree.clone());

    let mut keys: Vec<u64> = (0..100).collect();
    keys.shuffle(&mut thread_rng());

    for &key in &keys {
        trx.insert(k(key), v_bytes(key));
    }

    trx.commit(&tree).expect("commit");

    for &key in &keys {
        assert_eq!(tree.search(k(key)).expect("get"), Some(v_bytes(key)));
    }
}

#[test]
fn contending_parallel_transactions() {
    let dir = TempDir::new().unwrap();
    let order = 16;
    let tree = common::make_tree(&dir, order).expect("create tree");
    thread::scope(|s| {
        for i in 0..10u64 {
            let t = tree.clone();
            s.spawn(move || {
                let mut trx = WriteTransaction::new(t.clone());
                for j in 0..100u64 {
                    trx.insert(k(i * 100 + j), v_bytes(i * 100 + j));
                }
                trx.commit(&t).expect("commit");
            });
        }
    });
    for i in 0..1000u64 {
        assert_eq!(tree.search(k(i)).expect("get"), Some(v_bytes(i)));
    }
}

#[test]
fn commit_with_conflicting_transactions() {
    let dir = TempDir::new().unwrap();
    let order = 16;
    let tree = common::make_tree(&dir, order).expect("create tree");

    let mut t1 = WriteTransaction::new(tree.clone());
    let mut t2 = WriteTransaction::new(tree.clone());

    t1.insert(k(42), b"value_42_t1");
    t2.insert(k(42), b"value_42_t2");

    t1.commit(&tree).expect("commit t1");
    // t2 will rebase on top of t1's commit and win.
    t2.commit(&tree).expect("commit t2");

    tree.search(k(42)).expect("get").map_or_else(
        || panic!("Key 42 should exist after both commits"),
        |value| assert_eq!(value, b"value_42_t2", "Last writer (t2) should win"),
    );
}

#[test]
fn commit_failure_should_reclaim_nodes() {
    let dir = TempDir::new().unwrap();
    let order = 16;
    let tree = common::make_tree(&dir, order).expect("create tree");

    let mut trx = WriteTransaction::new(tree.clone());
    for i in 0..10u64 {
        trx.insert(k(i), v_bytes(i));
    }
    trx.commit(&tree).expect("commit");

    // Overwrite all values in a second transaction.
    let mut trx2 = WriteTransaction::new(tree.clone());
    for i in 0..10u64 {
        trx2.insert(k(i), v_bytes(i * 2));
    }
    trx2.commit(&tree).expect("commit overwrite");

    let deferred = tree.epoch_mgr().get_deferred_pages();
    assert!(
        deferred.is_empty(),
        "Deferred pages should be empty after successful commit"
    );
}

#[test]
fn noop_tx_commit_no_side_effects() {
    let dir = TempDir::new().unwrap();
    let order = 16;
    let tree = common::make_tree(&dir, order).expect("create tree");
    let root_before = tree.get_root_id();
    let mut trx = WriteTransaction::new(tree.clone());

    trx.commit(&tree).expect("commit with no operations");

    assert_eq!(
        tree.get_root_id(),
        root_before,
        "Root should be unchanged after a no-op commit"
    );
}