rdms 0.0.1

Composable storage for key,value index
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

use rand::prelude::random;
use rand::{rngs::SmallRng, Rng, SeedableRng};

use super::*;

use crate::core::{Reader, Writer};
use crate::error::Error;
use crate::llrb::Llrb;

#[test]
fn test_skip_scan() {
    let seed: u128 = random();

    let (n_ops, key_max) = (6_000, 2_000);
    let mut llrb: Box<Llrb<i64, i64>> = Llrb::new_lsm("test-llrb");
    random_llrb(n_ops, key_max, seed, &mut llrb);

    let testcases = vec![
        (Bound::Included(0), Bound::Included(5100)),
        (Bound::Included(6000), Bound::Excluded(5100)),
        (Bound::Included(4000), Bound::Unbounded),
        (Bound::Excluded(5000), Bound::Included(5100)),
        (Bound::Excluded(5000), Bound::Excluded(5100)),
        (Bound::Excluded(5000), Bound::Unbounded),
        (Bound::Unbounded, Bound::Excluded(5100)),
        (Bound::Unbounded, Bound::Included(5100)),
        (Bound::Unbounded, Bound::Unbounded),
    ];
    for within in testcases {
        let es: Vec<Entry<i64, i64>> = SkipScan::new(&*llrb, within)
            .map(|e| e.unwrap()) // unwrap
            .collect();
        for e in es {
            assert!(within.contains(&e.to_seqno()));
            for d in e.as_deltas() {
                assert!(within.contains(&d.to_seqno()));
            }
        }
    }

    let within = (Bound::Included(5000), Bound::Included(5000));
    let es: Vec<Entry<i64, i64>> = SkipScan::new(&*llrb, within)
        .map(|e| e.unwrap()) // unwrap
        .collect();
    assert_eq!(es.len(), 1);
    assert_eq!(es[0].to_seqno(), 5000);

    let within = (Bound::Included(5000), Bound::Excluded(5000));
    let es: Vec<Entry<i64, i64>> = SkipScan::new(&*llrb, within)
        .map(|e| e.unwrap()) // unwrap
        .collect();
    assert_eq!(es.len(), 0);

    let within = (Bound::Excluded(5000), Bound::Included(5000));
    let es: Vec<Entry<i64, i64>> = SkipScan::new(&*llrb, within)
        .map(|e| e.unwrap()) // unwrap
        .collect();
    assert_eq!(es.len(), 0);

    let within = (Bound::Excluded(5000), Bound::Excluded(5000));
    let es: Vec<Entry<i64, i64>> = SkipScan::new(&*llrb, within)
        .map(|e| e.unwrap()) // unwrap
        .collect();
    assert_eq!(es.len(), 0);

    let within = (Bound::Included(5000), Bound::Excluded(5001));
    let es: Vec<Entry<i64, i64>> = SkipScan::new(&*llrb, within)
        .map(|e| e.unwrap()) // unwrap
        .collect();
    assert_eq!(es.len(), 1);
    assert_eq!(es[0].to_seqno(), 5000);

    let within = (Bound::Excluded(5000), Bound::Included(5001));
    let es: Vec<Entry<i64, i64>> = SkipScan::new(&*llrb, within)
        .map(|e| e.unwrap()) // unwrap
        .collect();
    assert_eq!(es.len(), 1);
    assert_eq!(es[0].to_seqno(), 5001);

    let within = (Bound::Excluded(5000), Bound::Excluded(5001));
    let es: Vec<Entry<i64, i64>> = SkipScan::new(&*llrb, within)
        .map(|e| e.unwrap()) // unwrap
        .collect();
    assert_eq!(es.len(), 0);
}

fn random_llrb(n_ops: i64, key_max: i64, seed: u128, llrb: &mut Llrb<i64, i64>) {
    let mut rng = SmallRng::from_seed(seed.to_le_bytes());
    for _i in 0..n_ops {
        let key = (rng.gen::<i64>() % key_max).abs();
        let op = rng.gen::<usize>() % 3;
        //println!("key {} {} {} {}", key, llrb.to_seqno(), op);
        match op {
            0 => {
                let value: i64 = rng.gen();
                llrb.set(key, value).unwrap();
            }
            1 => {
                let value: i64 = rng.gen();
                {
                    let cas = match llrb.get(&key) {
                        Err(Error::KeyNotFound) => 0,
                        Err(_err) => unreachable!(),
                        Ok(e) => e.to_seqno(),
                    };
                    llrb.set_cas(key, value, cas).unwrap();
                }
            }
            2 => {
                llrb.delete(&key).unwrap();
            }
            _ => unreachable!(),
        }
    }
}