bitcasky 0.1.2

Bitcasky is a Rust implementation of the Bitcask key-value store
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

use std::time::{Duration, Instant};

use dashmap::{
    iter::{Iter, OwningIter},
    mapref::{multiple::RefMulti, one::Ref},
    DashMap,
};

use crate::database::{Database, RowLocation};
use crate::error::BitcaskyResult;

#[derive(Debug)]
pub struct KeyDirTelemetry {
    pub number_of_keys: usize,
    pub recovery_duration: Duration,
}

#[derive(Clone, Debug)]
pub struct KeyDir {
    index: DashMap<Vec<u8>, RowLocation>,
    recovery_duration: Duration,
}

impl KeyDir {
    pub fn new_empty_key_dir() -> KeyDir {
        let index = DashMap::new();
        KeyDir {
            index,
            recovery_duration: Duration::ZERO,
        }
    }

    pub fn new(database: &Database) -> BitcaskyResult<KeyDir> {
        let index = DashMap::new();
        let start = Instant::now();
        for ret in database.recovery_iter()? {
            let item = ret?;
            if item.invalid {
                index.remove(&item.key);
                continue;
            }

            index.insert(item.key, item.row_location);
        }
        Ok(KeyDir {
            index,
            recovery_duration: start.elapsed(),
        })
    }

    pub fn put(&self, key: Vec<u8>, value: RowLocation) -> Option<RowLocation> {
        self.index.insert(key, value)
    }

    pub fn checked_put(&self, key: Vec<u8>, value: RowLocation) -> Option<RowLocation> {
        let r = self.index.get(&key);
        if let Some(pos) = r {
            let old_pos: RowLocation = *(pos);
            // key was written again during merge
            if value.storage_id < old_pos.storage_id {
                return Option::None;
            }
        }
        self.index.insert(key, value)
    }

    pub fn get(&self, key: &Vec<u8>) -> Option<Ref<Vec<u8>, RowLocation>> {
        self.index.get(key)
    }

    pub fn contains_key(&self, key: &Vec<u8>) -> bool {
        self.index.contains_key(key)
    }

    pub fn len(&self) -> usize {
        self.index.len()
    }

    pub fn iter(&self) -> KeyDirIterator {
        KeyDirIterator {
            iter: self.index.iter(),
        }
    }

    pub fn into_iter(self) -> IntoKeyDirIterator {
        IntoKeyDirIterator {
            iter: self.index.into_iter(),
        }
    }

    pub fn delete(&self, key: &Vec<u8>) -> Option<(Vec<u8>, RowLocation)> {
        self.index.remove(key)
    }

    pub fn clear(&self) {
        self.index.clear();
    }

    pub fn get_telemetry_data(&self) -> KeyDirTelemetry {
        KeyDirTelemetry {
            number_of_keys: self.len(),
            recovery_duration: self.recovery_duration,
        }
    }
}

pub struct KeyDirIterator<'a> {
    iter: Iter<'a, Vec<u8>, RowLocation>,
}

impl<'a> Iterator for KeyDirIterator<'a> {
    type Item = RefMulti<'a, Vec<u8>, RowLocation>;

    fn next(&mut self) -> Option<Self::Item> {
        self.iter.next()
    }
}

pub struct IntoKeyDirIterator {
    iter: OwningIter<Vec<u8>, RowLocation>,
}

impl Iterator for IntoKeyDirIterator {
    type Item = (Vec<u8>, RowLocation);

    fn next(&mut self) -> Option<Self::Item> {
        self.iter.next()
    }
}