cobble 0.1.0

A flexible embedded key-value storage engine for distributed systems as well as single-node applications.
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

//! An iterator over memtable entries, sorted by key and ordinal.
//! This is used for compaction and flushing to ensure that entries are processed in the correct order.
use crate::error::Result;
use crate::iterator::KvIterator;
use crate::r#type::KvValue;
use crate::util::unsafe_bytes;
use bytes::Bytes;
use std::cmp::Ordering;

#[derive(PartialEq, Eq)]
struct OrderedKvEntry<'a> {
    key: &'a [u8],
    value: &'a [u8],
    ordinal: usize,
}

impl Ord for OrderedKvEntry<'_> {
    fn cmp(&self, other: &Self) -> Ordering {
        let ord = self.key.cmp(other.key);
        if ord == Ordering::Equal {
            other.ordinal.cmp(&self.ordinal)
        } else {
            ord
        }
    }
}

impl PartialOrd for OrderedKvEntry<'_> {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

pub(crate) struct OrderedMemtableKvIterator<'a> {
    entries: Vec<OrderedKvEntry<'a>>,
    idx: usize,
    current_key: Option<&'a [u8]>,
    current_value: Option<&'a [u8]>,
}

impl<'a> OrderedMemtableKvIterator<'a> {
    pub(crate) fn new(entries: Vec<(&'a [u8], &'a [u8], usize)>) -> Self {
        let mut entries: Vec<OrderedKvEntry<'a>> = entries
            .into_iter()
            .map(|(key, value, ordinal)| OrderedKvEntry {
                key,
                value,
                ordinal,
            })
            .collect();
        entries.sort();
        Self {
            entries,
            idx: 0,
            current_key: None,
            current_value: None,
        }
    }
}

impl<'a> KvIterator<'a> for OrderedMemtableKvIterator<'a> {
    fn seek(&mut self, target: &[u8]) -> Result<()> {
        let idx = match self.entries.binary_search_by(|entry| entry.key.cmp(target)) {
            Ok(mut idx) => {
                // Walk backwards to the first entry with this key,
                // since binary_search may land on any matching entry.
                while idx > 0 && self.entries[idx - 1].key == target {
                    idx -= 1;
                }
                idx
            }
            Err(idx) => idx,
        };
        self.idx = idx;
        self.current_key = None;
        self.current_value = None;
        Ok(())
    }

    fn seek_to_first(&mut self) -> Result<()> {
        self.idx = 0;
        self.current_key = None;
        self.current_value = None;
        Ok(())
    }

    fn next(&mut self) -> Result<bool> {
        if self.idx >= self.entries.len() {
            self.current_key = None;
            self.current_value = None;
            return Ok(false);
        }
        let entry = &self.entries[self.idx];
        self.idx += 1;
        self.current_key = Some(entry.key);
        self.current_value = Some(entry.value);
        Ok(true)
    }

    fn valid(&self) -> bool {
        self.current_key.is_some() && self.current_value.is_some()
    }

    fn key(&self) -> Result<Option<&[u8]>> {
        Ok(self.current_key)
    }

    fn take_key(&mut self) -> Result<Option<Bytes>> {
        // Safety: In flush, we keep the memtable alive until the flush completes, so the bytes we
        // return here will remain valid for the caller to use. In lookup or scan, the unsafe bytes
        // will be converted to decoded key before returning, so it's also safe.
        Ok(self.current_key.map(unsafe_bytes))
    }

    fn take_value(&mut self) -> Result<Option<KvValue>> {
        // Safety: In flush, we keep the memtable alive until the flush completes, so the bytes we
        // return here will remain valid for the caller to use. In lookup or scan, the unsafe bytes
        // will be converted to decoded value before returning, so it's also safe.
        Ok(self
            .current_value
            .map(|v| KvValue::Encoded(unsafe_bytes(v))))
    }
}