dharmadb 0.1.1

Persistent, fault tolerant key-value store written in Rust.
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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216

use crate::errors::Errors;
use crate::options::DharmaOpts;
use crate::storage::block::{create_blocks, write_block_to_disk, Record, Value};
use crate::traits::{ResourceKey, ResourceValue};
use buffered_offset_reader::{BufOffsetReader, OffsetReadMut};
use log;
use serde::de::DeserializeOwned;
use std::fs::{create_dir_all, File};
use std::path::{Path, PathBuf};

/// Write the list of key value pairs, sorted by key to a series of SSTables on disk.
/// # Arguments
/// * _option_  - Configurations options specified as `DharmaOpts`
/// * _tuples_  - List of key value pairs sorted by key.
///
/// # Returns
/// A `Result` which is
/// - `Ok`: - Empty value
/// - `Err`: - Error type as specified by `Errors` module
pub fn write_sstable<K: ResourceKey, V: ResourceValue>(
    options: &DharmaOpts,
    tuples: &Vec<(K, V)>,
    table_number: usize,
) -> Result<PathBuf, Errors> {
    let values: Vec<Value<K, V>> = tuples
        .iter()
        .map(|tup| {
            return Value::new(tup.0.clone(), tup.1.clone());
        })
        .collect();
    // pack values into blocks
    let mut blocks = Vec::new();
    // pack the values into blocks of fixed size as specified by `options.block_size_in_bytes`
    create_blocks(options, &values, &mut blocks);
    // write this chunk to disk
    let path_str = format!("{0}/tables/{1}.db", options.path, table_number);
    let path = Path::new(&path_str);
    if path.parent().is_some() && !path.parent().unwrap().exists() {
        create_dir_all(path.parent().unwrap());
    }
    // create file for SSTable
    let file_result = File::create(&path);
    if file_result.is_ok() {
        let mut file = file_result.unwrap();
        // write all blocks to SSTable file
        for (block_counter, block) in blocks.iter().enumerate() {
            let write_result = write_block_to_disk(options, &mut file, &block);
            if write_result.is_err() {
                log::error!(
                    "Failed to write block from chunk {0} to disk",
                    block_counter
                );
                return Err(Errors::SSTABLE_CREATION_FAILED);
            }
        }
    } else {
        log::error!("Failed to create SSTable from chunk from values");
        return Err(Errors::SSTABLE_CREATION_FAILED);
    }
    Ok(PathBuf::from(path_str))
}

/// Write the list of key value pairs, sorted by key to a series of SSTables on disk.
/// # Arguments
/// * _option_  - Configurations options specified as `DharmaOpts`
/// * _tuples_  - List of key value pairs sorted by key.
///
/// # Returns
/// A `Result` which is
/// - `Ok`: - Empty value
/// - `Err`: - Error type as specified by `Errors` module
pub fn write_sstable_at_path<K: ResourceKey, V: ResourceValue>(
    options: &DharmaOpts,
    tuples: &Vec<(K, V)>,
    path: &PathBuf,
) -> Result<(), Errors> {
    let values: Vec<Value<K, V>> = tuples
        .iter()
        .map(|tup| {
            return Value::new(tup.0.clone(), tup.1.clone());
        })
        .collect();
    // pack values into blocks
    let mut blocks = Vec::new();
    // pack the values into blocks of fixed size as specified by `options.block_size_in_bytes`
    create_blocks(options, &values, &mut blocks);
    // create file for SSTable
    let file_result = File::create(path);
    if file_result.is_ok() {
        let mut file = file_result.unwrap();
        // write all blocks to SSTable file
        for (block_counter, block) in blocks.iter().enumerate() {
            let write_result = write_block_to_disk(options, &mut file, &block);
            if write_result.is_err() {
                log::error!(
                    "Failed to write block from chunk {0} to disk",
                    block_counter
                );
                return Err(Errors::SSTABLE_CREATION_FAILED);
            }
        }
    } else {
        log::error!("Failed to create SSTable from chunk from values");
        return Err(Errors::SSTABLE_CREATION_FAILED);
    }
    Ok(())
}

/// Read the SSTable at the specified path and return the data persisted in it
/// as a `Vec` of `Value<K, V>`.
/// TODO(@deprecated) - Use SSTableReader instead.
///
/// # Arguments
/// * _option_ - Configuration options specified as `DharmaOpts`
/// * _path_ - File System Path to SSTable
///
/// # Returns
/// A `Result` that is
///  - `Ok`: The list of `Value<K, V>` persisted to the SSTable
///  - `Err`: Error type as specified by `Errors` module
pub fn read_sstable<K: DeserializeOwned, V: DeserializeOwned>(
    options: &DharmaOpts,
    path: &Path,
) -> Result<Vec<Value<K, V>>, Errors> {
    let mut output: Vec<Value<K, V>> = Vec::new();
    let file_result = File::open(path);
    if file_result.is_ok() {
        let file = file_result.unwrap();
        let metadata = file.metadata().unwrap();
        let total_size_in_bytes = metadata.len();
        // read number of blocks from metadata embedded in SStable rather than relying
        // on options. options might change which might cause read error due to mismatch
        // between written and supplied block size
        let block_count =
            (total_size_in_bytes as f64 / options.block_size_in_bytes as f64).ceil() as u64;
        let mut i = 0;
        let mut reader = BufOffsetReader::new(file);
        // buffer to accumulate data from records split across multiple blocks
        let mut record_byte_buffer = Vec::new();
        while i < block_count {
            let mut buffer = vec![0u8; options.block_size_in_bytes as usize];
            // read blocksize number of bytes
            // TODO: handle read error
            reader.read_at(&mut buffer, i * options.block_size_in_bytes as u64);
            // unpack bytes array into records
            let mut r = 0;
            while r < buffer.len() {
                let record_type = buffer[r];
                match record_type {
                    // padding record
                    0 => {
                        let remaining = buffer.len() - r;
                        if remaining <= Record::RECORD_BASE_SIZE_IN_BYTES {
                            r += remaining;
                        } else {
                            let upper_size_byte = buffer[r + 1] as u16;
                            let lower_size_byte = buffer[r + 2] as u16;
                            let size = (upper_size_byte << 8 | lower_size_byte) as usize;
                            // skip record type byte(1) and size bytes(2)
                            r += 3;
                            // skip bytes specified by padding
                            r += size;
                        }
                    }
                    // complete record
                    1 => {
                        // read size
                        let upper_size_byte = buffer[r + 1] as u16;
                        let lower_size_byte = buffer[r + 2] as u16;
                        let size = (upper_size_byte << 8 | lower_size_byte) as usize;
                        // skip record type byte(1) and size bytes(2)
                        r += 3;
                        // read size bytes
                        let decoded: Value<K, V> =
                            bincode::deserialize(&buffer[r..r + size]).unwrap();
                        output.push(decoded);
                        r += size;
                    }
                    // start and middle records
                    2 | 3 => {
                        let upper_size_byte = buffer[r + 1] as u16;
                        let lower_size_byte = buffer[r + 2] as u16;
                        let size = (upper_size_byte << 8 | lower_size_byte) as usize;
                        // skip record type byte(1) and size bytes(2)
                        r += 3;
                        for i in 0..size {
                            record_byte_buffer.push(buffer[r + i]);
                        }
                        r += size;
                    }
                    // end
                    4 => {
                        let upper_size_byte = buffer[r + 1] as u16;
                        let lower_size_byte = buffer[r + 2] as u16;
                        let size = (upper_size_byte << 8 | lower_size_byte) as usize;
                        // skip record type byte(1) and size bytes(2)
                        r += 3;
                        for i in 0..size {
                            record_byte_buffer.push(buffer[r + i]);
                        }
                        let decoded: Value<K, V> =
                            bincode::deserialize(record_byte_buffer.as_slice()).unwrap();
                        output.push(decoded);
                        r += size;
                        // last chunk in record processed so create a new buffer
                        record_byte_buffer = Vec::new();
                    }
                    _ => {}
                }
            }
            i += 1;
        }
        return Ok(output);
    }
    Err(Errors::SSTABLE_READ_FAILED)
}