trine-kv 0.5.13

Embedded LSM MVCC key-value database.
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

use std::{collections::BinaryHeap, sync::Arc};

use super::{Direction, Iter, RecordSource, ScanSelector, ScanSourceInput, SourceHeapEntry};
use crate::{
    blob::ValueRef,
    internal_key::{InternalKey, ValueKind},
    memtable::Memtable,
    snapshot::Snapshot,
    types::{KeyRange, Sequence},
};

#[test]
fn source_heap_orders_forward_and_reverse_keys() {
    let mut forward = BinaryHeap::new();
    forward.push(heap_entry(b"c", 0, Direction::Forward));
    forward.push(heap_entry(b"a", 1, Direction::Forward));
    forward.push(heap_entry(b"b", 2, Direction::Forward));

    assert_eq!(forward.pop().expect("entry").user_key, b"a");
    assert_eq!(forward.pop().expect("entry").user_key, b"b");
    assert_eq!(forward.pop().expect("entry").user_key, b"c");

    let mut reverse = BinaryHeap::new();
    reverse.push(heap_entry(b"c", 0, Direction::Reverse));
    reverse.push(heap_entry(b"a", 1, Direction::Reverse));
    reverse.push(heap_entry(b"b", 2, Direction::Reverse));

    assert_eq!(reverse.pop().expect("entry").user_key, b"c");
    assert_eq!(reverse.pop().expect("entry").user_key, b"b");
    assert_eq!(reverse.pop().expect("entry").user_key, b"a");
}

#[test]
fn lazy_scan_heap_merge_preserves_forward_and_reverse_order() {
    let left = memtable_with(&[(b"a", b"a1"), (b"c", b"c1")]);
    let right = memtable_with(&[(b"b", b"b1"), (b"d", b"d1")]);

    let forward = Iter::from_sources(
        Direction::Forward,
        ScanSourceInput {
            read_sequence: Sequence::new(4),
            read_pin: Snapshot::new(Sequence::new(4)),
            db_path: None,
            native_storage: None,
            blob_reads: None,
            scan_waste: None,
            range_tombstones: Vec::new(),
            sources: vec![
                RecordSource::memtable(
                    Arc::clone(&left),
                    ScanSelector::Range(KeyRange::all()),
                    Direction::Forward,
                ),
                RecordSource::memtable(
                    Arc::clone(&right),
                    ScanSelector::Range(KeyRange::all()),
                    Direction::Forward,
                ),
            ],
        },
    );
    assert_eq!(
        collect_keys(forward),
        vec![b"a".to_vec(), b"b".to_vec(), b"c".to_vec(), b"d".to_vec()]
    );

    let reverse = Iter::from_sources(
        Direction::Reverse,
        ScanSourceInput {
            read_sequence: Sequence::new(4),
            read_pin: Snapshot::new(Sequence::new(4)),
            db_path: None,
            native_storage: None,
            blob_reads: None,
            scan_waste: None,
            range_tombstones: Vec::new(),
            sources: vec![
                RecordSource::memtable(
                    left,
                    ScanSelector::Range(KeyRange::all()),
                    Direction::Reverse,
                ),
                RecordSource::memtable(
                    right,
                    ScanSelector::Range(KeyRange::all()),
                    Direction::Reverse,
                ),
            ],
        },
    );
    assert_eq!(
        collect_keys(reverse),
        vec![b"d".to_vec(), b"c".to_vec(), b"b".to_vec(), b"a".to_vec()]
    );
}

fn heap_entry(user_key: &[u8], source_index: usize, direction: Direction) -> SourceHeapEntry {
    SourceHeapEntry {
        user_key: user_key.to_vec(),
        source_index,
        direction,
    }
}

fn memtable_with(records: &[(&[u8], &[u8])]) -> Arc<Memtable> {
    let memtable = Arc::new(Memtable::default());
    {
        let mut entries = memtable.write_entries().expect("memtable lock");
        for (index, (key, value)) in records.iter().enumerate() {
            entries.insert(
                InternalKey::new(
                    *key,
                    Sequence::new(u64::try_from(index + 1).expect("test sequence fits")),
                    ValueKind::Put,
                    0,
                ),
                Some(ValueRef::Inline((*value).to_vec())),
            );
        }
    }
    memtable
}

fn collect_keys(iter: Iter) -> Vec<Vec<u8>> {
    iter.map(|item| item.expect("iterator item").key)
        .collect::<Vec<_>>()
}