From 84ac3652415b662aae9580c008a5aa996d58c9f4 Mon Sep 17 00:00:00 2001 From: Jan Tuomi Date: Sat, 3 May 2025 00:15:02 +0300 Subject: Rename to AutereDB --- Cargo.lock | 38 +- Cargo.toml | 2 +- README.md | 22 +- autere_db/Cargo.toml | 27 + autere_db/benches/benchmark.rs | 181 ++++++ autere_db/benches/utils.rs | 122 ++++ autere_db/src/common.rs | 574 ++++++++++++++++ autere_db/src/config.rs | 140 ++++ autere_db/src/engine.rs | 852 ++++++++++++++++++++++++ autere_db/src/lib.rs | 549 ++++++++++++++++ autere_db/src/lock.rs | 117 ++++ autere_db/src/log_reader_forward.rs | 134 ++++ autere_db/src/memtable_primary.rs | 40 ++ autere_db/src/memtable_secondary.rs | 66 ++ autere_db/src/record.rs | 38 ++ autere_db/src/row.rs | 70 ++ autere_db/src/schema.rs | 7 + autere_db/tests/integration.rs | 1012 +++++++++++++++++++++++++++++ autere_db/tests/resources/test_data_1 | Bin 0 -> 266 bytes autere_db/tests/resources/test_metadata_1 | Bin 0 -> 40 bytes demo/app.py | 2 +- log_db/Cargo.toml | 27 - log_db/benches/benchmark.rs | 181 ------ log_db/benches/utils.rs | 122 ---- log_db/src/common.rs | 574 ---------------- log_db/src/config.rs | 140 ---- log_db/src/engine.rs | 852 ------------------------ log_db/src/lib.rs | 549 ---------------- log_db/src/lock.rs | 117 ---- log_db/src/log_reader_forward.rs | 134 ---- log_db/src/memtable_primary.rs | 40 -- log_db/src/memtable_secondary.rs | 66 -- log_db/src/record.rs | 38 -- log_db/src/row.rs | 70 -- log_db/src/schema.rs | 7 - log_db/tests/integration.rs | 1012 ----------------------------- log_db/tests/resources/test_data_1 | Bin 266 -> 0 bytes log_db/tests/resources/test_metadata_1 | Bin 40 -> 0 bytes py_bindings/Cargo.toml | 2 +- py_bindings/example.py | 6 +- py_bindings/pyproject.toml | 2 +- py_bindings/src/lib.rs | 66 +- 42 files changed, 3999 insertions(+), 3999 deletions(-) create mode 100644 autere_db/Cargo.toml create mode 100644 autere_db/benches/benchmark.rs create mode 100644 autere_db/benches/utils.rs create mode 100644 autere_db/src/common.rs create mode 100644 autere_db/src/config.rs create mode 100644 autere_db/src/engine.rs create mode 100644 autere_db/src/lib.rs create mode 100644 autere_db/src/lock.rs create mode 100644 autere_db/src/log_reader_forward.rs create mode 100644 autere_db/src/memtable_primary.rs create mode 100644 autere_db/src/memtable_secondary.rs create mode 100644 autere_db/src/record.rs create mode 100644 autere_db/src/row.rs create mode 100644 autere_db/src/schema.rs create mode 100644 autere_db/tests/integration.rs create mode 100644 autere_db/tests/resources/test_data_1 create mode 100644 autere_db/tests/resources/test_metadata_1 delete mode 100644 log_db/Cargo.toml delete mode 100644 log_db/benches/benchmark.rs delete mode 100644 log_db/benches/utils.rs delete mode 100644 log_db/src/common.rs delete mode 100644 log_db/src/config.rs delete mode 100644 log_db/src/engine.rs delete mode 100644 log_db/src/lib.rs delete mode 100644 log_db/src/lock.rs delete mode 100644 log_db/src/log_reader_forward.rs delete mode 100644 log_db/src/memtable_primary.rs delete mode 100644 log_db/src/memtable_secondary.rs delete mode 100644 log_db/src/record.rs delete mode 100644 log_db/src/row.rs delete mode 100644 log_db/src/schema.rs delete mode 100644 log_db/tests/integration.rs delete mode 100644 log_db/tests/resources/test_data_1 delete mode 100644 log_db/tests/resources/test_metadata_1 diff --git a/Cargo.lock b/Cargo.lock index 6a3c89c..7335a00 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -84,6 +84,24 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" +[[package]] +name = "autere_db" +version = "0.1.0" +dependencies = [ + "criterion", + "ctor", + "env_logger", + "fs2", + "log", + "once_cell", + "rand", + "rust_decimal", + "serial_test", + "tempfile", + "thiserror", + "uuid", +] + [[package]] name = "autocfg" version = "1.4.0" @@ -608,24 +626,6 @@ version = "0.4.25" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "04cbf5b083de1c7e0222a7a51dbfdba1cbe1c6ab0b15e29fff3f6c077fd9cd9f" -[[package]] -name = "log_db" -version = "0.1.0" -dependencies = [ - "criterion", - "ctor", - "env_logger", - "fs2", - "log", - "once_cell", - "rand", - "rust_decimal", - "serial_test", - "tempfile", - "thiserror", - "uuid", -] - [[package]] name = "memchr" version = "2.7.4" @@ -782,7 +782,7 @@ dependencies = [ name = "py_bindings" version = "0.1.0" dependencies = [ - "log_db", + "autere_db", "pyo3", "rust_decimal", ] diff --git a/Cargo.toml b/Cargo.toml index ec5c3ea..60b177e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,3 +1,3 @@ [workspace] resolver = "2" -members = ["log_db", "py_bindings"] +members = ["autere_db", "py_bindings"] diff --git a/README.md b/README.md index 34e8d7c..5d0e82c 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,8 @@ -# LogDB +# AutereDB -An educational endeavor in implementing a log-structured database with a focus on simplicity, understandability and performance. +AutereDB is an educational endeavor in implementing a log-structured database engine with a focus on simplicity, understandability and performance. -LogDB has the following features: +AutereDB has the following features: - Log-structured single-table storage, based on a durable append-only log - In-memory indexes for fast lookups (primary and secondary) @@ -13,37 +13,37 @@ LogDB has the following features: - Transactions based on eager exclusive locking - Batch read operations for improved performance -LogDB does not support: +AutereDB does not support: - Authentication or authorization in any capacity - Multiple tables - Type checking or schema evolution. These are outsourced to the application layer. -See the [ARCHITECTURE.md](ARCHITECTURE.md) document for more details on the design and implementation of LogDB. +For production workloads, AutereDB would benefit from a porcelain layer that provides features such as query language, networking, and monitoring. AutereDB does not come with such a layer. See the [ARCHITECTURE.md](ARCHITECTURE.md) document for more details on the design and implementation. ## Inspiration -The most significant sources of inspiration for LogDB are: +The most significant sources of inspiration for AutereDB are: - [SQLite](https://www.sqlite.org/index.html) for its filesystem storage and locking mechanisms. - [Designing Data-Intensive Applications (book)](https://www.oreilly.com/library/view/designing-data-intensive-applications/9781491903063/) for its excellent overview of database internals and in-depth analysis of log-structured storage. - LogDB is heavily based on the design outlined in chapter 3. + AutereDB is heavily based on the design outlined in chapter 3. ## Usage in Rust -Add LogDB as a dependency in your `Cargo.toml`. +Add AutereDB as a dependency in your `Cargo.toml`. ```toml [dependencies] -log_db = { git = "https://github.com/jantuomi/log_db.git" } +autere_db = { git = "https://github.com/jantuomi/autere_db.git" } ``` Then use it in your code like so: ```rust -use log_db::*; +use autere_db::*; // Define a type that represents your fields (columns) #[derive(Eq, PartialEq, Clone, Debug)] @@ -170,4 +170,4 @@ db.upsert([Value.int(10)]) ## Copyright and license -LogDB is licensed under the Apache License, Version 2.0. © 2024 Jan Tuomi. +AutereDB is licensed under the Apache License, Version 2.0. © 2024 Jan Tuomi. diff --git a/autere_db/Cargo.toml b/autere_db/Cargo.toml new file mode 100644 index 0000000..c9ae7d2 --- /dev/null +++ b/autere_db/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "autere_db" +version = "0.1.0" +edition = "2021" + +[lib] +crate-type = ["lib", "cdylib", "staticlib"] + +[dependencies] +fs2 = "0.4.3" +log = "0.4.22" +once_cell = "1.20.2" +rust_decimal = { version = "1.36.0", features = [] } +tempfile = "3.13.0" +thiserror = "2.0.1" +uuid = { version = "1.11.0", features = ["v4"] } + +[dev-dependencies] +ctor = "0.2.8" +env_logger = "0.11.5" +serial_test = "3.1.1" +criterion = { version = "0.5", features = ["html_reports"] } +rand = "0.8.5" + +[[bench]] +name = "benchmark" +harness = false diff --git a/autere_db/benches/benchmark.rs b/autere_db/benches/benchmark.rs new file mode 100644 index 0000000..648c9f6 --- /dev/null +++ b/autere_db/benches/benchmark.rs @@ -0,0 +1,181 @@ +mod utils; + +use std::collections::HashSet; + +use autere_db::*; +use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion}; +use tempfile; +use utils::*; + +pub fn upsert_compacted(c: &mut Criterion) { + let mut group = c.benchmark_group("upsert_compacted"); + let data_dir_obj = tempfile::tempdir().expect("Failed to get tmpdir"); + let data_dir = data_dir_obj + .path() + .to_str() + .expect("Failed to convert tmpdir path to str"); + let mut db = DB::configure() + .fields(Inst::fields()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(data_dir) + .initialize() + .expect("Failed to initialize DB"); + + let mut insts = Vec::new(); + for size in [100_000, 1_000_000, 10_000_000] { + println!("Prefilling DB to {} entries", size); + prefill_db(&mut db, &mut insts, size, true).expect("Failed to prefill DB"); + + group.bench_with_input(BenchmarkId::from_parameter(size), &size, |b, &_size| { + b.iter(|| { + let inst = random_inst(0, size as i64 + 1); + let result = db.upsert(black_box(inst)).unwrap(); + assert!(result == ()); + }); + }); + } +} + +pub fn delete_existing_compacted(c: &mut Criterion) { + let mut group = c.benchmark_group("delete_existing_compacted"); + let data_dir_obj = tempfile::tempdir().expect("Failed to get tmpdir"); + let data_dir = data_dir_obj + .path() + .to_str() + .expect("Failed to convert tmpdir path to str"); + let mut db = DB::configure() + .fields(Inst::fields()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(data_dir) + .initialize() + .expect("Failed to initialize DB"); + + let mut insts = Vec::new(); + for size in [100_000, 1_000_000, 10_000_000] { + println!("Prefilling DB to {} entries", size); + prefill_db(&mut db, &mut insts, size, true).expect("Failed to prefill DB"); + + let pk_set: HashSet = insts.iter().map(|inst| inst.id).collect(); + let mut pk_set_it = pk_set.into_iter(); + + group.bench_with_input(BenchmarkId::from_parameter(size), &size, |b, &_size| { + b.iter(|| { + let result = db + .delete(black_box(&Value::Int(pk_set_it.next().unwrap()))) + .unwrap(); + assert!(result.is_some()); + }); + }); + } +} + +pub fn upsert_write_durability(c: &mut Criterion) { + let mut group = c.benchmark_group("upsert_write_durability"); + + for mode in [WriteDurability::Flush, WriteDurability::FlushSync] { + group.bench_with_input(BenchmarkId::from_parameter(&mode), &mode, |b, _mode| { + let data_dir_obj = tempfile::tempdir().expect("Failed to get tmpdir"); + let data_dir = data_dir_obj + .path() + .to_str() + .expect("Failed to convert tmpdir path to str"); + let mut db = DB::configure() + .fields(Inst::fields()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(data_dir) + .write_durability(mode.clone()) + .initialize() + .expect("Failed to initialize DB"); + + b.iter(|| { + let inst = random_inst(0, 1000); + let result = db.upsert(black_box(inst)).unwrap(); + assert!(result == ()); + }); + }); + } +} + +pub fn get_existing_compacted(c: &mut Criterion) { + let mut group = c.benchmark_group("get_existing_compacted"); + + let data_dir_obj = tempfile::tempdir().expect("Failed to get tmpdir"); + let data_dir = data_dir_obj + .path() + .to_str() + .expect("Failed to convert tmpdir path to str"); + let mut db = DB::configure() + .fields(Inst::fields()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(data_dir) + .initialize() + .expect("Failed to initialize DB"); + + let mut insts = Vec::new(); + let mut inst_index = 0; + for size in [100_000, 1_000_000, 10_000_000] { + println!("Prefilling DB to {} entries", size); + prefill_db(&mut db, &mut insts, size, true).expect("Failed to prefill DB"); + + group.bench_with_input(BenchmarkId::from_parameter(size), &size, |b, &_size| { + b.iter(|| { + let id = insts[inst_index].id; + let result = db.get(black_box(&Value::Int(id))).unwrap(); + assert!(result.is_some()); + inst_index = (inst_index + 1) % insts.len(); + }); + }); + } +} + +pub fn find_by_existing_compacted(c: &mut Criterion) { + let mut group = c.benchmark_group("find_by_existing_compacted"); + + let data_dir_path = tempfile::tempdir() + .expect("Failed to get tmpdir") + .into_path(); + let data_dir = data_dir_path + .to_str() + .expect("Failed to convert tmpdir path to str"); + let mut db = DB::configure() + .fields(Inst::fields()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(data_dir) + .initialize() + .expect("Failed to initialize DB"); + + let mut insts = Vec::new(); + let mut inst_index = 0; + for size in [100_000, 1_000_000, 10_000_000] { + println!("Prefilling DB to {} entries", size); + prefill_db(&mut db, &mut insts, size, true).expect("Failed to prefill DB"); + println!("DB prefilling done, insts.len = {}", insts.len()); + + group.bench_with_input(BenchmarkId::from_parameter(size), &size, |b, &_size| { + b.iter(|| { + let name = insts[inst_index].name.clone(); + let result = db + .find_by(black_box(&Field::Name), black_box(&Value::String(name))) + .unwrap(); + assert!(result.len() > 0); + inst_index = (inst_index + 1) % insts.len(); + }); + }); + } +} + +// Register the benchmark group +criterion_group!( + benches, + upsert_compacted, + delete_existing_compacted, + upsert_write_durability, + get_existing_compacted, + find_by_existing_compacted, +); +criterion_main!(benches); diff --git a/autere_db/benches/utils.rs b/autere_db/benches/utils.rs new file mode 100644 index 0000000..9b0a0ea --- /dev/null +++ b/autere_db/benches/utils.rs @@ -0,0 +1,122 @@ +use autere_db::*; +use rand::distributions::Alphanumeric; +use rand::Rng; +use std::fmt::Debug; + +#[derive(Eq, PartialEq, Clone, Debug)] +pub enum Field { + Id, + Name, + Data, +} + +#[derive(PartialEq, Eq, Debug, Clone)] +pub struct Inst { + pub id: i64, + pub name: String, + pub data: Vec, +} + +impl AsRef for Field { + fn as_ref(&self) -> &str { + match self { + Field::Id => "id", + Field::Name => "name", + Field::Data => "data", + } + } +} + +impl Into for Field { + fn into(self) -> String { + self.as_ref().to_string() + } +} + +impl From for Record { + fn from(inst: Inst) -> Self { + vec![ + Value::Int(inst.id), + Value::String(inst.name), + Value::Bytes(inst.data), + ] + .into() + } +} + +impl From for Inst { + fn from(record: Record) -> Self { + let mut it = record.into_iter(); + Inst { + id: match it.next().unwrap() { + Value::Int(id) => id, + _ => panic!("Expected Int"), + }, + name: match it.next().unwrap() { + Value::String(name) => name, + _ => panic!("Expected String"), + }, + data: match it.next().unwrap() { + Value::Bytes(data) => data, + _ => panic!("Expected Bytes"), + }, + } + } +} + +impl Inst { + pub fn fields() -> Vec { + vec![Field::Id, Field::Name, Field::Data] + } + pub fn primary_key() -> Field { + Field::Id + } + pub fn secondary_keys() -> Vec { + vec![Field::Name] + } +} + +// Function to generate a random integer +pub fn random_int(from: i64, to: i64) -> i64 { + let mut rng = rand::thread_rng(); + rng.gen_range(from..to) +} + +// Function to generate a random string +pub fn random_string(len: usize) -> String { + let mut rng = rand::thread_rng(); + (0..len).map(|_| rng.sample(Alphanumeric) as char).collect() +} + +// Function to generate random bytes +pub fn random_bytes(len: usize) -> Vec { + let mut rng = rand::thread_rng(); + (0..len).map(|_| rng.gen()).collect() +} + +// Function to generate a random Inst +pub fn random_inst(from_id: i64, to_id: i64) -> Inst { + Inst { + id: random_int(from_id, to_id), // Random int value between 0..1000 + name: random_string(5), // Random string of length 5 + data: random_bytes(10), // Random bytes of length 10 + } +} + +pub fn prefill_db( + db: &mut DB, + insts: &mut Vec, + n_records: usize, + compact: bool, +) -> DBResult<()> { + for i in 0..(n_records - insts.len()) { + let inst = random_inst(0, n_records as i64); + insts.push(inst.clone()); + db.upsert(inst)?; + if i % 1000 == 0 && compact { + db.do_maintenance_tasks()?; + } + } + + Ok(()) +} diff --git a/autere_db/src/common.rs b/autere_db/src/common.rs new file mode 100644 index 0000000..0f6a058 --- /dev/null +++ b/autere_db/src/common.rs @@ -0,0 +1,574 @@ +use super::*; + +use std::collections::btree_map::Values; +// For Unix-like systems +#[cfg(unix)] +use std::os::unix::fs::MetadataExt; + +// For Windows +#[cfg(windows)] +use std::os::windows::fs::MetadataExt; + +pub const ACTIVE_SYMLINK_FILENAME: &str = "active"; +pub const LOCK_FILENAME: &str = "lock"; +pub const EXCL_LOCK_REQ_FILENAME: &str = "excl_lock_req"; +pub const INITIALIZED_FILENAME: &str = "initialized"; + +pub const METADATA_FILE_HEADER_SIZE: usize = 24; +pub const METADATA_ROW_LENGTH: usize = 16; +pub const LOCK_WAIT_MAX_MS: u64 = 1000; + +// Serialized value tags +pub const B_NULL: u8 = 0x0; +pub const B_INT: u8 = 0x1; +pub const B_DECIMAL: u8 = 0x2; +pub const B_STRING: u8 = 0x3; +pub const B_BYTES: u8 = 0x4; +// Tombstone marker tags +pub const B_LIVE: u8 = 0x0; +pub const B_TOMBSTONE: u8 = 0xFF; + +pub fn metadata_filename(num: u16) -> String { + format!("metadata.{}", num) +} + +pub type DBResult = Result; + +#[derive(Debug, Error)] +pub enum DBError { + #[error("lock request failed: {0}")] + LockRequestError(String), + #[error("validation failed: {0}")] + ValidationError(String), + #[error("consistency check failed: {0}")] + ConsistencyError(String), + #[error("invalid transaction: {0}")] + TransactionError(String), + #[error("unexpected IO error: {0}")] + IOError(#[from] io::Error), +} + +#[derive(Debug, Error)] +pub enum LogKeyMapError { + #[error("log key not found in map")] + NotFoundError, + #[error("attempted to remove last element of non-empty map")] + RemovingLastElementError, +} + +/// LogKey is a packed struct that contains: +/// - a log segment number (16 bits) +/// - a log index within the segment (48 bits) +#[derive(Debug, Clone, Eq, PartialEq, Hash, Ord, PartialOrd)] +pub struct LogKey(u64); + +impl LogKey { + pub fn new(segment_num: u16, index: u64) -> Self { + assert!(index < (1 << 48), "Index must fit in 48 bits"); + LogKey((segment_num as u64) << 48 | index) + } + + pub fn segment_num(&self) -> u16 { + (self.0 >> 48) as u16 + } + + pub fn index(&self) -> u64 { + self.0 & 0x0000_FFFF_FFFF_FFFF + } +} + +/// LogKeyMap is a non-empty map of PK => LogKey mappings. +#[derive(Debug, Clone, Eq, PartialEq)] +pub struct LogKeyMap { + map: BTreeMap, +} + +impl LogKeyMap { + /// Create a new LogKeyMap with an initial mapping. + /// The initial mapping is required since LogKeyMap must be non-empty. + pub fn new_with_initial(pk: IndexableValue, log_key: LogKey) -> Self { + let mut map = BTreeMap::new(); + map.insert(pk, log_key); + LogKeyMap { map } + } + + pub fn contains_pk(&self, key: &IndexableValue) -> bool { + self.map.contains_key(key) + } + + /// The number of LogKeys in the map. + pub fn len(&self) -> usize { + self.map.len() + } + + /// Insert a PK -> LogKey mapping into the map. + pub fn insert(&mut self, key: IndexableValue, log_key: LogKey) { + self.map.insert(key, log_key); + } + + /// Remove a mapping from the map. Return Ok(()) if the key was found and removed. + /// Return `LogKeyMapError::RemovingLastElementError` if trying to remove the last element. + /// Return `LogKeyMapError::NotFoundError` if the key was not found. + pub fn remove_pk(&mut self, key: &IndexableValue) -> Result<(), LogKeyMapError> { + if self.map.len() == 1 { + return Err(LogKeyMapError::RemovingLastElementError); + } + let removed = self.map.remove(key); + + if removed.is_none() { + return Err(LogKeyMapError::NotFoundError); + } + + assert!( + self.map.len() > 0, + "LogKeyMap should not be empty after removal" + ); + + Ok(()) + } + + /// Get a reference to the set of LogKeys. + pub fn log_keys(&self) -> Values { + self.map.values() + } +} + +pub static APPEND_MODE: Lazy = Lazy::new(|| { + let mut options = fs::OpenOptions::new(); + options.read(true).append(true); + options +}); +pub static READ_MODE: Lazy = Lazy::new(|| { + let mut options = fs::OpenOptions::new(); + options.read(true); + options +}); +pub static WRITE_MODE: Lazy = Lazy::new(|| { + let mut options = fs::OpenOptions::new(); + options.read(true).write(true); + options +}); + +pub struct MetadataHeader { + pub version: u8, + pub uuid: Uuid, +} + +const METADATA_HEADER_PADDING: &[u8] = &[0; 7]; +impl MetadataHeader { + pub fn serialize(&self) -> [u8; METADATA_FILE_HEADER_SIZE] { + let mut header = [0u8; METADATA_FILE_HEADER_SIZE]; + header[0] = self.version; + header[1..8].copy_from_slice(METADATA_HEADER_PADDING); + header[8..].copy_from_slice(self.uuid.as_bytes()); + + header + } + + pub fn deserialize(bytes: &[u8]) -> Self { + assert_eq!(bytes.len(), METADATA_FILE_HEADER_SIZE); + + let version = bytes[0]; + let uuid = Uuid::from_slice(&bytes[8..24]).expect("Failed to deserialize Uuid"); + + MetadataHeader { version, uuid } + } +} + +#[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Hash)] +pub enum IndexableValue { + Null, + Int(i64), + Decimal(Decimal), + String(String), +} + +#[derive(Debug, Clone)] +pub enum Value { + Null, + Int(i64), + Decimal(Decimal), + String(String), + Bytes(Vec), +} + +impl PartialEq for Value { + fn eq(&self, other: &Self) -> bool { + match (self, other) { + (Value::Int(a), Value::Int(b)) => a == b, + (Value::Decimal(a), Value::Decimal(b)) => a == b, + (Value::String(a), Value::String(b)) => a == b, + (Value::Bytes(a), Value::Bytes(b)) => a == b, + (Value::Null, Value::Null) => true, + _ => false, + } + } +} +impl Eq for Value {} + +impl Value { + pub fn serialize(&self) -> Vec { + match self { + Value::Null => vec![B_NULL], + Value::Int(i) => { + let mut bytes = Vec::with_capacity(1 + 16); + bytes.push(B_INT); + bytes.extend_from_slice(&i.to_be_bytes()); + bytes + } + Value::Decimal(d) => { + let mut bytes = Vec::with_capacity(1 + 16); + bytes.push(B_DECIMAL); + bytes.extend_from_slice(&d.serialize()); + bytes + } + Value::String(s) => { + let len = s.len(); + let mut bytes = Vec::with_capacity(1 + 8 + len); + bytes.push(B_STRING); + bytes.extend_from_slice(&(len as u64).to_be_bytes()); + bytes.extend_from_slice(s.as_bytes()); + bytes + } + Value::Bytes(b) => { + let len = b.len(); + let mut bytes = Vec::with_capacity(1 + 8 + len); + bytes.push(B_BYTES); + bytes.extend_from_slice(&(len as u64).to_be_bytes()); + bytes.extend_from_slice(b); + bytes + } + } + } + + /// Deserialize a Value from a byte slice. + /// Returns the deserialized Value and the number of bytes consumed. + pub fn deserialize(bytes: &[u8]) -> (Value, usize) { + match bytes[0] { + B_NULL => (Value::Null, 1), + B_INT => { + let mut int_bytes = [0; 8]; + int_bytes.copy_from_slice(&bytes[1..1 + 8]); + (Value::Int(i64::from_be_bytes(int_bytes)), 1 + 8) + } + B_DECIMAL => { + let mut decimal_bytes = [0; 16]; + decimal_bytes.copy_from_slice(&bytes[1..1 + 16]); + (Value::Decimal(Decimal::deserialize(decimal_bytes)), 1 + 16) + } + B_STRING => { + let length_bytes = &bytes[1..1 + 8]; + let length = u64::from_be_bytes(length_bytes.try_into().unwrap()) as usize; + ( + Value::String( + String::from_utf8(bytes[1 + 8..1 + 8 + length].to_vec()).unwrap(), + ), + 1 + 8 + length, + ) + } + B_BYTES => { + let length_bytes = &bytes[1..1 + 8]; + let length = u64::from_be_bytes(length_bytes.try_into().unwrap()) as usize; + ( + Value::Bytes(bytes[1 + 8..1 + 8 + length].to_vec()), + 1 + 8 + length, + ) + } + _ => panic!("Invalid tag: {}", bytes[0]), + } + } + + pub fn as_indexable(&self) -> Option { + match self { + Value::Null => Some(IndexableValue::Null), + Value::Int(i) => Some(IndexableValue::Int(*i)), + Value::Decimal(d) => Some(IndexableValue::Decimal(d.clone())), + Value::String(s) => Some(IndexableValue::String(s.clone())), + _ => None, + } + } +} + +pub fn get_secondary_memtable_index_by_field(sks: &Vec, field: &str) -> Option { + sks.iter().position(|schema_field| schema_field == field) +} + +pub fn is_file_same_as_path(file: &File, path: &PathBuf) -> DBResult { + // Get the metadata for the open file handle + let file_metadata = file.metadata()?; + + // Get the metadata for the file at the specified path + let path_metadata = metadata(path)?; + + // Platform-specific comparison + #[cfg(unix)] + { + Ok( + file_metadata.dev() == path_metadata.dev() + && file_metadata.ino() == path_metadata.ino(), + ) + } + + #[cfg(windows)] + { + Ok(file_metadata.file_index() == path_metadata.file_index() + && file_metadata.volume_serial_number() == path_metadata.volume_serial_number()) + } +} + +pub fn symlink(original: &Path, link: &Path) -> io::Result<()> { + #[cfg(unix)] + { + std::os::unix::fs::symlink(original, link) + } + + #[cfg(windows)] + { + std::os::windows::fs::symlink_file(original, link) + } +} + +/// Set the active segment to the segment with the given ordinal number. +pub fn set_active_segment(data_dir_path: &Path, segment_num: u16) -> DBResult<()> { + let tmp_uuid = Uuid::new_v4(); + let tmp_filename = format!("active_{}", tmp_uuid.to_string()); + let tmp_path = data_dir_path.join(tmp_filename); + + let metadata_filename = format!("metadata.{}", segment_num); + let metadata_path = Path::new(&metadata_filename); + let active_symlink = data_dir_path.join(ACTIVE_SYMLINK_FILENAME); + + symlink(&metadata_path, &tmp_path)?; + fs::rename(&tmp_path, &active_symlink)?; + + Ok(()) +} + +/// Create a new segment metadata file and return its number and path. +/// A metadata file contains the segment metadata, including the UUID of the data file. +/// See `ARCHITECTURE.md` for the file format. +pub fn create_segment_metadata_file( + data_dir_path: &Path, + data_file_uuid: &Uuid, +) -> DBResult<(u16, PathBuf)> { + let current_greatest_num = greatest_segment_number(data_dir_path)?; + let new_num = current_greatest_num + 1; + + let metadata_filename = format!("metadata.{}", new_num); + let metadata_path = data_dir_path.join(metadata_filename); + + let mut metadata_file = fs::OpenOptions::new() + .create(true) + .append(true) + .open(&metadata_path)?; + + let metadata_header = MetadataHeader { + version: 1, + uuid: *data_file_uuid, + }; + + metadata_file.write_all(&metadata_header.serialize())?; + metadata_file.flush()?; + + let len = metadata_file.seek(io::SeekFrom::End(0))?; + assert!(len >= METADATA_FILE_HEADER_SIZE as u64); + assert_eq!((len - METADATA_FILE_HEADER_SIZE as u64) % 16, 0); + + Ok((new_num, metadata_path)) +} + +/// Parse the segment number from a metadata file path +pub fn parse_segment_number(metadata_path: &Path) -> DBResult { + let filename = metadata_path + .file_name() + .expect("No filename in symlink") + .to_str() + .expect("Filename was not valid UTF-8"); + + // parse number from format "metadata.1" + let segment_number = filename + .split('.') + .last() + .expect("Filename did not have a number") + .parse::(); + + segment_number.map_err(|_| { + DBError::ValidationError("Failed to parse segment number from filename".to_owned()) + }) +} + +/// Get the number of the segment with the greatest ordinal. +/// This is the newest segment, i.e. the one that is pointed to by the `active` symlink. +/// If there are no segments yet, returns 0. +pub fn greatest_segment_number(data_dir_path: &Path) -> DBResult { + let active_symlink = data_dir_path.join(ACTIVE_SYMLINK_FILENAME); + + if !fs::exists(&active_symlink)? { + return Ok(0); + } + + let segment_metadata_path = fs::read_link(&active_symlink)?; + parse_segment_number(&segment_metadata_path) +} + +/// Create a new segment data file and return its UUID. +/// A data file contains the segment data, tightly packed without separators. +/// An accompanying metadata file is required to interpret the data. +pub fn create_segment_data_file(data_dir_path: &Path) -> DBResult<(Uuid, PathBuf)> { + let uuid = Uuid::new_v4(); + let new_segment_path = data_dir_path.join(uuid.to_string()); + fs::OpenOptions::new() + .create(true) + .append(true) + .open(&new_segment_path)?; + + Ok((uuid, new_segment_path)) +} + +/// Reads the metadata header from the metadata file. +/// Leaves the file seek head at the beginning of the records, after the header. +pub fn read_metadata_header(metadata_file: &mut fs::File) -> DBResult { + metadata_file.seek(SeekFrom::Start(0))?; + let mut buf = [0u8; METADATA_FILE_HEADER_SIZE]; + metadata_file.read_exact(&mut buf)?; + + let header = MetadataHeader::deserialize(&buf); + Ok(header) +} + +pub fn validate_metadata_header(header: &MetadataHeader) -> DBResult<()> { + if header.version != 1 { + return Err(DBError::ValidationError( + "Unsupported metadata file version".to_owned(), + )); + } + + Ok(()) +} + +pub enum IsMetadatafileValidResult { + Ok, + ReplaceFile, + TruncateToSize(u64), +} + +pub fn is_metadata_file_valid(metadata_file: &mut fs::File) -> DBResult { + let size = metadata_file.seek(SeekFrom::End(0))? as usize; + + if size < METADATA_FILE_HEADER_SIZE { + return Ok(IsMetadatafileValidResult::ReplaceFile); + } + + // The data section must be a multiple of 16 bytes. + // Otherwise, the non-aligned part of the file is dropped. + let data_section_len = size - METADATA_FILE_HEADER_SIZE; + let remainder = data_section_len % 16; + if remainder != 0 { + return Ok(IsMetadatafileValidResult::TruncateToSize( + (size - remainder) as u64, + )); + } + + Ok(IsMetadatafileValidResult::Ok) +} + +/// Check that the active metadata file is well-formed and repair it if necessary. +/// The metadata file is considered well-formed if its size is, in pseudocode, `header_size + n * record_size`. +/// If the file is not well-formed, it is truncated to the last well-formed record using +/// a temporary file and an atomic move operation. +/// +/// `self.active_metadata_file` must be a locked file handle opened with read permissions. +/// The function leaves the seek head in an unspecified position. +/// +/// Returns `false` if the file was repaired and rotated, `true` if no action was taken. +pub fn ensure_active_metadata_is_valid( + data_dir: &Path, + metadata_file: &mut fs::File, +) -> DBResult { + let current_len = metadata_file.seek(SeekFrom::End(0))? as usize; + + match is_metadata_file_valid(metadata_file)? { + IsMetadatafileValidResult::Ok => return Ok(true), + IsMetadatafileValidResult::ReplaceFile => { + let active_target = fs::read_link(data_dir.join(ACTIVE_SYMLINK_FILENAME))?; + let active_path = data_dir.join(&active_target); + warn!( + "Metadata file \"{}\" is malformed ({} bytes), replacing it with an empty file", + active_target.display(), + current_len, + ); + let mut tmp_file = tempfile::NamedTempFile::new()?; + + let header = MetadataHeader { + version: 1, + uuid: Uuid::new_v4(), + }; + + tmp_file.write_all(&header.serialize())?; + tmp_file.flush()?; + + fs::rename(tmp_file.path(), active_path)?; + + debug!("Replaced metadata file"); + return Ok(false); + } + IsMetadatafileValidResult::TruncateToSize(new_size) => { + let active_target = fs::read_link(data_dir.join(ACTIVE_SYMLINK_FILENAME))?; + let active_path = data_dir.join(&active_target); + warn!( + "Metadata file \"{}\" is malformed ({} bytes), truncating it to {} bytes", + active_target.display(), + current_len, + new_size + ); + + let mut tmp_file = tempfile::NamedTempFile::new()?; + + let mut buf = vec![0; new_size as usize]; + metadata_file.seek(SeekFrom::Start(0))?; + metadata_file.read_exact(&mut buf)?; + + tmp_file.write_all(&buf)?; + tmp_file.flush()?; + + fs::rename(tmp_file.path(), active_path)?; + + debug!("Truncated metadata file"); + return Ok(false); + } + } +} + +pub struct OwnedBounds { + start: Bound, + end: Bound, +} + +impl OwnedBounds { + pub fn new(start: Bound, end: Bound) -> Self { + OwnedBounds { start, end } + } +} + +impl RangeBounds for OwnedBounds { + fn start_bound(&self) -> Bound<&T> { + self.start.as_ref() + } + + fn end_bound(&self) -> Bound<&T> { + self.end.as_ref() + } +} + +#[derive(Debug, Clone)] +pub struct QueryParams { + pub offset: usize, + pub limit: usize, + pub sort_asc: bool, +} + +pub static DEFAULT_QUERY_PARAMS: QueryParams = QueryParams { + offset: 0, + limit: usize::MAX, + sort_asc: true, +}; diff --git a/autere_db/src/config.rs b/autere_db/src/config.rs new file mode 100644 index 0000000..e8c6fa1 --- /dev/null +++ b/autere_db/src/config.rs @@ -0,0 +1,140 @@ +use super::*; + +pub struct ConfigBuilder { + data_dir: Option, + segment_size: Option, + write_durability: Option, + read_consistency: Option, + + fields: Option>, + primary_key: Option, + secondary_keys: Option>, +} + +impl ConfigBuilder { + pub fn new() -> ConfigBuilder { + ConfigBuilder { + data_dir: None, + segment_size: None, + write_durability: None, + read_consistency: None, + + fields: None, + primary_key: None, + secondary_keys: None, + } + } + + /// The directory where the database will store its data. + pub fn data_dir(mut self, data_dir: impl Into) -> Self { + self.data_dir = Some(data_dir.into()); + self + } + + /// The maximum size of a segment file in bytes. + /// Once a segment file reaches this size, it can be closed, rotated and compacted. + /// Note that this is not a hard limit: if `db.do_maintenance_tasks()` is not called, + /// the segment file may continue to grow. + pub fn segment_size(mut self, segment_size: usize) -> Self { + self.segment_size = Some(segment_size); + self + } + + /// The write durability policy for the database. + /// This determines how writes are persisted to disk. + /// The default is WriteDurability::Flush. + pub fn write_durability(mut self, write_durability: WriteDurability) -> Self { + self.write_durability = Some(write_durability); + self + } + + /// The read consistency policy for the database. + /// This determines how recent writes are visible when reading. + /// See individual `ReadConsistency` enum values for more information. + /// The default is ReadConsistency::Strong. + pub fn read_consistency(mut self, read_consistency: ReadConsistency) -> Self { + self.read_consistency = Some(read_consistency); + self + } + + pub fn fields(mut self, schema: Vec>) -> Self { + self.fields = Some(schema.into_iter().map(|s| s.into()).collect()); + self + } + + pub fn primary_key(mut self, primary_key: impl Into) -> Self { + self.primary_key = Some(primary_key.into()); + self + } + + pub fn secondary_keys(mut self, secondary_keys: Vec>) -> Self { + self.secondary_keys = Some(secondary_keys.into_iter().map(|s| s.into()).collect()); + self + } + + pub fn initialize(self) -> DBResult { + let schema = self + .fields + .ok_or_else(|| DBError::ValidationError("Schema not set".to_string()))?; + let primary_key = self + .primary_key + .ok_or_else(|| DBError::ValidationError("Primary key not set".to_string()))?; + + let config = Config { + schema, + primary_key, + secondary_keys: self.secondary_keys.unwrap_or_default(), + + data_dir: self.data_dir.clone().unwrap_or("db_data".to_string()), + segment_size: self.segment_size.unwrap_or(4 * 1024 * 1024), // 4MB + write_durability: self + .write_durability + .clone() + .unwrap_or(WriteDurability::Flush), + read_consistency: self + .read_consistency + .clone() + .unwrap_or(ReadConsistency::Strong), + }; + + DB::initialize(config) + } +} + +#[derive(Clone)] +pub struct Config { + pub schema: Vec, + pub primary_key: String, + pub secondary_keys: Vec, + pub data_dir: String, + pub segment_size: usize, + pub write_durability: WriteDurability, + pub read_consistency: ReadConsistency, +} + +#[derive(Debug, Clone, Eq, PartialEq)] +pub enum ReadConsistency { + /// Reads by client A are guaranteed to see writes by themselves and any writes by other clients B + /// that were done before last index refresh. You must call `refresh_indexes()` manually to refresh indexes. + Eventual, + /// Reads by client A are guaranteed to see all writes. This is slower: all reads must first + /// refresh indexes. + Strong, +} + +#[derive(Debug, Clone, Eq, PartialEq)] +pub enum WriteDurability { + /// Changes are written to the OS write buffer but not immediately synced to disk. + /// This is generally recommended. Most OSes will sync the write buffer to disk within a few seconds. + Flush, + /// Changes are written to the OS write buffer and synced to disk immediately. + /// Offers the best durability guarantees but is a lot slower. + FlushSync, +} + +impl Display for WriteDurability { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> { + write!(f, "{:?}", self)?; + Ok(()) + } +} diff --git a/autere_db/src/engine.rs b/autere_db/src/engine.rs new file mode 100644 index 0000000..6fb9edf --- /dev/null +++ b/autere_db/src/engine.rs @@ -0,0 +1,852 @@ +use super::*; + +pub struct Engine { + pub config: Config, + pub lock_manager: LockManager, + + data_dir_path: PathBuf, + primary_key_index: usize, + refresh_next_logkey: LogKey, + + pub tx_active: bool, + pub tx_log: Vec, + + active_metadata_file: fs::File, + active_data_file: fs::File, + + // TODO: these could be made private. Currently they are public for testing in lib.rs. + pub primary_memtable: PrimaryMemtable, + pub secondary_memtables: Vec, +} + +impl Engine { + pub fn initialize(config: Config) -> DBResult { + info!("Initializing DB..."); + // If data_dir does not exist or is empty, create it and any necessary files. + // After creation, the directory should always be in a complete state without missing files. + + // Ensure the data directory exists + let data_dir_path = Path::new(&config.data_dir).to_path_buf(); + match fs::create_dir(&data_dir_path) { + Ok(_) => {} + Err(e) => { + if e.kind() != io::ErrorKind::AlreadyExists { + return Err(DBError::IOError(e)); + } + } + } + + // Create the lock file first to prevent multiple concurrent initializations + let mut lock_manager = LockManager::new(data_dir_path.clone())?; + lock_manager.lock_exclusive()?; + + // We have acquired the lock, check if the data directory is in a complete state + // If not, initialize it, otherwise skip. + if !fs::exists(data_dir_path.join(INITIALIZED_FILENAME))? { + // Delete all files except the lock files to ensure a clean state + for entry in fs::read_dir(&data_dir_path)? { + let entry = entry?; + let path = entry.path(); + if path.is_file() + && path.file_name().unwrap() != LOCK_FILENAME + && path.file_name().unwrap() != EXCL_LOCK_REQ_FILENAME + { + fs::remove_file(&path)?; + } + } + + // Create the initial segment files + let (segment_uuid, _) = create_segment_data_file(&data_dir_path)?; + let (segment_num, _) = create_segment_metadata_file(&data_dir_path, &segment_uuid)?; + set_active_segment(&data_dir_path, segment_num)?; + + // Create the initialized file to indicate that the directory is in a complete state + fs::File::create(data_dir_path.join(INITIALIZED_FILENAME))?; + } + + // Calculate the index of the primary value in a record + let primary_key_index = config + .schema + .iter() + .position(|field| field == &config.primary_key) + .ok_or(DBError::ValidationError( + "Primary key not found in schema after initialize".to_owned(), + ))?; + + // Join primary key and secondary keys vec into a single vec + let mut all_keys = vec![&config.primary_key]; + all_keys.extend(&config.secondary_keys); + + // If any of the keys is not in the schema or + // is not an IndexableValue, return an error + for &key in &all_keys { + let _ = config.schema.iter().find(|&field| field == key).ok_or( + DBError::ValidationError("Key must be present in the field schema".to_owned()), + )?; + } + let primary_memtable = PrimaryMemtable::new(); + let secondary_memtables = config + .secondary_keys + .iter() + .map(|_| SecondaryMemtable::new()) + .collect(); + + let active_symlink = Path::new(&config.data_dir).join(ACTIVE_SYMLINK_FILENAME); + + let active_target = fs::read_link(&active_symlink)?; + let active_metadata_path = Path::new(&config.data_dir).join(active_target); + let mut active_metadata_file = APPEND_MODE.open(&active_metadata_path)?; + + let active_metadata_header = read_metadata_header(&mut active_metadata_file)?; + validate_metadata_header(&active_metadata_header)?; + + let active_data_path = + Path::new(&config.data_dir).join(active_metadata_header.uuid.to_string()); + let active_data_file = APPEND_MODE.open(&active_data_path)?; + + let mut engine = Engine { + config, + lock_manager, + data_dir_path, + primary_key_index, + primary_memtable, + secondary_memtables, + active_metadata_file, + active_data_file, + refresh_next_logkey: LogKey::new(1, 0), + tx_active: false, + tx_log: vec![], + }; + + info!("Rebuilding memtable indexes..."); + engine.refresh_indexes()?; + + info!("Database ready."); + + engine.lock_manager.unlock()?; + Ok(engine) + } + + pub fn refresh_indexes(&mut self) -> DBResult<()> { + let active_symlink_path = self.data_dir_path.join(ACTIVE_SYMLINK_FILENAME); + let active_target = fs::read_link(active_symlink_path)?; + let active_metadata_path = self.data_dir_path.join(active_target); + + let to_segnum = parse_segment_number(&active_metadata_path)?; + let from_segnum = self.refresh_next_logkey.segment_num(); + let mut from_index = self.refresh_next_logkey.index(); + + for segnum in from_segnum..=to_segnum { + let metadata_path = self.data_dir_path.join(metadata_filename(segnum)); + let mut metadata_file = READ_MODE.open(&metadata_path)?; + + let metadata_len = metadata_file.seek(SeekFrom::End(0))?; + if (metadata_len - METADATA_FILE_HEADER_SIZE as u64) % METADATA_ROW_LENGTH as u64 != 0 { + return Err(DBError::ConsistencyError(format!( + "Metadata file {} has invalid size: {}", + metadata_path.display(), + metadata_len + ))); + } + + let metadata_header = read_metadata_header(&mut metadata_file)?; + validate_metadata_header(&metadata_header)?; + + let data_path = self.data_dir_path.join(metadata_header.uuid.to_string()); + let data_file = READ_MODE.open(data_path)?; + + for ForwardLogReaderItem { row, index } in + ForwardLogReader::new_with_index(metadata_file, data_file, from_index) + { + let log_key = LogKey::new(segnum, index); + + if row.tombstone { + self.remove_row_from_memtables(&row.values); + } else { + self.insert_row_to_memtables(log_key, row.values); + } + + // Update from_index in case this is the last iteration: we need to know the next + // index that should be read on later invocations of refresh_indexes. + from_index = index + 1 + } + + // If there are still segments to read, set from_index to zero to read them + // from beginning. Otherwise we leave from_index as the index of the next record to read. + if segnum != to_segnum { + from_index = 0 + } + } + + self.refresh_next_logkey = LogKey::new(to_segnum, from_index); + + Ok(()) + } + + fn insert_row_to_memtables(&mut self, log_key: LogKey, row_values: Vec) { + let pk = row_values[self.primary_key_index].as_indexable().unwrap(); + + for (sk_index, sk_field) in self.config.secondary_keys.iter().enumerate() { + let secondary_memtable = &mut self.secondary_memtables[sk_index]; + let sk_field_index = self + .config + .schema + .iter() + .position(|f| sk_field == f) + .unwrap(); + let sk = row_values[sk_field_index].as_indexable().unwrap(); + + secondary_memtable.set(pk.clone(), sk, log_key.clone()); + } + + // Doing this last because this moves log_key + self.primary_memtable.set(pk, log_key); + } + + fn remove_row_from_memtables(&mut self, row_values: &Vec) { + let pk = row_values[self.primary_key_index].as_indexable().unwrap(); + + if let Some(_) = self.primary_memtable.remove(&pk) { + for (sk_index, sk_field) in self.config.secondary_keys.iter_mut().enumerate() { + let secondary_memtable = &mut self.secondary_memtables[sk_index]; + let sk_field_index = self + .config + .schema + .iter() + .position(|f| sk_field == f) + .unwrap(); + let sk = row_values[sk_field_index].as_indexable().unwrap(); + + secondary_memtable.remove(&pk, &sk); + } + } + } + + pub fn upsert_record(&mut self, record: Row) -> DBResult<()> { + debug!("Opening file in append mode..."); + + if !self.ensure_metadata_file_is_active()? + || !ensure_active_metadata_is_valid( + &self.data_dir_path, + &mut self.active_metadata_file, + )? + { + // The log file has been rotated, so we must try again + return self.upsert_record(record); + } + + self.tx_log.push(TxEntry::Upsert { row: record }); + + if !self.tx_active { + self.commit_transaction()?; + self.tx_log.clear(); + } + + Ok(()) + } + + pub fn batch_find_by_records<'a>( + &mut self, + field: &str, + values: impl Iterator, + params: &QueryParams, + ) -> DBResult> { + let indexables = values + .map(|value| { + value.as_indexable().ok_or(DBError::ValidationError( + "Queried value must be indexable".to_owned(), + )) + }) + .collect::>>()?; + + // Otherwise, continue with querying secondary indexes. + debug!("Finding all records with matching fields"); + + if self.config.read_consistency == ReadConsistency::Strong { + self.refresh_indexes()?; + } + + let log_key_batches = indexables + .into_iter() + .map(|query_key| { + if field == &self.config.primary_key { + let opt = self.primary_memtable.get(&query_key); + let log_keys = match opt { + Some(log_key) => vec![log_key], + None => vec![], + }; + Ok(log_keys) + } else { + let smemtable_index = match get_secondary_memtable_index_by_field( + &self.config.secondary_keys, + field, + ) { + Some(index) => index, + None => { + return Err(DBError::ValidationError( + "Cannot find_by by non-indexed key".to_owned(), + )) + } + }; + + let log_keys = self.secondary_memtables[smemtable_index] + .find_by(&query_key) + .into_iter() + .collect(); + Ok(log_keys) + } + }) + .collect::>>>()?; + + debug!("Found log keys in memtable: {:?}", log_key_batches); + + let mut tagged = vec![]; + for (tag, batch) in log_key_batches.into_iter().enumerate() { + let mapped = batch.into_iter().map(|log_key| (tag, log_key)); + tagged.extend(mapped); + } + + if !params.sort_asc { + tagged.reverse(); + } + let bound_low = params.offset; + let bound_high = (params.offset + params.limit).min(tagged.len()); + let sliced = &tagged[bound_low..bound_high]; + let mut tagged_records = self.read_tagged_log_keys(sliced.into_iter())?; + + debug!("Read {} records", tagged_records.len()); + + if !params.sort_asc { + tagged_records.reverse(); + } + Ok(tagged_records) + } + + /// Read records from segment files based on log keys. + /// The log keys are accompanied by an integer tag that can be used to identify and group them later. + fn read_tagged_log_keys<'a>( + &self, + log_keys: impl Iterator, + ) -> DBResult> { + let mut records = vec![]; + let mut log_keys_map = BTreeMap::new(); + + for (tag, log_key) in log_keys { + if !log_keys_map.contains_key(&log_key.segment_num()) { + log_keys_map.insert(log_key.segment_num(), vec![(tag, log_key.index())]); + } else { + log_keys_map + .get_mut(&log_key.segment_num()) + .unwrap() + .push((tag, log_key.index())); + } + } + + for (segment_num, mut segment_indexes) in log_keys_map { + segment_indexes.sort_unstable(); + + let metadata_path = &self.data_dir_path.join(metadata_filename(segment_num)); + let mut metadata_file = READ_MODE.open(&metadata_path)?; + + let metadata_header = read_metadata_header(&mut metadata_file)?; + + let data_path = &self.data_dir_path.join(metadata_header.uuid.to_string()); + let mut data_file = READ_MODE.open(&data_path)?; + + let header_size = METADATA_FILE_HEADER_SIZE as i64; + let row_length = METADATA_ROW_LENGTH as i64; + let mut current_metadata_offset = header_size; + for (tag, segment_index) in segment_indexes { + let new_metadata_offset = header_size + segment_index as i64 * row_length; + metadata_file.seek_relative(new_metadata_offset - current_metadata_offset)?; + + let mut metadata_buf = [0; METADATA_ROW_LENGTH]; + metadata_file.read_exact(&mut metadata_buf)?; + + let data_offset = u64::from_be_bytes(metadata_buf[0..8].try_into().unwrap()); + let data_length = u64::from_be_bytes(metadata_buf[8..16].try_into().unwrap()); + assert!(data_length > 0); + + data_file.seek(SeekFrom::Start(data_offset))?; + + let mut data_buf = vec![0; data_length as usize]; + data_file.read_exact(&mut data_buf)?; + + let record = Row::deserialize(&data_buf); + records.push((*tag, record)); + + current_metadata_offset = new_metadata_offset + row_length; + } + } + + Ok(records) + } + + pub fn range_by_records>( + &mut self, + field: &str, + range: B, + params: &QueryParams, + ) -> DBResult> { + fn range_bound_to_indexable(bound: Bound<&Value>) -> DBResult> { + match bound { + Bound::Included(value) => value + .as_indexable() + .ok_or(DBError::ValidationError( + "Queried value must be indexable".to_owned(), + )) + .map(Bound::Included), + Bound::Excluded(value) => value + .as_indexable() + .ok_or(DBError::ValidationError( + "Queried value must be indexable".to_owned(), + )) + .map(Bound::Excluded), + Bound::Unbounded => Ok(Bound::Unbounded), + } + } + + let start_indexable = range_bound_to_indexable(range.start_bound())?; + let end_indexable = range_bound_to_indexable(range.end_bound())?; + + let indexable_bounds = OwnedBounds::new(start_indexable, end_indexable); + + if self.config.read_consistency == ReadConsistency::Strong { + self.refresh_indexes()?; + } + + let log_keys = if field == &self.config.primary_key { + self.primary_memtable.range(indexable_bounds) + } else { + let index = get_secondary_memtable_index_by_field(&self.config.secondary_keys, field) + .ok_or_else(|| { + DBError::ValidationError("Cannot range_by by non-indexed key".to_owned()) + })?; + + self.secondary_memtables[index].range(indexable_bounds) + }; + + let mut log_key_batches: Vec<(usize, &LogKey)> = + log_keys.into_iter().map(|log_key| (0, log_key)).collect(); + + if !params.sort_asc { + log_key_batches.reverse(); + } + + let bound_low = params.offset; + let bound_high = (params.offset + params.limit).min(log_key_batches.len()); + let sliced = &log_key_batches[bound_low..bound_high]; + let tagged_records = self.read_tagged_log_keys(sliced.into_iter()); + + let mut result_records: Vec = + tagged_records?.into_iter().map(|(_, rec)| rec).collect(); + + if !params.sort_asc { + result_records.reverse(); + } + + Ok(result_records) + } + + /// Ensures that the `self.metadata_file` and `self.data_file` handles are still pointing to the correct files. + /// If the segment has been rotated, the handle will be closed and reopened. + /// Returns `false` if the file has been rotated and the handle has been reopened, `true` otherwise. + fn ensure_metadata_file_is_active(&mut self) -> DBResult { + let active_target = fs::read_link(&self.data_dir_path.join(ACTIVE_SYMLINK_FILENAME))?; + let active_metadata_path = &self.data_dir_path.join(active_target); + + let correct = is_file_same_as_path(&self.active_metadata_file, &active_metadata_path)?; + if !correct { + debug!("Metadata file has been rotated. Reopening..."); + let metadata_file = APPEND_MODE.open(&active_metadata_path)?; + + let metadata_header = read_metadata_header(&mut self.active_metadata_file)?; + + validate_metadata_header(&metadata_header)?; + + let data_file_path = &self.data_dir_path.join(metadata_header.uuid.to_string()); + + self.active_metadata_file = metadata_file; + self.active_data_file = APPEND_MODE.open(&data_file_path)?; + + return Ok(false); + } else { + return Ok(true); + } + } + + pub fn delete_by_field(&mut self, field: &str, value: &Value) -> DBResult> { + let recs: Vec = self + .batch_find_by_records(field, std::iter::once(value), &DEFAULT_QUERY_PARAMS)? + .into_iter() + .map(|(_, mut rec)| { + rec.tombstone = true; + rec + }) + .collect(); + + // TODO: refactor the clone out of here + for record in &recs { + self.tx_log.push(TxEntry::Delete { + row: record.clone(), + }); + } + + if !self.tx_active { + self.commit_transaction()?; + self.tx_log.clear(); + } + + debug!("Records deleted"); + + Ok(recs) + } + + pub fn commit_transaction(&mut self) -> DBResult<()> { + let active_symlink_path = self.data_dir_path.join(ACTIVE_SYMLINK_FILENAME); + let active_target = fs::read_link(active_symlink_path)?; + let segment_num = parse_segment_number(&active_target)?; + + let initial_data_offset = self.active_data_file.seek(SeekFrom::End(0))?; + let initial_metadata_offset = self.active_metadata_file.seek(SeekFrom::End(0))?; + let mut serialized_data: Vec = vec![]; + let mut serialized_metadata: Vec = Vec::with_capacity(self.tx_log.len() * 16); + let mut pending_memtable_ops: Vec<(LogKey, TxEntry)> = vec![]; + + let mut metadata_buf = [0u8; 16]; + + debug!("Serializing tx_log to byte arrays"); + for tx_entry in &self.tx_log { + let record = match tx_entry { + TxEntry::Upsert { row: record } => record, + TxEntry::Delete { row: record } => record, + }; + + let serialized = record.serialize(); + let record_offset = initial_data_offset + serialized_data.len() as u64; + let record_length = serialized.len() as u64; + assert!(record_length > 0); + + serialized_data.extend(serialized); + + let metadata_pos = initial_metadata_offset + serialized_metadata.len() as u64; + let metadata_index = + (metadata_pos - METADATA_FILE_HEADER_SIZE as u64) / METADATA_ROW_LENGTH as u64; + + // Write the record metadata to the fixed-size metadata buffer + metadata_buf[..8].copy_from_slice(&record_offset.to_be_bytes()); + metadata_buf[8..].copy_from_slice(&record_length.to_be_bytes()); + + serialized_metadata.extend_from_slice(&metadata_buf); + + let log_key = LogKey::new(segment_num, metadata_index); + pending_memtable_ops.push((log_key, tx_entry.clone())); + } + + debug!("Writing serialized bytearrays to log files"); + self.active_data_file.write_all(&serialized_data)?; + self.active_metadata_file.write_all(&serialized_metadata)?; + + // Flush and sync data and metadata to disk + if self.config.write_durability == WriteDurability::Flush { + self.active_data_file.flush()?; + self.active_metadata_file.flush()?; + } else if self.config.write_durability == WriteDurability::FlushSync { + self.active_data_file.flush()?; + self.active_data_file.sync_all()?; + self.active_metadata_file.flush()?; + self.active_metadata_file.sync_all()?; + } + + debug!("Updating memtables"); + for (log_key, tx_entry) in pending_memtable_ops { + match tx_entry { + TxEntry::Upsert { row } => self.insert_row_to_memtables(log_key, row.values), + TxEntry::Delete { row } => self.remove_row_from_memtables(&row.values), + } + } + debug!("Commit done"); + + Ok(()) + } + + pub fn do_maintenance_tasks(&mut self) -> DBResult<()> { + ensure_active_metadata_is_valid(&self.data_dir_path, &mut self.active_metadata_file)?; + + let metadata_size = self.active_metadata_file.seek(SeekFrom::End(0))?; + if metadata_size >= self.config.segment_size as u64 { + self.rotate_and_compact()?; + } + + Ok(()) + } + + fn rotate_and_compact(&mut self) -> DBResult<()> { + debug!("Active log size exceeds threshold, starting rotation and compaction..."); + + let original_data_len = self.active_data_file.seek(SeekFrom::End(0))?; + + let active_target = fs::read_link(&self.data_dir_path.join(ACTIVE_SYMLINK_FILENAME))?; + let active_num = parse_segment_number(&active_target)?; + + debug!("Reading segment data into a BTreeMap"); + let mut pk_to_item_map: BTreeMap<&IndexableValue, &Row> = BTreeMap::new(); + let forward_read_items: Vec<(IndexableValue, Row)> = ForwardLogReader::new( + self.active_metadata_file.try_clone()?, + self.active_data_file.try_clone()?, + ) + .map(|item| { + ( + item.row.values[self.primary_key_index] + .as_indexable() + .expect("Primary key was not indexable"), + item.row, + ) + }) + .collect(); + + for (pk, record) in forward_read_items.iter() { + pk_to_item_map.insert(pk, record); + } + + debug!( + "Read {} records, out of which {} were unique", + forward_read_items.len(), + pk_to_item_map.len() + ); + + // Create a new log data file and write it + debug!("Opening new data file and writing compacted data"); + let (new_data_uuid, new_data_path) = create_segment_data_file(&self.data_dir_path)?; + let mut new_data_file = APPEND_MODE.open(&new_data_path)?; + + let mut pk_to_data_map = BTreeMap::new(); + let mut offset = 0u64; + for (pk, record) in pk_to_item_map.into_iter() { + let serialized = record.serialize(); + let len = serialized.len() as u64; + new_data_file.write_all(&serialized)?; + + pk_to_data_map.insert(pk, (offset, len)); + offset += len; + } + + // Sync the data file to disk. + // This is fine to do without consulting WriteDurability because this is a one-off + // operation that is not part of the normal write path. + new_data_file.flush()?; + new_data_file.sync_all()?; + + let final_data_len = new_data_file.seek(io::SeekFrom::End(0))?; + debug!( + "Wrote compacted data, reduced data size: {} -> {}", + original_data_len, final_data_len + ); + + // Create a new log metadata file and write it + debug!("Opening temp metadata file and writing pointers to compacted data file"); + let temp_metadata_file = tempfile::NamedTempFile::new()?; + let temp_metadata_path = temp_metadata_file.as_ref(); + let mut temp_metadata_file = WRITE_MODE.open(temp_metadata_path)?; + + let metadata_header = MetadataHeader { + version: 1, + uuid: new_data_uuid, + }; + + temp_metadata_file.write_all(&metadata_header.serialize())?; + + let mut metadata_buf = [0u8; 16]; + for (pk, _) in forward_read_items.iter() { + let (offset, len) = pk_to_data_map.get(&pk).unwrap(); + + metadata_buf[..8].copy_from_slice(&offset.to_be_bytes()); + metadata_buf[8..].copy_from_slice(&len.to_be_bytes()); + + temp_metadata_file.write_all(&metadata_buf)?; + } + + // Sync the metadata file to disk, see comment above about sync. + temp_metadata_file.flush()?; + temp_metadata_file.sync_all()?; + + debug!("Moving temporary files to their final locations"); + let new_data_path = &self.data_dir_path.join(new_data_uuid.to_string()); + let active_metadata_path = &self.data_dir_path.join(metadata_filename(active_num)); // overwrite active + + fs::rename(&temp_metadata_path, &active_metadata_path)?; + + debug!("Compaction complete, creating new segment"); + + let new_segment_num = active_num + 1; + let new_metadata_path = self.data_dir_path.join(metadata_filename(new_segment_num)); + let mut new_metadata_file = APPEND_MODE.clone().create(true).open(&new_metadata_path)?; + + let new_metadata_header = MetadataHeader { + version: 1, + uuid: new_data_uuid, + }; + + new_metadata_file.write_all(&new_metadata_header.serialize())?; + + set_active_segment(&self.data_dir_path, new_segment_num)?; + + self.active_metadata_file = APPEND_MODE.open(&new_metadata_path)?; + self.active_data_file = APPEND_MODE.open(&new_data_path)?; + + debug!( + "Active log file {} rotated and compacted, new segment: {}", + active_num, new_segment_num + ); + + Ok(()) + } + + #[inline] + pub fn with_exclusive_lock( + &mut self, + f: impl FnOnce(&mut Self) -> DBResult, + ) -> DBResult { + // No need to acquire a lock if a transaction is already active + // because the lock is already held. + if !self.tx_active { + self.lock_manager.lock_exclusive()?; + } + let result = f(self); + if !self.tx_active { + self.lock_manager.unlock()?; + } + result + } + + #[inline] + pub fn with_shared_lock(&mut self, f: impl FnOnce(&mut Self) -> DBResult) -> DBResult { + // No need to acquire a lock if a transaction is already active + // because the lock is already held. + if !self.tx_active { + self.lock_manager.lock_shared()?; + } + let result = f(self); + if !self.tx_active { + self.lock_manager.unlock()?; + } + + result + } +} + +#[cfg(test)] +mod tests { + use ctor::ctor; + use env_logger; + + use super::*; + + #[ctor] + fn init_logger() { + let _ = env_logger::builder().is_test(true).try_init(); + } + + #[derive(Eq, PartialEq, Clone, Debug)] + enum Field { + Id, + Name, + } + + impl Into for Field { + fn into(self) -> String { + match self { + Field::Id => "id".to_owned(), + Field::Name => "name".to_owned(), + } + } + } + + #[derive(PartialEq, Eq, Debug, Clone)] + struct TestInst2 { + id: i64, + name: String, + } + + impl From for Vec { + fn from(inst: TestInst2) -> Self { + vec![Value::Int(inst.id), Value::String(inst.name)] + } + } + + impl From> for TestInst2 { + fn from(record: Vec) -> Self { + let mut it = record.into_iter(); + TestInst2 { + id: match it.next().unwrap() { + Value::Int(i) => i, + _ => panic!("Expected int"), + }, + name: match it.next().unwrap() { + Value::String(s) => s, + _ => panic!("Expected string"), + }, + } + } + } + + #[test] + fn test_memtable_insert_and_delete() { + let temp_dir = tempfile::tempdir().unwrap(); + let data_dir = temp_dir.path(); + + let capacity = 5; + let segment_size = capacity * 2 * 8 + METADATA_FILE_HEADER_SIZE; + + let mut db = DB::configure() + .data_dir(data_dir.to_str().unwrap()) + .fields(vec![Field::Id, Field::Name]) + .primary_key(Field::Id) + .secondary_keys(vec![Field::Name]) + .segment_size(segment_size) + .initialize() + .expect("Failed to create DB"); + + let engine = &mut db.engine; + + let inst = TestInst2 { + id: 0, + name: "foo".to_owned(), + }; + let id = IndexableValue::Int(0); + + assert_eq!(engine.primary_memtable.get(&id), None); + assert_eq!( + engine.secondary_memtables[0] + .find_by(&IndexableValue::String("foo".to_owned())) + .len(), + 0 + ); + engine.insert_row_to_memtables(LogKey::new(1, 0), inst.clone().into()); + assert_eq!(engine.primary_memtable.get(&id), Some(&LogKey::new(1, 0))); + assert_eq!( + engine.secondary_memtables[0] + .find_by(&IndexableValue::String("foo".to_owned())) + .len(), + 1 + ); + + engine.insert_row_to_memtables(LogKey::new(1, 1), inst.clone().into()); + assert_eq!(engine.primary_memtable.get(&id), Some(&LogKey::new(1, 1))); + assert_eq!( + engine.secondary_memtables[0] + .find_by(&IndexableValue::String("foo".to_owned())) + .len(), + 1 + ); + + engine.remove_row_from_memtables(&inst.into()); + assert_eq!(engine.primary_memtable.get(&id), None); + assert_eq!( + engine.secondary_memtables[0] + .find_by(&IndexableValue::String("foo".to_owned())) + .len(), + 0 + ); + } +} diff --git a/autere_db/src/lib.rs b/autere_db/src/lib.rs new file mode 100644 index 0000000..006056c --- /dev/null +++ b/autere_db/src/lib.rs @@ -0,0 +1,549 @@ +#[macro_use] +extern crate log; + +use once_cell::sync::Lazy; +use rust_decimal::Decimal; +use std::collections::BTreeMap; +use std::fmt::Debug; +use std::fmt::Display; +use std::fs::{self, metadata, File}; +use std::io::{self, Read, Seek, SeekFrom, Write}; +use std::ops::*; +use std::path::{Path, PathBuf}; +use std::thread; +use thiserror::Error; +use uuid::Uuid; + +#[macro_use] +mod common; +mod config; +mod engine; +mod lock; +mod log_reader_forward; +mod memtable_primary; +mod memtable_secondary; +mod record; +mod row; +mod schema; + +pub use common::{DBError, DBResult, OwnedBounds, QueryParams, Value, DEFAULT_QUERY_PARAMS}; +pub use config::{ReadConsistency, WriteDurability}; +pub use record::Record; +pub use schema::Schema; + +use common::*; +use config::*; +use engine::*; +use lock::*; +use log_reader_forward::*; +use memtable_primary::PrimaryMemtable; +use memtable_secondary::SecondaryMemtable; +use row::*; + +pub struct DB { + engine: Engine, +} + +impl DB { + /// Create a new database configuration builder. + pub fn configure() -> ConfigBuilder { + ConfigBuilder::new() + } + + fn initialize(config: Config) -> DBResult { + let engine = Engine::initialize(config)?; + Ok(DB { engine }) + } + + /// Insert a record into the database. If the primary key value already exists, + /// the existing record will be replaced by the supplied one. + pub fn upsert(&mut self, record: impl Into) -> DBResult<()> { + let row = Row { + values: record.into().into(), + tombstone: false, + }; + debug!("Upserting record: {:?}", row); + + self.engine + .with_exclusive_lock(move |engine| engine.upsert_record(row))?; + + Ok(()) + } + + /// Get a record by its primary index value. + /// E.g. `db.get(Value::Int(10))`. + pub fn get(&mut self, value: &Value) -> DBResult> { + let tagged_rows = self.engine.with_shared_lock(|engine| { + engine.batch_find_by_records( + // TODO: This clone is only here to appease the borrow checker + &engine.config.primary_key.clone(), + std::iter::once(value), + &DEFAULT_QUERY_PARAMS, + ) + })?; + + assert!(tagged_rows.len() <= 1); + + Ok(tagged_rows + .into_iter() + .next() + .map(|(_, row)| Record::from(row))) + } + + /// Get a collection of records based on an indexed field value. + pub fn find_by(&mut self, field: impl AsRef, value: &Value) -> DBResult> { + let tagged_rows = self.engine.with_shared_lock(|engine| { + engine.batch_find_by_records( + field.as_ref(), + std::iter::once(value), + &DEFAULT_QUERY_PARAMS, + ) + })?; + + Ok(tagged_rows + .into_iter() + .map(|(_, row)| Record::from(row)) + .collect()) + } + + /// Get a collection of records based on an indexed field value, with additional parameters. + pub fn find_by_with_params( + &mut self, + field: impl AsRef, + value: &Value, + params: &QueryParams, + ) -> DBResult> { + let recs = self.engine.with_shared_lock(|engine| { + engine.batch_find_by_records(field.as_ref(), std::iter::once(value), params) + })?; + + Ok(recs.into_iter().map(|(_, row)| Record::from(row)).collect()) + } + + /// Get a collection of records based on a sequence of indexed field values. + /// Returns a vector of pairs where the first value is an index into the given sequence of values, + /// and the second value is the record. + pub fn batch_find_by( + &mut self, + field: impl Into, + values: &[Value], + ) -> DBResult> { + let recs = self.engine.with_shared_lock(|engine| { + engine.batch_find_by_records(&field.into(), values.iter(), &DEFAULT_QUERY_PARAMS) + })?; + + Ok(recs + .into_iter() + .map(|(tag, row)| (tag, Record::from(row))) + .collect()) + } + + /// Get a collection of records based on a sequence of indexed field values, with additional parameters. + /// Returns a vector of pairs where the first value is an index into the given sequence of values, + /// and the second value is the record. + pub fn batch_find_by_with_params( + &mut self, + field: impl AsRef, + values: &[Value], + params: &QueryParams, + ) -> DBResult> { + let recs = self.engine.with_shared_lock(|engine| { + engine.batch_find_by_records(field.as_ref(), values.iter(), params) + })?; + + Ok(recs + .into_iter() + .map(|(tag, row)| (tag, Record::from(row))) + .collect()) + } + + /// Get a collection of records based on a range of indexed field values. + /// This method can be used to run comparison-like queries, e.g. `field >= 10` + /// could be expressed as `db.range_by(Field::Id, 10..)`. + pub fn range_by>( + &mut self, + field: impl AsRef, + range: B, + ) -> DBResult> { + let recs = self.engine.with_shared_lock(|engine| { + engine.range_by_records(field.as_ref(), range, &DEFAULT_QUERY_PARAMS) + })?; + + Ok(recs.into_iter().map(|row| Record::from(row)).collect()) + } + + /// Get a collection of records based on a range of indexed field values, with additional parameters. + /// This method can be used to run comparison-like queries, e.g. `field >= 10` + /// could be expressed as `db.range_by(Field::Id, 10..)`. + pub fn range_by_with_params>( + &mut self, + field: impl AsRef, + range: B, + params: &QueryParams, + ) -> DBResult> { + let recs = self + .engine + .with_shared_lock(|engine| engine.range_by_records(field.as_ref(), range, params))?; + + Ok(recs.into_iter().map(|row| Record::from(row)).collect()) + } + + /// Delete records by a field value. + /// E.g. `db.delete_by(Field::Name, "John")`, assuming `Field` is the DB field type and `Field::Name` is secondary indexed. + /// Returns a vector of deleted records. If no records were deleted, the vector will be empty. + /// + /// Deletion is done by marking the record as a tombstone. The record will still be present in the log file, + /// but will be ignored by reads. Upon compaction, tombstoned records will be removed. + pub fn delete_by(&mut self, field: impl AsRef, value: &Value) -> DBResult> { + let recs = self + .engine + .with_exclusive_lock(|engine| engine.delete_by_field(field.as_ref(), value))?; + + Ok(recs + .into_iter() + .map(|row| Record::from(row.values)) + .collect()) + } + + /// Delete record by primary key. + pub fn delete(&mut self, pk: &Value) -> DBResult> { + let recs = self.engine.with_exclusive_lock(|engine| { + engine + // TODO: This clone is only here to appease the borrow checker + .delete_by_field(&engine.config.primary_key.clone(), pk) + })?; + + assert!(recs.len() <= 1); + + Ok(recs.into_iter().next().map(|row| Record::from(row.values))) + } + + /// Check if there are any pending tasks and do them. Tasks include: + /// - Rotating the active log file if it has reached capacity and compacting it. + /// + /// This function should be called periodically to ensure that the database remains in an optimal state. + /// Note that this function is synchronous and may block for a relatively long time. + /// You may call this function in a separate thread or process to avoid blocking the main thread. + /// However, the database will be exclusively locked, so all writes and reads will be blocked during the tasks. + pub fn do_maintenance_tasks(&mut self) -> DBResult<()> { + self.engine + .with_exclusive_lock(|engine| engine.do_maintenance_tasks()) + } + + /// Refresh the in-memory indexes from the log files. + /// This needs to only be called if the read consistency is set to `ReadConsistency::Eventual`. + pub fn refresh_indexes(&mut self) -> DBResult<()> { + self.engine + .with_exclusive_lock(|engine| engine.refresh_indexes()) + } + + /// Begin a transaction. This will acquire an exclusive lock on the database, + /// preventing other clients from using the database until the transaction is committed or rolled back. + pub fn tx_begin(&mut self) -> DBResult<()> { + if self.engine.tx_active { + return Err(DBError::TransactionError( + "Transaction already active".to_string(), + )); + } + + self.engine.lock_manager.lock_exclusive()?; + self.engine.tx_active = true; + Ok(()) + } + + /// Commit the active transaction. A transaction must be active, otherwise + /// a `DBError::TransactionError` will be returned. + pub fn tx_commit(&mut self) -> DBResult<()> { + if !self.engine.tx_active { + return Err(DBError::TransactionError( + "No active transaction to commit".to_string(), + )); + } + + self.engine.commit_transaction()?; + self.engine.tx_log.clear(); + self.engine.tx_active = false; + self.engine.lock_manager.unlock()?; + Ok(()) + } + + /// Rollback the active transaction. A transaction must be active, otherwise + /// a `DBError::TransactionError` will be returned. + pub fn tx_rollback(&mut self) -> DBResult<()> { + if !self.engine.tx_active { + return Err(DBError::TransactionError( + "No active transaction to roll back".to_string(), + )); + } + + self.engine.tx_log.clear(); + self.engine.tx_active = false; + self.engine.lock_manager.unlock()?; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use ctor::ctor; + use env_logger; + + use super::*; + + #[ctor] + fn init_logger() { + let _ = env_logger::builder().is_test(true).try_init(); + } + + #[derive(Eq, PartialEq, Clone, Debug)] + enum Field { + Id, + Name, + } + + impl Into for Field { + fn into(self) -> String { + match self { + Field::Id => "id".to_string(), + Field::Name => "name".to_string(), + } + } + } + + struct TestInst1 { + id: i64, + } + + impl From for Record { + fn from(inst: TestInst1) -> Self { + vec![Value::Int(inst.id)].into() + } + } + + impl From for TestInst1 { + fn from(record: Record) -> Self { + let mut it = record.into_iter(); + TestInst1 { + id: match it.next().unwrap() { + Value::Int(i) => i, + _ => panic!("Expected int"), + }, + } + } + } + + struct TestInst2 { + id: i64, + name: String, + } + + impl From for Record { + fn from(inst: TestInst2) -> Self { + vec![Value::Int(inst.id), Value::String(inst.name)].into() + } + } + + impl From for TestInst2 { + fn from(record: Record) -> Self { + let mut it = record.into_iter(); + TestInst2 { + id: match it.next().unwrap() { + Value::Int(i) => i, + _ => panic!("Expected int"), + }, + name: match it.next().unwrap() { + Value::String(s) => s, + _ => panic!("Expected string"), + }, + } + } + } + + #[test] + fn test_compaction() { + let temp_dir = tempfile::tempdir().unwrap(); + let data_dir = temp_dir.path(); + + let capacity = 5; + let segment_size = capacity * 2 * 8 + METADATA_FILE_HEADER_SIZE; + + let mut db = DB::configure() + .data_dir(data_dir.to_str().unwrap()) + .fields(vec![Field::Id]) + .primary_key(Field::Id) + .segment_size(segment_size) + .initialize() + .expect("Failed to create DB"); + + // Insert records with same value until we reach the capacity + for _ in 0..capacity { + db.upsert(TestInst1 { id: 0 }) + .expect("Failed to insert record"); + } + + let mut segment1_file = READ_MODE.open(data_dir.join(metadata_filename(1))).unwrap(); + let segment1_metadata_size_original = segment1_file.seek(io::SeekFrom::End(0)).unwrap(); + + let segment1_header = read_metadata_header(&mut segment1_file).unwrap(); + let mut segment1_data_file = READ_MODE + .open(data_dir.join(segment1_header.uuid.to_string())) + .unwrap(); + let segment1_data_size_original = segment1_data_file.seek(io::SeekFrom::End(0)).unwrap(); + + // Rotate and compact + db.do_maintenance_tasks() + .expect("Failed to do maintenance tasks"); + + // Insert one extra with different value, this goes into another segment + db.upsert(TestInst1 { id: 1 }) + .expect("Failed to insert record"); + + // Check that rotation resulted in 2 segments + assert!(fs::exists(data_dir.join(metadata_filename(1))).unwrap()); + assert!(fs::exists(data_dir.join(metadata_filename(2))).unwrap()); + // Note negation here + assert!(!fs::exists(data_dir.join(metadata_filename(3))).unwrap()); + + // Check that the compacted metadata file has the same size + let mut segment1_metadata_file_compacted = + READ_MODE.open(data_dir.join(metadata_filename(1))).unwrap(); + let segment1_metadata_size_compacted = segment1_metadata_file_compacted + .seek(io::SeekFrom::End(0)) + .unwrap(); + assert_eq!( + segment1_metadata_size_compacted, + segment1_metadata_size_original + ); + + // Check that the compacted data file is smaller + let segment1_header_compacted = + read_metadata_header(&mut segment1_metadata_file_compacted).unwrap(); + let mut segment1_data_file_compacted = READ_MODE + .open(data_dir.join(segment1_header_compacted.uuid.to_string())) + .unwrap(); + let segment1_data_size_compacted = segment1_data_file_compacted + .seek(io::SeekFrom::End(0)) + .unwrap(); + assert!( + segment1_data_size_compacted < segment1_data_size_original, + "Original: {}, Compacted: {}", + segment1_data_size_original, + segment1_data_size_compacted + ); + + // Check that the records can be read + let inst0: TestInst1 = db + .get(&Value::Int(0 as i64)) + .expect("Failed to get record") + .expect("Record not found") + .into(); + + assert!(inst0.id == 0); + + let inst1: TestInst1 = db + .get(&Value::Int(1 as i64)) + .expect("Failed to get record") + .expect("Record not found") + .into(); + + assert!(inst1.id == 1); + } + + #[test] + fn test_repair() { + let temp_dir = tempfile::tempdir().unwrap(); + let data_dir = temp_dir.path(); + + let mut db = DB::configure() + .data_dir(data_dir.to_str().unwrap()) + .fields(vec![Field::Id]) + .primary_key(Field::Id) + .initialize() + .expect("Failed to create DB"); + + // Insert records + let n_recs: u64 = 100; + for i in 0..n_recs { + db.upsert(TestInst1 { id: i as i64 }) + .expect("Failed to insert record"); + } + + // Open the segment file and write garbage to it to simulate corruption + let segment_metadata_path = data_dir.join(metadata_filename(1)); + let mut file = APPEND_MODE + .open(&segment_metadata_path) + .expect("Failed to open file"); + + file.write_all(&[1, 0, 0, 0]) // A partially written integer value ([1] + some bytes) + .expect("Failed to write garbage"); + file.flush().unwrap(); + + let len = file.seek(SeekFrom::End(0)).expect("Failed to seek"); + assert_ne!(len, METADATA_FILE_HEADER_SIZE as u64 + n_recs * 16); + + // Try to refresh indexes, reading the file from beginning to end: should lead to error + db.refresh_indexes() + .expect_err("refresh_indexes should fail because of partial write"); + + // Trigger autorepair + db.do_maintenance_tasks() + .expect("Failed to run maintenance tasks"); + + // Try to refresh indexes, reading the file from beginning to end: should work now + db.refresh_indexes() + .expect("refresh_indexes should succeed"); + + // Reopen file and check that it has the correct size + let mut file = READ_MODE + .open(&segment_metadata_path) + .expect("Failed to open file"); + let len = file.seek(SeekFrom::End(0)).expect("Failed to seek"); + assert_eq!(len, METADATA_FILE_HEADER_SIZE as u64 + n_recs * 16); + } + + #[test] + fn test_memtables_updated_on_write() { + let temp_dir = tempfile::tempdir().unwrap(); + let data_dir = temp_dir.path(); + + let mut db = DB::configure() + .data_dir(data_dir.to_str().unwrap()) + .fields(vec![Field::Id, Field::Name]) + .primary_key(Field::Id) + .secondary_keys(vec![Field::Name]) + .initialize() + .expect("Failed to create DB"); + + // Check that the key is not indexed before write + assert_eq!( + db.engine.primary_memtable.get(&IndexableValue::Int(0)), + None + ); + assert_eq!( + db.engine.secondary_memtables[0] + .find_by(&IndexableValue::String("John".to_string())) + .len(), + 0 + ); + + // Insert record + db.upsert(TestInst2 { + id: 0, + name: "John".to_owned(), + }) + .expect("Failed to insert record"); + + // Check that the key is now indexed + let expected_log_key = LogKey::new(1, 0); + let expected_pk = IndexableValue::Int(0); + assert_eq!( + db.engine.primary_memtable.get(&expected_pk), + Some(&expected_log_key) + ); + let expected_vals = vec![&expected_log_key]; + let actual_vals = db.engine.secondary_memtables[0] + .find_by(&IndexableValue::String("John".to_string())) + .collect::>(); + assert_eq!(actual_vals, expected_vals); + } +} diff --git a/autere_db/src/lock.rs b/autere_db/src/lock.rs new file mode 100644 index 0000000..f00429f --- /dev/null +++ b/autere_db/src/lock.rs @@ -0,0 +1,117 @@ +use super::*; + +pub struct LockManager { + lock_file: fs::File, + excl_lock_file: fs::File, + + state: LockState, +} + +#[derive(Debug, PartialEq, Eq)] +enum LockState { + NotLocked, + Shared, + Exclusive, +} + +impl LockManager { + pub fn new(data_dir_path: PathBuf) -> DBResult { + let lock_file = fs::File::create(data_dir_path.join(LOCK_FILENAME))?; + let excl_lock_file = fs::File::create(data_dir_path.join(EXCL_LOCK_REQ_FILENAME))?; + + Ok(LockManager { + lock_file, + excl_lock_file, + state: LockState::NotLocked, + }) + } + + fn is_exclusive_lock_requested(&self) -> DBResult { + // Attempt to acquire a shared lock on the lock request file + // If the file is already locked, return false + match fs2::FileExt::try_lock_shared(&self.excl_lock_file) { + Err(e) => { + if e.kind() == fs2::lock_contended_error().kind() { + return Ok(true); + } + return Err(DBError::IOError(e)); + } + + Ok(_) => { + fs2::FileExt::unlock(&self.excl_lock_file)?; + return Ok(false); + } + } + } + + pub fn lock_shared(&mut self) -> DBResult<()> { + if self.state == LockState::Shared { + return Err(DBError::LockRequestError( + "Already holding a shared lock".to_owned(), + )); + } else if self.state == LockState::Exclusive { + return Err(DBError::LockRequestError( + "Cannot acquire shared lock while holding an exclusive lock".to_owned(), + )); + } + + let mut timeout = 5; + loop { + if self.is_exclusive_lock_requested()? { + debug!( + "Exclusive lock requested, waiting for {}ms before requesting a shared lock again", + timeout + ); + thread::sleep(std::time::Duration::from_millis(timeout)); + timeout *= 2; + + if timeout > LOCK_WAIT_MAX_MS { + return Err(DBError::LockRequestError( + "Acquisition of shared lock timed out after {LOCK_WAIT_MAX_MS}".to_owned(), + )); + } + } else { + fs2::FileExt::lock_shared(&self.lock_file)?; + self.state = LockState::Shared; + return Ok(()); + } + } + } + + pub fn lock_exclusive(&mut self) -> DBResult<()> { + if self.state == LockState::Exclusive { + return Err(DBError::LockRequestError( + "Already holding an exclusive lock".to_owned(), + )); + } else if self.state == LockState::Shared { + return Err(DBError::LockRequestError( + "Cannot acquire exclusive lock while holding a shared lock".to_owned(), + )); + } + + // Create a lock on the exclusive lock request file to signal to readers that they should wait + // This will block until the lock is acquired + fs2::FileExt::lock_exclusive(&self.excl_lock_file)?; + + // Acquire an exclusive lock on the actual lock files + fs2::FileExt::lock_exclusive(&self.lock_file)?; + self.state = LockState::Exclusive; + + // Unlock the request file + fs2::FileExt::unlock(&self.excl_lock_file)?; + + Ok(()) + } + + pub fn unlock(&mut self) -> DBResult<()> { + if self.state == LockState::NotLocked { + return Err(DBError::LockRequestError( + "Not holding any locks".to_owned(), + )); + } + + fs2::FileExt::unlock(&self.lock_file)?; + self.state = LockState::NotLocked; + Ok(()) + } +} diff --git a/autere_db/src/log_reader_forward.rs b/autere_db/src/log_reader_forward.rs new file mode 100644 index 0000000..f3fc16c --- /dev/null +++ b/autere_db/src/log_reader_forward.rs @@ -0,0 +1,134 @@ +use super::*; + +pub struct ForwardLogReader { + metadata_reader: io::BufReader, + data_reader: io::BufReader, +} + +pub struct ForwardLogReaderItem { + pub row: Row, + pub index: u64, +} + +impl ForwardLogReader { + pub fn new(metadata_file: fs::File, data_file: fs::File) -> ForwardLogReader { + let mut ret = ForwardLogReader { + metadata_reader: io::BufReader::new(metadata_file), + data_reader: io::BufReader::new(data_file), + }; + + ret.metadata_reader + .seek(io::SeekFrom::Start(METADATA_FILE_HEADER_SIZE as u64)) + .expect("Seek failed"); + + ret + } + + pub fn new_with_index( + metadata_file: fs::File, + data_file: fs::File, + index: u64, + ) -> ForwardLogReader { + let mut ret = ForwardLogReader { + metadata_reader: io::BufReader::new(metadata_file), + data_reader: io::BufReader::new(data_file), + }; + + ret.metadata_reader + .seek(io::SeekFrom::Start( + METADATA_FILE_HEADER_SIZE as u64 + METADATA_ROW_LENGTH as u64 * index, + )) + .expect("Seek failed"); + + ret + } + + fn read_record(&mut self) -> Result, io::Error> { + loop { + let pos = self.metadata_reader.stream_position()?; + let index = (pos - METADATA_FILE_HEADER_SIZE as u64) / METADATA_ROW_LENGTH as u64; + + let mut metadata_entry_buf = vec![0; 16]; // 2x u64 + if let Err(e) = self.metadata_reader.read_exact(&mut metadata_entry_buf) { + if e.kind() == io::ErrorKind::UnexpectedEof { + return Ok(None); + } else { + return Err(e); + } + } + + // First u64 is the offset of the record in the data file, second is the length of the record + let entry_offset = u64::from_be_bytes(metadata_entry_buf[0..8].try_into().unwrap()); + let entry_length = u64::from_be_bytes(metadata_entry_buf[8..16].try_into().unwrap()); + + if entry_offset == 0 && entry_length == 0 { + // This is an unused entry in the metadata file, skip + continue; + } + + // Use .seek_relative instead of .seek to avoid dropping the BufReader internal buffer when + // the seek distance is small + let seek_distance = entry_offset as i64 - self.data_reader.stream_position()? as i64; + self.data_reader.seek_relative(seek_distance)?; + + let mut result_buf = vec![0; entry_length as usize]; + self.data_reader.read_exact(&mut result_buf)?; + + let row = Row::deserialize(&result_buf); + return Ok(Some(ForwardLogReaderItem { row, index })); + } + } +} + +impl Iterator for ForwardLogReader { + type Item = ForwardLogReaderItem; + + fn next(&mut self) -> Option { + self.read_record().unwrap_or_else(|err| { + panic!("Error reading record: {:?}", err); + }) + } +} + +#[cfg(test)] +mod tests { + use ctor::ctor; + use env_logger; + + use super::*; + + #[ctor] + fn init_logger() { + let _ = env_logger::builder().is_test(true).try_init(); + } + + const TEST_RESOURCES_DIR: &str = "tests/resources"; + + #[test] + fn test_forward_log_reader_fixture_db1() { + let metadata_path = Path::new(TEST_RESOURCES_DIR).join("test_metadata_1"); + let data_path = Path::new(TEST_RESOURCES_DIR).join("test_data_1"); + let metadata_file = fs::OpenOptions::new() + .read(true) + .open(&metadata_path) + .expect("Failed to open metadata file"); + let data_file = fs::OpenOptions::new() + .read(true) + .open(&data_path) + .expect("Failed to open data file"); + + let mut forward_log_reader = ForwardLogReader::new(metadata_file, data_file); + + // There are two records in the log with "schema" with one field: Bytes + + let ForwardLogReaderItem { row, index: _ } = forward_log_reader + .next() + .expect("Failed to read the first record"); + assert!(match &row.values[..] { + [Value::Bytes(bytes)] => bytes.len() == 256, + _ => false, + }); + + assert!(forward_log_reader.next().is_none()); + } +} diff --git a/autere_db/src/memtable_primary.rs b/autere_db/src/memtable_primary.rs new file mode 100644 index 0000000..573592e --- /dev/null +++ b/autere_db/src/memtable_primary.rs @@ -0,0 +1,40 @@ +use super::*; +use std::collections::BTreeMap; + +pub struct PrimaryMemtable { + /// Map of records indexed by key. Used as a shared heap of records + /// for all secondary memtables also. Secondary memtables store an + /// IndexableValue as their record value, which is used to get + /// the actual record from the primary memtable `records` map. + /// + /// Note: it must be invariant that all memtables (primary and secondary) + /// contain the same keys. + records: BTreeMap, +} + +impl PrimaryMemtable { + pub fn new() -> PrimaryMemtable { + PrimaryMemtable { + records: BTreeMap::new(), + } + } + + pub fn set(&mut self, key: IndexableValue, value: LogKey) { + self.records.insert(key, value); + } + + pub fn get(&self, key: &IndexableValue) -> Option<&LogKey> { + self.records.get(key) + } + + pub fn remove(&mut self, key: &IndexableValue) -> Option { + self.records.remove(key) + } + + pub fn range>(&self, range: B) -> Vec<&LogKey> { + self.records + .range(range) + .map(|(_, log_key)| log_key) + .collect() + } +} diff --git a/autere_db/src/memtable_secondary.rs b/autere_db/src/memtable_secondary.rs new file mode 100644 index 0000000..e1ca835 --- /dev/null +++ b/autere_db/src/memtable_secondary.rs @@ -0,0 +1,66 @@ +use once_cell::sync::Lazy; + +use super::*; +use std::collections::{btree_map::Values, BTreeMap}; + +pub struct SecondaryMemtable { + /// A 2-layer map of records indexed by SK => PK => LogKey. + /// The PK information is required to tell two records apart. + records: BTreeMap, +} + +static EMPTY_MAP: Lazy> = Lazy::new(|| BTreeMap::new()); + +impl SecondaryMemtable { + pub fn new() -> SecondaryMemtable { + SecondaryMemtable { + records: BTreeMap::new(), + } + } + + pub fn set(&mut self, pk: IndexableValue, sk: IndexableValue, value: LogKey) { + match self.records.get_mut(&sk) { + Some(map) => { + map.insert(pk, value); + } + None => { + self.records + .insert(sk, LogKeyMap::new_with_initial(pk, value)); + } + }; + } + + pub fn find_by(&self, key: &IndexableValue) -> Values { + match self.records.get(key) { + Some(set) => set.log_keys(), + None => EMPTY_MAP.values(), + } + } + + // Remove a single mapping associated with the given PK and SK. Returns `true` + // if the log key existed and was removed, `false` otherwise. + pub fn remove(&mut self, pk: &IndexableValue, sk: &IndexableValue) -> bool { + let map = match self.records.get_mut(sk) { + Some(set) => set, + None => return false, + }; + if map.len() == 1 && map.contains_pk(pk) { + self.records.remove(sk); + true + } else { + return match map.remove_pk(pk) { + Ok(_) => true, + Err(LogKeyMapError::NotFoundError) => false, + Err(e) => panic!("{:?}", e), + }; + } + } + + pub fn range>(&self, range: B) -> Vec<&LogKey> { + let mut keys = Vec::new(); + for (_, map) in self.records.range(range) { + keys.extend(map.log_keys()); + } + keys + } +} diff --git a/autere_db/src/record.rs b/autere_db/src/record.rs new file mode 100644 index 0000000..b349853 --- /dev/null +++ b/autere_db/src/record.rs @@ -0,0 +1,38 @@ +use super::*; + +pub struct Record { + values: Vec, +} + +impl Record { + pub fn values(&self) -> &[Value] { + &self.values + } +} + +impl IntoIterator for Record { + type Item = Value; + type IntoIter = std::vec::IntoIter; + + fn into_iter(self) -> Self::IntoIter { + self.values.into_iter() + } +} + +impl From> for Record { + fn from(values: Vec) -> Self { + Record { values } + } +} + +impl From for Vec { + fn from(record: Record) -> Self { + record.values + } +} + +impl From for Record { + fn from(row: Row) -> Self { + Record { values: row.values } + } +} diff --git a/autere_db/src/row.rs b/autere_db/src/row.rs new file mode 100644 index 0000000..d8140d2 --- /dev/null +++ b/autere_db/src/row.rs @@ -0,0 +1,70 @@ +use super::*; + +#[derive(Debug, Clone)] +pub struct Row { + pub values: Vec, + pub tombstone: bool, +} + +impl Row { + pub fn serialize(&self) -> Vec { + let mut bytes = Vec::new(); + + if self.tombstone { + bytes.extend(&[B_TOMBSTONE]); + } else { + bytes.extend(&[B_LIVE]); + } + + for value in &self.values { + bytes.extend(value.serialize()); + } + bytes + } + + pub fn deserialize(bytes: &[u8]) -> Row { + assert!(bytes.len() > 0); + + let mut values = Vec::new(); + + let tombstone = bytes[0] == B_TOMBSTONE; + + let mut start = 1; + while start < bytes.len() { + let (rv, consumed) = Value::deserialize(&bytes[start..]); + values.push(rv); + start += consumed; + } + Row { values, tombstone } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_record_serialize_deserialize() { + let record = Row { + values: vec![ + Value::Int(1), + Value::String("hello".to_string()), + Value::Bytes(vec![0, 1, 2, 3]), + ], + tombstone: true, + }; + + let serialized = record.serialize(); + let deserialized = Row::deserialize(&serialized); + let reserialized = deserialized.serialize(); + + assert_eq!(serialized.len(), reserialized.len()); + assert_eq!(record.values, deserialized.values); + } +} + +#[derive(Clone, Debug)] +pub enum TxEntry { + Upsert { row: Row }, + Delete { row: Row }, +} diff --git a/autere_db/src/schema.rs b/autere_db/src/schema.rs new file mode 100644 index 0000000..98d0df7 --- /dev/null +++ b/autere_db/src/schema.rs @@ -0,0 +1,7 @@ +use super::*; + +pub struct Schema { + pub fields: Vec, + pub primary_key: String, + pub secondary_keys: Vec, +} diff --git a/autere_db/tests/integration.rs b/autere_db/tests/integration.rs new file mode 100644 index 0000000..2150ea1 --- /dev/null +++ b/autere_db/tests/integration.rs @@ -0,0 +1,1012 @@ +#[macro_use] +extern crate log; +extern crate ctor; +extern crate tempfile; + +use autere_db::*; +use ctor::ctor; +use env_logger; +use serial_test::serial; +use std::fs::{self}; +use std::path::Path; +use std::thread; +use std::time::Duration; +use tempfile::tempdir; + +pub fn tmp_dir() -> String { + let dir = tempdir() + .expect("Failed to create temporary directory") + .path() + .to_str() + .expect("Failed to convert temporary directory path to string") + .to_string(); + fs::create_dir_all(&dir).expect("Failed to create temporary directory"); + dir +} + +#[ctor] +fn init_logger() { + let _ = env_logger::builder().is_test(true).try_init(); +} + +#[derive(Eq, PartialEq, Clone, Debug)] +enum Field { + Id, + Name, + Data, +} + +impl AsRef for Field { + fn as_ref(&self) -> &str { + match self { + Field::Id => "id", + Field::Name => "name", + Field::Data => "data", + } + } +} + +impl From for String { + fn from(field: Field) -> String { + field.as_ref().to_owned() + } +} + +#[derive(Debug)] +struct Inst { + pub id: i64, + pub name: Option, + pub data: Vec, +} + +impl From for Record { + fn from(inst: Inst) -> Record { + vec![ + Value::Int(inst.id), + match inst.name { + Some(name) => Value::String(name), + None => Value::Null, + }, + Value::Bytes(inst.data), + ] + .into() + } +} + +impl From for Inst { + fn from(record: Record) -> Self { + let mut it = record.into_iter(); + Inst { + id: match it.next().unwrap() { + Value::Int(id) => id, + other => panic!("Invalid value type: {:?}", other), + }, + name: match it.next().unwrap() { + Value::String(name) => Some(name), + Value::Null => None, + other => panic!("Invalid value type: {:?}", other), + }, + data: match it.next().unwrap() { + Value::Bytes(data) => data, + other => panic!("Invalid value type: {:?}", other), + }, + } + } +} + +impl Inst { + fn schema() -> Vec { + vec![Field::Id, Field::Name, Field::Data] + } + fn primary_key() -> Field { + Field::Id + } + fn secondary_keys() -> Vec { + vec![Field::Name] + } +} + +#[test] +fn test_initialize_only() { + let data_dir = tmp_dir(); + let _db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); +} + +#[test] +fn test_upsert_and_get_with_primary_memtable() { + let data_dir = tmp_dir(); + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + let id = 1; + let inst = Inst { + id, + name: Some("Alice".to_string()), + data: vec![0, 1, 2], + }; + db.upsert(inst).unwrap(); + + let result: Inst = db.get(&Value::Int(1)).unwrap().unwrap().into(); + + // Check that the IDs match + assert!(result.id == id); +} + +#[test] +fn test_upsert_and_get() { + let data_dir = tmp_dir(); + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + // Insert some records + db.upsert(Inst { + id: 0, + name: None, + data: vec![3, 4, 5], + }) + .unwrap(); + + db.upsert(Inst { + id: 1, + name: Some("Alice".to_string()), + data: vec![0, 1, 2], + }) + .unwrap(); + + db.upsert(Inst { + id: 1, + name: Some("Bob".to_string()), + data: vec![0, 1, 2], + }) + .unwrap(); + + db.upsert(Inst { + id: 2, + name: Some("George".to_string()), + data: vec![], + }) + .unwrap(); + + // Get with ID = 0 + let result: Inst = db.get(&Value::Int(0)).unwrap().unwrap().into(); + + // Should match id == 0 + assert!(result.id == 0); + assert!(result.name == None); + + // Get with ID = 1 + let result: Inst = db.get(&Value::Int(1)).unwrap().unwrap().into(); + + // Should match newest inst with id == 1 + assert!(result.id == 1); + assert!(result.name == Some("Bob".to_owned())); +} + +#[test] +fn test_get_nonexistant() { + let data_dir = tmp_dir(); + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + let result = db.get(&Value::Int(0)).unwrap(); + assert!(result.is_none()); +} + +#[test] +fn test_upsert_and_find_by() { + let data_dir = tmp_dir(); + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + // Insert some records + db.upsert(Inst { + id: 0, + name: Some("John".to_string()), + data: vec![3, 4, 5], + }) + .unwrap(); + + db.upsert(Inst { + id: 1, + name: Some("John".to_string()), + data: vec![1, 2, 3], + }) + .unwrap(); + + db.upsert(Inst { + id: 2, + name: Some("George".to_string()), + data: vec![1, 2, 3], + }) + .unwrap(); + + // There should be 2 Johns + let johns = db + .find_by(&Field::Name, &Value::String("John".to_string())) + .expect("Failed to find all Johns"); + + assert_eq!(johns.len(), 2); +} + +struct InstSingleId { + pub id: i64, +} + +impl From for Record { + fn from(inst: InstSingleId) -> Record { + vec![Value::Int(inst.id)].into() + } +} + +impl From for InstSingleId { + fn from(record: Record) -> Self { + let mut it = record.into_iter(); + InstSingleId { + id: match it.next().unwrap() { + Value::Int(id) => id, + other => panic!("Invalid value type: {:?}", other), + }, + } + } +} + +impl InstSingleId { + fn schema() -> Vec { + vec![Field::Id] + } + fn primary_key() -> Field { + Field::Id + } + fn secondary_keys() -> Vec { + vec![] + } +} + +#[test] +#[serial] +fn test_multiple_writing_threads() { + let data_dir = tmp_dir(); + debug!("Data dir: {:?}", data_dir); + let mut threads = vec![]; + let threads_n = 100; + + for i in 0..threads_n { + let data_dir = data_dir.clone(); + threads.push(thread::spawn(move || { + let mut db = DB::configure() + .fields(InstSingleId::schema()) + .primary_key(InstSingleId::primary_key()) + .secondary_keys(InstSingleId::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + db.upsert(InstSingleId { id: i }) + .expect("Failed to upsert record"); + })); + } + + for thread in threads { + thread.join().expect("Failed to join thread"); + } + + // Read the records + let mut db = DB::configure() + .fields(InstSingleId::schema()) + .primary_key(InstSingleId::primary_key()) + .secondary_keys(InstSingleId::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + for i in 0..threads_n { + let result: InstSingleId = db + .get(&Value::Int(i)) + .expect("Failed to get record") + .expect("Record not found") + .into(); + + assert!(result.id == i); + } +} + +#[test] +#[serial] +fn test_one_writer_and_multiple_reading_threads() { + let data_dir = tmp_dir(); + let mut threads = vec![]; + let threads_n = 100; + + // Add readers that poll for the records + for i in 0..threads_n { + let data_dir = data_dir.clone(); + threads.push(thread::spawn(move || { + let mut db = DB::configure() + .fields(InstSingleId::schema()) + .primary_key(InstSingleId::primary_key()) + .secondary_keys(InstSingleId::secondary_keys()) + .data_dir(&data_dir) + .segment_size(1000) // should cause rotations + .initialize() + .expect("Failed to initialize DB instance"); + + let mut timeout = 5; + loop { + let result = db.get(&Value::Int(i)).expect("Failed to get record"); + match result { + None => { + thread::sleep(Duration::from_millis(timeout)); + timeout = std::cmp::min(timeout * 2, 100); + continue; + } + Some(result) => { + let result: InstSingleId = result.into(); + assert!(result.id == i); + break; + } + }; + } + })); + } + + // Add a writer that inserts the records + threads.push(thread::spawn(move || { + let mut db = DB::configure() + .fields(InstSingleId::schema()) + .primary_key(InstSingleId::primary_key()) + .secondary_keys(InstSingleId::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + for i in 0..threads_n { + db.upsert(InstSingleId { id: i }) + .expect("Failed to upsert record"); + + db.do_maintenance_tasks() // Run maintenance tasks after every write, just to test it + .expect("Failed to do maintenance tasks"); + } + })); + + for thread in threads { + thread.join().expect("Failed to join thread"); + } +} + +#[test] +fn test_log_is_rotated_when_capacity_reached() { + let data_dir = tmp_dir(); + let data_dir_path = Path::new(&data_dir); + + // Hand-calculated record length, find record below + let record_len = 1 // tombstone tag + + (1 + 8) // int tag + i64 + + (1 + 8 + 4) // string tag + string length + string data + + (1 + 8 + 3); // bytes tag + bytes length + bytes data + + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .segment_size(10 * record_len) // small log segment size + .initialize() + .expect("Failed to initialize DB instance"); + + // Insert more records than fits the capacity + for _ in 0..25 { + db.upsert(Inst { + id: 0, + name: Some("John".to_string()), + data: vec![3, 4, 5], + }) + .expect("Failed to upsert record"); + + db.do_maintenance_tasks() + .expect("Failed to do maintenance tasks"); + } + + // Check that the rotated segments exist + assert!(data_dir_path.join("metadata").with_extension("1").exists()); + assert!(data_dir_path.join("metadata").with_extension("2").exists()); + + // 3rd segment should not exist (note negation) + assert!(!data_dir_path.join("metadata").with_extension("3").exists()); +} + +#[test] +fn test_delete() { + let data_dir = tmp_dir(); + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + // Insert some records + db.upsert(Inst { + id: 0, + name: Some("John".to_string()), + data: vec![3, 4, 5], + }) + .unwrap(); + + db.upsert(Inst { + id: 1, + name: Some("John".to_string()), + data: vec![1, 2, 3], + }) + .unwrap(); + + db.delete(&Value::Int(0)).unwrap(); + + // Check that the record is deleted + assert!(db.get(&Value::Int(0)).unwrap().is_none()); + + // Check that the secondary index is updated + assert_eq!( + db.find_by(&Field::Name, &Value::String("John".to_string())) + .unwrap() + .len(), + 1 + ); +} + +#[test] +fn test_delete_by() { + let data_dir = tmp_dir(); + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + // Insert some records + db.upsert(Inst { + id: 0, + name: Some("John".to_string()), + data: vec![3, 4, 5], + }) + .unwrap(); + + db.upsert(Inst { + id: 1, + name: Some("John".to_string()), + data: vec![1, 2, 3], + }) + .unwrap(); + + db.upsert(Inst { + id: 2, + name: Some("Bob".to_string()), + data: vec![1, 2, 3], + }) + .unwrap(); + + db.delete_by(&Field::Name, &Value::String("John".to_string())) + .unwrap(); + + // Check that the record is deleted + assert!(db.get(&Value::Int(0)).unwrap().is_none()); + assert!(db.get(&Value::Int(1)).unwrap().is_none()); + + // Check that the secondary index is updated + assert_eq!( + db.find_by(&Field::Name, &Value::String("John".to_string())) + .unwrap() + .len(), + 0 + ); + assert_eq!( + db.find_by(&Field::Name, &Value::String("Bob".to_string())) + .unwrap() + .len(), + 1 + ); +} + +#[test] +fn test_range_by_id() { + let data_dir = tmp_dir(); + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + let inserted_ids = vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9]; + + for id in inserted_ids.iter() { + db.upsert(Inst { + id: *id, + name: Some("Foobar".to_string()), + data: vec![], + }) + .unwrap(); + } + + // Test range [3, 7) + let received = db + .range_by(&Field::Id, &Value::Int(3)..&Value::Int(7)) + .unwrap(); + let received_ids: Vec = received.into_iter().map(|rec| Inst::from(rec).id).collect(); + + assert_eq!(received_ids, vec![3, 4, 5, 6]); + + // Test range [3, 7] + let received = db + .range_by(&Field::Id, &Value::Int(3)..=&Value::Int(7)) + .unwrap(); + let received_ids: Vec = received.into_iter().map(|rec| Inst::from(rec).id).collect(); + + assert_eq!(received_ids, vec![3, 4, 5, 6, 7]); + + // Test range (-inf, 7] + let received = db.range_by(&Field::Id, ..=&Value::Int(7)).unwrap(); + let received_ids: Vec = received.into_iter().map(|rec| Inst::from(rec).id).collect(); + + assert_eq!(received_ids, vec![0, 1, 2, 3, 4, 5, 6, 7]); + + // Test range (3, inf) + let received = db.range_by(&Field::Id, &Value::Int(3)..).unwrap(); + let received_ids: Vec = received.into_iter().map(|rec| Inst::from(rec).id).collect(); + + assert_eq!(received_ids, vec![3, 4, 5, 6, 7, 8, 9]); +} + +#[test] +fn test_batch_find_by() { + let data_dir = tmp_dir(); + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + let inserted_ids = vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9]; + + for id in inserted_ids.iter() { + db.upsert(Inst { + id: *id, + name: Some("Foobar".to_string()), + data: vec![], + }) + .unwrap(); + } + + let batch: Vec = (2..5).map(Value::Int).collect(); + let result = db.batch_find_by(Field::Id, &batch).unwrap(); + + assert_eq!(result.len(), batch.len()); + assert_eq!( + result.iter().map(|(tag, _)| *tag).collect::>(), + vec![0, 1, 2] + ); + assert_eq!( + result + .into_iter() + .map(|(_, rec)| Inst::from(rec).id) + .collect::>(), + vec![2, 3, 4] + ); +} + +#[test] +fn test_commit_transaction() { + let data_dir = tmp_dir(); + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + db.tx_begin().expect("Failed to begin transaction"); + + db.upsert(Inst { + id: 0, + name: Some("John".to_string()), + data: vec![3, 4, 5], + }) + .unwrap(); + + db.upsert(Inst { + id: 1, + name: Some("John".to_string()), + data: vec![1, 2, 3], + }) + .unwrap(); + + db.tx_commit().expect("Failed to commit transaction"); + + let johns = db + .find_by(&Field::Name, &Value::String("John".to_string())) + .unwrap(); + + assert_eq!(johns.len(), 2); +} + +#[test] +fn test_rollback_transaction() { + let data_dir = tmp_dir(); + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + db.tx_begin().expect("Failed to begin transaction"); + + db.upsert(Inst { + id: 0, + name: Some("John".to_string()), + data: vec![3, 4, 5], + }) + .unwrap(); + + db.upsert(Inst { + id: 1, + name: Some("John".to_string()), + data: vec![1, 2, 3], + }) + .unwrap(); + + db.tx_rollback().expect("Failed to roll back transaction"); + + let johns = db + .find_by(&Field::Name, &Value::String("John".to_string())) + .unwrap(); + + assert_eq!(johns.len(), 0); +} + +#[derive(Eq, PartialEq, Clone, Debug)] +enum FieldWithNewNullableField { + Id, + Name, + Data, + MaybeStr, +} + +impl AsRef for FieldWithNewNullableField { + fn as_ref(&self) -> &str { + match self { + FieldWithNewNullableField::Id => "id", + FieldWithNewNullableField::Name => "name", + FieldWithNewNullableField::Data => "data", + FieldWithNewNullableField::MaybeStr => "maybe_str", + } + } +} + +impl Into for FieldWithNewNullableField { + fn into(self) -> String { + self.as_ref().to_owned() + } +} + +struct InstWithNewNullableField { + pub id: i64, + pub name: Option, + pub data: Vec, + pub maybe_str: Option, +} + +impl From for Record { + fn from(inst: InstWithNewNullableField) -> Record { + vec![ + Value::Int(inst.id), + match inst.name { + Some(name) => Value::String(name), + None => Value::Null, + }, + Value::Bytes(inst.data), + match inst.maybe_str { + Some(maybe_str) => Value::String(maybe_str), + None => Value::Null, + }, + ] + .into() + } +} + +impl From for InstWithNewNullableField { + fn from(record: Record) -> Self { + let mut it = record.into_iter(); + InstWithNewNullableField { + id: match it.next().unwrap() { + Value::Int(id) => id, + other => panic!("Invalid value type: {:?}", other), + }, + name: match it.next().unwrap() { + Value::String(name) => Some(name), + Value::Null => None, + other => panic!("Invalid value type: {:?}", other), + }, + data: match it.next().unwrap() { + Value::Bytes(data) => data, + other => panic!("Invalid value type: {:?}", other), + }, + maybe_str: match it.next() { + Some(Value::String(maybe_str)) => Some(maybe_str), + Some(Value::Null) => None, + None => None, + other => panic!("Invalid value type: {:?}", other), + }, + } + } +} + +impl InstWithNewNullableField { + fn schema() -> Vec { + vec![ + FieldWithNewNullableField::Id, + FieldWithNewNullableField::Name, + FieldWithNewNullableField::Data, + FieldWithNewNullableField::MaybeStr, + ] + } + + fn primary_key() -> FieldWithNewNullableField { + FieldWithNewNullableField::Id + } + + fn secondary_keys() -> Vec { + vec![FieldWithNewNullableField::Name] + } +} + +#[test] +fn test_add_nullable_field() { + let data_dir = tmp_dir(); + + // Insert a record with 3 fields + { + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + db.upsert(Inst { + id: 0, + name: Some("John".to_string()), + data: vec![3, 4, 5], + }) + .unwrap(); + } + + // Insert a record with 4 fields (last is nullable) + let mut db = DB::configure() + .fields(InstWithNewNullableField::schema()) + .primary_key(InstWithNewNullableField::primary_key()) + .secondary_keys(InstWithNewNullableField::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + db.upsert(InstWithNewNullableField { + id: 1, + name: Some("John".to_string()), + data: vec![3, 4, 5], + maybe_str: None, + }) + .unwrap(); + + let johns = db + .find_by( + &FieldWithNewNullableField::Name, + &Value::String("John".to_string()), + ) + .unwrap(); + + assert_eq!(johns.len(), 2); +} + +#[test] +fn test_delete_by_multiple_indexes() { + let data_dir = tmp_dir(); + + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + // Insert some identical records + for _ in 0..10 { + db.upsert(Inst { + id: 0, + name: Some("foo".to_string()), + data: vec![], + }) + .unwrap(); + } + + // Delete by name + db.delete_by(&Field::Name, &Value::String("foo".to_string())) + .unwrap(); + + // Check that the records are deleted by finding by name + let result = db + .find_by(&Field::Name, &Value::String("foo".to_string())) + .unwrap(); + assert_eq!(result.len(), 0); + + // Double check with find by id + let result = db.find_by(&Field::Id, &Value::Int(0)).unwrap(); + assert_eq!(result.len(), 0); +} + +#[test] +fn test_find_by_with_offset_and_limit() { + let data_dir = tmp_dir(); + + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + // Insert some records + for i in 0..10 { + db.upsert(Inst { + id: i, + name: Some("foo".to_string()), + data: vec![], + }) + .unwrap(); + } + + // Find by name with offset and limit + let result = db + .find_by_with_params( + &Field::Name, + &Value::String("foo".to_string()), + &QueryParams { + offset: 2, + limit: 3, + sort_asc: true, + }, + ) + .unwrap() + .into_iter() + .map(|rec| Inst::from(rec)) + .collect::>(); + + assert_eq!(result.len(), 3); + assert_eq!(result[0].id, 2); + assert_eq!(result[1].id, 3); + assert_eq!(result[2].id, 4); +} + +#[test] +fn test_batch_find_by_with_offset_and_limit() { + let data_dir = tmp_dir(); + + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + // Insert some records + for i in 0..10 { + db.upsert(Inst { + id: i, + name: Some("foo".to_string()), + data: vec![], + }) + .unwrap(); + } + + // Batch find by id with offset and limit + let batch: Vec = (2..5).map(Value::Int).collect(); + let result = db + .batch_find_by_with_params( + &Field::Id, + &batch, + &QueryParams { + offset: 1, + limit: 2, + sort_asc: true, + }, + ) + .unwrap() + .into_iter() + .map(|(_, rec)| Inst::from(rec)) + .collect::>(); + + assert_eq!(result.len(), 2); + assert_eq!(result[0].id, 3); + assert_eq!(result[1].id, 4); +} + +#[test] +fn test_range_by_with_offset_and_limit() { + let data_dir = tmp_dir(); + + let mut db = DB::configure() + .fields(Inst::schema()) + .primary_key(Inst::primary_key()) + .secondary_keys(Inst::secondary_keys()) + .data_dir(&data_dir) + .initialize() + .expect("Failed to initialize DB instance"); + + // Insert some records + for i in 0..10 { + db.upsert(Inst { + id: i, + name: Some("foo".to_string()), + data: vec![], + }) + .unwrap(); + } + + // Range by id with offset and limit + let result = db + .range_by_with_params( + &Field::Id, + &Value::Int(2)..&Value::Int(8), + &QueryParams { + offset: 1, + limit: 3, + sort_asc: true, + }, + ) + .unwrap() + .into_iter() + .map(|rec| Inst::from(rec)) + .collect::>(); + + assert_eq!(result.len(), 3); + assert_eq!(result[0].id, 3); + assert_eq!(result[1].id, 4); + assert_eq!(result[2].id, 5); +} diff --git a/autere_db/tests/resources/test_data_1 b/autere_db/tests/resources/test_data_1 new file mode 100644 index 0000000..5365384 Binary files /dev/null and b/autere_db/tests/resources/test_data_1 differ diff --git a/autere_db/tests/resources/test_metadata_1 b/autere_db/tests/resources/test_metadata_1 new file mode 100644 index 0000000..98dd35e Binary files /dev/null and b/autere_db/tests/resources/test_metadata_1 differ diff --git a/demo/app.py b/demo/app.py index bab20f0..ce305e9 100644 --- a/demo/app.py +++ b/demo/app.py @@ -13,7 +13,7 @@ import datetime from dataclasses import dataclass import uuid -from log_db import DB, Bound, Value +from autere_db import DB, Bound, Value app = Flask(__name__) Compress(app) diff --git a/log_db/Cargo.toml b/log_db/Cargo.toml deleted file mode 100644 index 45df64d..0000000 --- a/log_db/Cargo.toml +++ /dev/null @@ -1,27 +0,0 @@ -[package] -name = "log_db" -version = "0.1.0" -edition = "2021" - -[lib] -crate-type = ["lib", "cdylib", "staticlib"] - -[dependencies] -fs2 = "0.4.3" -log = "0.4.22" -once_cell = "1.20.2" -rust_decimal = { version = "1.36.0", features = [] } -tempfile = "3.13.0" -thiserror = "2.0.1" -uuid = { version = "1.11.0", features = ["v4"] } - -[dev-dependencies] -ctor = "0.2.8" -env_logger = "0.11.5" -serial_test = "3.1.1" -criterion = { version = "0.5", features = ["html_reports"] } -rand = "0.8.5" - -[[bench]] -name = "benchmark" -harness = false diff --git a/log_db/benches/benchmark.rs b/log_db/benches/benchmark.rs deleted file mode 100644 index d8e6280..0000000 --- a/log_db/benches/benchmark.rs +++ /dev/null @@ -1,181 +0,0 @@ -mod utils; - -use std::collections::HashSet; - -use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion}; -use log_db::*; -use tempfile; -use utils::*; - -pub fn upsert_compacted(c: &mut Criterion) { - let mut group = c.benchmark_group("upsert_compacted"); - let data_dir_obj = tempfile::tempdir().expect("Failed to get tmpdir"); - let data_dir = data_dir_obj - .path() - .to_str() - .expect("Failed to convert tmpdir path to str"); - let mut db = DB::configure() - .fields(Inst::fields()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(data_dir) - .initialize() - .expect("Failed to initialize DB"); - - let mut insts = Vec::new(); - for size in [100_000, 1_000_000, 10_000_000] { - println!("Prefilling DB to {} entries", size); - prefill_db(&mut db, &mut insts, size, true).expect("Failed to prefill DB"); - - group.bench_with_input(BenchmarkId::from_parameter(size), &size, |b, &_size| { - b.iter(|| { - let inst = random_inst(0, size as i64 + 1); - let result = db.upsert(black_box(inst)).unwrap(); - assert!(result == ()); - }); - }); - } -} - -pub fn delete_existing_compacted(c: &mut Criterion) { - let mut group = c.benchmark_group("delete_existing_compacted"); - let data_dir_obj = tempfile::tempdir().expect("Failed to get tmpdir"); - let data_dir = data_dir_obj - .path() - .to_str() - .expect("Failed to convert tmpdir path to str"); - let mut db = DB::configure() - .fields(Inst::fields()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(data_dir) - .initialize() - .expect("Failed to initialize DB"); - - let mut insts = Vec::new(); - for size in [100_000, 1_000_000, 10_000_000] { - println!("Prefilling DB to {} entries", size); - prefill_db(&mut db, &mut insts, size, true).expect("Failed to prefill DB"); - - let pk_set: HashSet = insts.iter().map(|inst| inst.id).collect(); - let mut pk_set_it = pk_set.into_iter(); - - group.bench_with_input(BenchmarkId::from_parameter(size), &size, |b, &_size| { - b.iter(|| { - let result = db - .delete(black_box(&Value::Int(pk_set_it.next().unwrap()))) - .unwrap(); - assert!(result.is_some()); - }); - }); - } -} - -pub fn upsert_write_durability(c: &mut Criterion) { - let mut group = c.benchmark_group("upsert_write_durability"); - - for mode in [WriteDurability::Flush, WriteDurability::FlushSync] { - group.bench_with_input(BenchmarkId::from_parameter(&mode), &mode, |b, _mode| { - let data_dir_obj = tempfile::tempdir().expect("Failed to get tmpdir"); - let data_dir = data_dir_obj - .path() - .to_str() - .expect("Failed to convert tmpdir path to str"); - let mut db = DB::configure() - .fields(Inst::fields()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(data_dir) - .write_durability(mode.clone()) - .initialize() - .expect("Failed to initialize DB"); - - b.iter(|| { - let inst = random_inst(0, 1000); - let result = db.upsert(black_box(inst)).unwrap(); - assert!(result == ()); - }); - }); - } -} - -pub fn get_existing_compacted(c: &mut Criterion) { - let mut group = c.benchmark_group("get_existing_compacted"); - - let data_dir_obj = tempfile::tempdir().expect("Failed to get tmpdir"); - let data_dir = data_dir_obj - .path() - .to_str() - .expect("Failed to convert tmpdir path to str"); - let mut db = DB::configure() - .fields(Inst::fields()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(data_dir) - .initialize() - .expect("Failed to initialize DB"); - - let mut insts = Vec::new(); - let mut inst_index = 0; - for size in [100_000, 1_000_000, 10_000_000] { - println!("Prefilling DB to {} entries", size); - prefill_db(&mut db, &mut insts, size, true).expect("Failed to prefill DB"); - - group.bench_with_input(BenchmarkId::from_parameter(size), &size, |b, &_size| { - b.iter(|| { - let id = insts[inst_index].id; - let result = db.get(black_box(&Value::Int(id))).unwrap(); - assert!(result.is_some()); - inst_index = (inst_index + 1) % insts.len(); - }); - }); - } -} - -pub fn find_by_existing_compacted(c: &mut Criterion) { - let mut group = c.benchmark_group("find_by_existing_compacted"); - - let data_dir_path = tempfile::tempdir() - .expect("Failed to get tmpdir") - .into_path(); - let data_dir = data_dir_path - .to_str() - .expect("Failed to convert tmpdir path to str"); - let mut db = DB::configure() - .fields(Inst::fields()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(data_dir) - .initialize() - .expect("Failed to initialize DB"); - - let mut insts = Vec::new(); - let mut inst_index = 0; - for size in [100_000, 1_000_000, 10_000_000] { - println!("Prefilling DB to {} entries", size); - prefill_db(&mut db, &mut insts, size, true).expect("Failed to prefill DB"); - println!("DB prefilling done, insts.len = {}", insts.len()); - - group.bench_with_input(BenchmarkId::from_parameter(size), &size, |b, &_size| { - b.iter(|| { - let name = insts[inst_index].name.clone(); - let result = db - .find_by(black_box(&Field::Name), black_box(&Value::String(name))) - .unwrap(); - assert!(result.len() > 0); - inst_index = (inst_index + 1) % insts.len(); - }); - }); - } -} - -// Register the benchmark group -criterion_group!( - benches, - upsert_compacted, - delete_existing_compacted, - upsert_write_durability, - get_existing_compacted, - find_by_existing_compacted, -); -criterion_main!(benches); diff --git a/log_db/benches/utils.rs b/log_db/benches/utils.rs deleted file mode 100644 index 33a45cf..0000000 --- a/log_db/benches/utils.rs +++ /dev/null @@ -1,122 +0,0 @@ -use log_db::*; -use rand::distributions::Alphanumeric; -use rand::Rng; -use std::fmt::Debug; - -#[derive(Eq, PartialEq, Clone, Debug)] -pub enum Field { - Id, - Name, - Data, -} - -#[derive(PartialEq, Eq, Debug, Clone)] -pub struct Inst { - pub id: i64, - pub name: String, - pub data: Vec, -} - -impl AsRef for Field { - fn as_ref(&self) -> &str { - match self { - Field::Id => "id", - Field::Name => "name", - Field::Data => "data", - } - } -} - -impl Into for Field { - fn into(self) -> String { - self.as_ref().to_string() - } -} - -impl From for Record { - fn from(inst: Inst) -> Self { - vec![ - Value::Int(inst.id), - Value::String(inst.name), - Value::Bytes(inst.data), - ] - .into() - } -} - -impl From for Inst { - fn from(record: Record) -> Self { - let mut it = record.into_iter(); - Inst { - id: match it.next().unwrap() { - Value::Int(id) => id, - _ => panic!("Expected Int"), - }, - name: match it.next().unwrap() { - Value::String(name) => name, - _ => panic!("Expected String"), - }, - data: match it.next().unwrap() { - Value::Bytes(data) => data, - _ => panic!("Expected Bytes"), - }, - } - } -} - -impl Inst { - pub fn fields() -> Vec { - vec![Field::Id, Field::Name, Field::Data] - } - pub fn primary_key() -> Field { - Field::Id - } - pub fn secondary_keys() -> Vec { - vec![Field::Name] - } -} - -// Function to generate a random integer -pub fn random_int(from: i64, to: i64) -> i64 { - let mut rng = rand::thread_rng(); - rng.gen_range(from..to) -} - -// Function to generate a random string -pub fn random_string(len: usize) -> String { - let mut rng = rand::thread_rng(); - (0..len).map(|_| rng.sample(Alphanumeric) as char).collect() -} - -// Function to generate random bytes -pub fn random_bytes(len: usize) -> Vec { - let mut rng = rand::thread_rng(); - (0..len).map(|_| rng.gen()).collect() -} - -// Function to generate a random Inst -pub fn random_inst(from_id: i64, to_id: i64) -> Inst { - Inst { - id: random_int(from_id, to_id), // Random int value between 0..1000 - name: random_string(5), // Random string of length 5 - data: random_bytes(10), // Random bytes of length 10 - } -} - -pub fn prefill_db( - db: &mut DB, - insts: &mut Vec, - n_records: usize, - compact: bool, -) -> DBResult<()> { - for i in 0..(n_records - insts.len()) { - let inst = random_inst(0, n_records as i64); - insts.push(inst.clone()); - db.upsert(inst)?; - if i % 1000 == 0 && compact { - db.do_maintenance_tasks()?; - } - } - - Ok(()) -} diff --git a/log_db/src/common.rs b/log_db/src/common.rs deleted file mode 100644 index 0f6a058..0000000 --- a/log_db/src/common.rs +++ /dev/null @@ -1,574 +0,0 @@ -use super::*; - -use std::collections::btree_map::Values; -// For Unix-like systems -#[cfg(unix)] -use std::os::unix::fs::MetadataExt; - -// For Windows -#[cfg(windows)] -use std::os::windows::fs::MetadataExt; - -pub const ACTIVE_SYMLINK_FILENAME: &str = "active"; -pub const LOCK_FILENAME: &str = "lock"; -pub const EXCL_LOCK_REQ_FILENAME: &str = "excl_lock_req"; -pub const INITIALIZED_FILENAME: &str = "initialized"; - -pub const METADATA_FILE_HEADER_SIZE: usize = 24; -pub const METADATA_ROW_LENGTH: usize = 16; -pub const LOCK_WAIT_MAX_MS: u64 = 1000; - -// Serialized value tags -pub const B_NULL: u8 = 0x0; -pub const B_INT: u8 = 0x1; -pub const B_DECIMAL: u8 = 0x2; -pub const B_STRING: u8 = 0x3; -pub const B_BYTES: u8 = 0x4; -// Tombstone marker tags -pub const B_LIVE: u8 = 0x0; -pub const B_TOMBSTONE: u8 = 0xFF; - -pub fn metadata_filename(num: u16) -> String { - format!("metadata.{}", num) -} - -pub type DBResult = Result; - -#[derive(Debug, Error)] -pub enum DBError { - #[error("lock request failed: {0}")] - LockRequestError(String), - #[error("validation failed: {0}")] - ValidationError(String), - #[error("consistency check failed: {0}")] - ConsistencyError(String), - #[error("invalid transaction: {0}")] - TransactionError(String), - #[error("unexpected IO error: {0}")] - IOError(#[from] io::Error), -} - -#[derive(Debug, Error)] -pub enum LogKeyMapError { - #[error("log key not found in map")] - NotFoundError, - #[error("attempted to remove last element of non-empty map")] - RemovingLastElementError, -} - -/// LogKey is a packed struct that contains: -/// - a log segment number (16 bits) -/// - a log index within the segment (48 bits) -#[derive(Debug, Clone, Eq, PartialEq, Hash, Ord, PartialOrd)] -pub struct LogKey(u64); - -impl LogKey { - pub fn new(segment_num: u16, index: u64) -> Self { - assert!(index < (1 << 48), "Index must fit in 48 bits"); - LogKey((segment_num as u64) << 48 | index) - } - - pub fn segment_num(&self) -> u16 { - (self.0 >> 48) as u16 - } - - pub fn index(&self) -> u64 { - self.0 & 0x0000_FFFF_FFFF_FFFF - } -} - -/// LogKeyMap is a non-empty map of PK => LogKey mappings. -#[derive(Debug, Clone, Eq, PartialEq)] -pub struct LogKeyMap { - map: BTreeMap, -} - -impl LogKeyMap { - /// Create a new LogKeyMap with an initial mapping. - /// The initial mapping is required since LogKeyMap must be non-empty. - pub fn new_with_initial(pk: IndexableValue, log_key: LogKey) -> Self { - let mut map = BTreeMap::new(); - map.insert(pk, log_key); - LogKeyMap { map } - } - - pub fn contains_pk(&self, key: &IndexableValue) -> bool { - self.map.contains_key(key) - } - - /// The number of LogKeys in the map. - pub fn len(&self) -> usize { - self.map.len() - } - - /// Insert a PK -> LogKey mapping into the map. - pub fn insert(&mut self, key: IndexableValue, log_key: LogKey) { - self.map.insert(key, log_key); - } - - /// Remove a mapping from the map. Return Ok(()) if the key was found and removed. - /// Return `LogKeyMapError::RemovingLastElementError` if trying to remove the last element. - /// Return `LogKeyMapError::NotFoundError` if the key was not found. - pub fn remove_pk(&mut self, key: &IndexableValue) -> Result<(), LogKeyMapError> { - if self.map.len() == 1 { - return Err(LogKeyMapError::RemovingLastElementError); - } - let removed = self.map.remove(key); - - if removed.is_none() { - return Err(LogKeyMapError::NotFoundError); - } - - assert!( - self.map.len() > 0, - "LogKeyMap should not be empty after removal" - ); - - Ok(()) - } - - /// Get a reference to the set of LogKeys. - pub fn log_keys(&self) -> Values { - self.map.values() - } -} - -pub static APPEND_MODE: Lazy = Lazy::new(|| { - let mut options = fs::OpenOptions::new(); - options.read(true).append(true); - options -}); -pub static READ_MODE: Lazy = Lazy::new(|| { - let mut options = fs::OpenOptions::new(); - options.read(true); - options -}); -pub static WRITE_MODE: Lazy = Lazy::new(|| { - let mut options = fs::OpenOptions::new(); - options.read(true).write(true); - options -}); - -pub struct MetadataHeader { - pub version: u8, - pub uuid: Uuid, -} - -const METADATA_HEADER_PADDING: &[u8] = &[0; 7]; -impl MetadataHeader { - pub fn serialize(&self) -> [u8; METADATA_FILE_HEADER_SIZE] { - let mut header = [0u8; METADATA_FILE_HEADER_SIZE]; - header[0] = self.version; - header[1..8].copy_from_slice(METADATA_HEADER_PADDING); - header[8..].copy_from_slice(self.uuid.as_bytes()); - - header - } - - pub fn deserialize(bytes: &[u8]) -> Self { - assert_eq!(bytes.len(), METADATA_FILE_HEADER_SIZE); - - let version = bytes[0]; - let uuid = Uuid::from_slice(&bytes[8..24]).expect("Failed to deserialize Uuid"); - - MetadataHeader { version, uuid } - } -} - -#[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Hash)] -pub enum IndexableValue { - Null, - Int(i64), - Decimal(Decimal), - String(String), -} - -#[derive(Debug, Clone)] -pub enum Value { - Null, - Int(i64), - Decimal(Decimal), - String(String), - Bytes(Vec), -} - -impl PartialEq for Value { - fn eq(&self, other: &Self) -> bool { - match (self, other) { - (Value::Int(a), Value::Int(b)) => a == b, - (Value::Decimal(a), Value::Decimal(b)) => a == b, - (Value::String(a), Value::String(b)) => a == b, - (Value::Bytes(a), Value::Bytes(b)) => a == b, - (Value::Null, Value::Null) => true, - _ => false, - } - } -} -impl Eq for Value {} - -impl Value { - pub fn serialize(&self) -> Vec { - match self { - Value::Null => vec![B_NULL], - Value::Int(i) => { - let mut bytes = Vec::with_capacity(1 + 16); - bytes.push(B_INT); - bytes.extend_from_slice(&i.to_be_bytes()); - bytes - } - Value::Decimal(d) => { - let mut bytes = Vec::with_capacity(1 + 16); - bytes.push(B_DECIMAL); - bytes.extend_from_slice(&d.serialize()); - bytes - } - Value::String(s) => { - let len = s.len(); - let mut bytes = Vec::with_capacity(1 + 8 + len); - bytes.push(B_STRING); - bytes.extend_from_slice(&(len as u64).to_be_bytes()); - bytes.extend_from_slice(s.as_bytes()); - bytes - } - Value::Bytes(b) => { - let len = b.len(); - let mut bytes = Vec::with_capacity(1 + 8 + len); - bytes.push(B_BYTES); - bytes.extend_from_slice(&(len as u64).to_be_bytes()); - bytes.extend_from_slice(b); - bytes - } - } - } - - /// Deserialize a Value from a byte slice. - /// Returns the deserialized Value and the number of bytes consumed. - pub fn deserialize(bytes: &[u8]) -> (Value, usize) { - match bytes[0] { - B_NULL => (Value::Null, 1), - B_INT => { - let mut int_bytes = [0; 8]; - int_bytes.copy_from_slice(&bytes[1..1 + 8]); - (Value::Int(i64::from_be_bytes(int_bytes)), 1 + 8) - } - B_DECIMAL => { - let mut decimal_bytes = [0; 16]; - decimal_bytes.copy_from_slice(&bytes[1..1 + 16]); - (Value::Decimal(Decimal::deserialize(decimal_bytes)), 1 + 16) - } - B_STRING => { - let length_bytes = &bytes[1..1 + 8]; - let length = u64::from_be_bytes(length_bytes.try_into().unwrap()) as usize; - ( - Value::String( - String::from_utf8(bytes[1 + 8..1 + 8 + length].to_vec()).unwrap(), - ), - 1 + 8 + length, - ) - } - B_BYTES => { - let length_bytes = &bytes[1..1 + 8]; - let length = u64::from_be_bytes(length_bytes.try_into().unwrap()) as usize; - ( - Value::Bytes(bytes[1 + 8..1 + 8 + length].to_vec()), - 1 + 8 + length, - ) - } - _ => panic!("Invalid tag: {}", bytes[0]), - } - } - - pub fn as_indexable(&self) -> Option { - match self { - Value::Null => Some(IndexableValue::Null), - Value::Int(i) => Some(IndexableValue::Int(*i)), - Value::Decimal(d) => Some(IndexableValue::Decimal(d.clone())), - Value::String(s) => Some(IndexableValue::String(s.clone())), - _ => None, - } - } -} - -pub fn get_secondary_memtable_index_by_field(sks: &Vec, field: &str) -> Option { - sks.iter().position(|schema_field| schema_field == field) -} - -pub fn is_file_same_as_path(file: &File, path: &PathBuf) -> DBResult { - // Get the metadata for the open file handle - let file_metadata = file.metadata()?; - - // Get the metadata for the file at the specified path - let path_metadata = metadata(path)?; - - // Platform-specific comparison - #[cfg(unix)] - { - Ok( - file_metadata.dev() == path_metadata.dev() - && file_metadata.ino() == path_metadata.ino(), - ) - } - - #[cfg(windows)] - { - Ok(file_metadata.file_index() == path_metadata.file_index() - && file_metadata.volume_serial_number() == path_metadata.volume_serial_number()) - } -} - -pub fn symlink(original: &Path, link: &Path) -> io::Result<()> { - #[cfg(unix)] - { - std::os::unix::fs::symlink(original, link) - } - - #[cfg(windows)] - { - std::os::windows::fs::symlink_file(original, link) - } -} - -/// Set the active segment to the segment with the given ordinal number. -pub fn set_active_segment(data_dir_path: &Path, segment_num: u16) -> DBResult<()> { - let tmp_uuid = Uuid::new_v4(); - let tmp_filename = format!("active_{}", tmp_uuid.to_string()); - let tmp_path = data_dir_path.join(tmp_filename); - - let metadata_filename = format!("metadata.{}", segment_num); - let metadata_path = Path::new(&metadata_filename); - let active_symlink = data_dir_path.join(ACTIVE_SYMLINK_FILENAME); - - symlink(&metadata_path, &tmp_path)?; - fs::rename(&tmp_path, &active_symlink)?; - - Ok(()) -} - -/// Create a new segment metadata file and return its number and path. -/// A metadata file contains the segment metadata, including the UUID of the data file. -/// See `ARCHITECTURE.md` for the file format. -pub fn create_segment_metadata_file( - data_dir_path: &Path, - data_file_uuid: &Uuid, -) -> DBResult<(u16, PathBuf)> { - let current_greatest_num = greatest_segment_number(data_dir_path)?; - let new_num = current_greatest_num + 1; - - let metadata_filename = format!("metadata.{}", new_num); - let metadata_path = data_dir_path.join(metadata_filename); - - let mut metadata_file = fs::OpenOptions::new() - .create(true) - .append(true) - .open(&metadata_path)?; - - let metadata_header = MetadataHeader { - version: 1, - uuid: *data_file_uuid, - }; - - metadata_file.write_all(&metadata_header.serialize())?; - metadata_file.flush()?; - - let len = metadata_file.seek(io::SeekFrom::End(0))?; - assert!(len >= METADATA_FILE_HEADER_SIZE as u64); - assert_eq!((len - METADATA_FILE_HEADER_SIZE as u64) % 16, 0); - - Ok((new_num, metadata_path)) -} - -/// Parse the segment number from a metadata file path -pub fn parse_segment_number(metadata_path: &Path) -> DBResult { - let filename = metadata_path - .file_name() - .expect("No filename in symlink") - .to_str() - .expect("Filename was not valid UTF-8"); - - // parse number from format "metadata.1" - let segment_number = filename - .split('.') - .last() - .expect("Filename did not have a number") - .parse::(); - - segment_number.map_err(|_| { - DBError::ValidationError("Failed to parse segment number from filename".to_owned()) - }) -} - -/// Get the number of the segment with the greatest ordinal. -/// This is the newest segment, i.e. the one that is pointed to by the `active` symlink. -/// If there are no segments yet, returns 0. -pub fn greatest_segment_number(data_dir_path: &Path) -> DBResult { - let active_symlink = data_dir_path.join(ACTIVE_SYMLINK_FILENAME); - - if !fs::exists(&active_symlink)? { - return Ok(0); - } - - let segment_metadata_path = fs::read_link(&active_symlink)?; - parse_segment_number(&segment_metadata_path) -} - -/// Create a new segment data file and return its UUID. -/// A data file contains the segment data, tightly packed without separators. -/// An accompanying metadata file is required to interpret the data. -pub fn create_segment_data_file(data_dir_path: &Path) -> DBResult<(Uuid, PathBuf)> { - let uuid = Uuid::new_v4(); - let new_segment_path = data_dir_path.join(uuid.to_string()); - fs::OpenOptions::new() - .create(true) - .append(true) - .open(&new_segment_path)?; - - Ok((uuid, new_segment_path)) -} - -/// Reads the metadata header from the metadata file. -/// Leaves the file seek head at the beginning of the records, after the header. -pub fn read_metadata_header(metadata_file: &mut fs::File) -> DBResult { - metadata_file.seek(SeekFrom::Start(0))?; - let mut buf = [0u8; METADATA_FILE_HEADER_SIZE]; - metadata_file.read_exact(&mut buf)?; - - let header = MetadataHeader::deserialize(&buf); - Ok(header) -} - -pub fn validate_metadata_header(header: &MetadataHeader) -> DBResult<()> { - if header.version != 1 { - return Err(DBError::ValidationError( - "Unsupported metadata file version".to_owned(), - )); - } - - Ok(()) -} - -pub enum IsMetadatafileValidResult { - Ok, - ReplaceFile, - TruncateToSize(u64), -} - -pub fn is_metadata_file_valid(metadata_file: &mut fs::File) -> DBResult { - let size = metadata_file.seek(SeekFrom::End(0))? as usize; - - if size < METADATA_FILE_HEADER_SIZE { - return Ok(IsMetadatafileValidResult::ReplaceFile); - } - - // The data section must be a multiple of 16 bytes. - // Otherwise, the non-aligned part of the file is dropped. - let data_section_len = size - METADATA_FILE_HEADER_SIZE; - let remainder = data_section_len % 16; - if remainder != 0 { - return Ok(IsMetadatafileValidResult::TruncateToSize( - (size - remainder) as u64, - )); - } - - Ok(IsMetadatafileValidResult::Ok) -} - -/// Check that the active metadata file is well-formed and repair it if necessary. -/// The metadata file is considered well-formed if its size is, in pseudocode, `header_size + n * record_size`. -/// If the file is not well-formed, it is truncated to the last well-formed record using -/// a temporary file and an atomic move operation. -/// -/// `self.active_metadata_file` must be a locked file handle opened with read permissions. -/// The function leaves the seek head in an unspecified position. -/// -/// Returns `false` if the file was repaired and rotated, `true` if no action was taken. -pub fn ensure_active_metadata_is_valid( - data_dir: &Path, - metadata_file: &mut fs::File, -) -> DBResult { - let current_len = metadata_file.seek(SeekFrom::End(0))? as usize; - - match is_metadata_file_valid(metadata_file)? { - IsMetadatafileValidResult::Ok => return Ok(true), - IsMetadatafileValidResult::ReplaceFile => { - let active_target = fs::read_link(data_dir.join(ACTIVE_SYMLINK_FILENAME))?; - let active_path = data_dir.join(&active_target); - warn!( - "Metadata file \"{}\" is malformed ({} bytes), replacing it with an empty file", - active_target.display(), - current_len, - ); - let mut tmp_file = tempfile::NamedTempFile::new()?; - - let header = MetadataHeader { - version: 1, - uuid: Uuid::new_v4(), - }; - - tmp_file.write_all(&header.serialize())?; - tmp_file.flush()?; - - fs::rename(tmp_file.path(), active_path)?; - - debug!("Replaced metadata file"); - return Ok(false); - } - IsMetadatafileValidResult::TruncateToSize(new_size) => { - let active_target = fs::read_link(data_dir.join(ACTIVE_SYMLINK_FILENAME))?; - let active_path = data_dir.join(&active_target); - warn!( - "Metadata file \"{}\" is malformed ({} bytes), truncating it to {} bytes", - active_target.display(), - current_len, - new_size - ); - - let mut tmp_file = tempfile::NamedTempFile::new()?; - - let mut buf = vec![0; new_size as usize]; - metadata_file.seek(SeekFrom::Start(0))?; - metadata_file.read_exact(&mut buf)?; - - tmp_file.write_all(&buf)?; - tmp_file.flush()?; - - fs::rename(tmp_file.path(), active_path)?; - - debug!("Truncated metadata file"); - return Ok(false); - } - } -} - -pub struct OwnedBounds { - start: Bound, - end: Bound, -} - -impl OwnedBounds { - pub fn new(start: Bound, end: Bound) -> Self { - OwnedBounds { start, end } - } -} - -impl RangeBounds for OwnedBounds { - fn start_bound(&self) -> Bound<&T> { - self.start.as_ref() - } - - fn end_bound(&self) -> Bound<&T> { - self.end.as_ref() - } -} - -#[derive(Debug, Clone)] -pub struct QueryParams { - pub offset: usize, - pub limit: usize, - pub sort_asc: bool, -} - -pub static DEFAULT_QUERY_PARAMS: QueryParams = QueryParams { - offset: 0, - limit: usize::MAX, - sort_asc: true, -}; diff --git a/log_db/src/config.rs b/log_db/src/config.rs deleted file mode 100644 index e8c6fa1..0000000 --- a/log_db/src/config.rs +++ /dev/null @@ -1,140 +0,0 @@ -use super::*; - -pub struct ConfigBuilder { - data_dir: Option, - segment_size: Option, - write_durability: Option, - read_consistency: Option, - - fields: Option>, - primary_key: Option, - secondary_keys: Option>, -} - -impl ConfigBuilder { - pub fn new() -> ConfigBuilder { - ConfigBuilder { - data_dir: None, - segment_size: None, - write_durability: None, - read_consistency: None, - - fields: None, - primary_key: None, - secondary_keys: None, - } - } - - /// The directory where the database will store its data. - pub fn data_dir(mut self, data_dir: impl Into) -> Self { - self.data_dir = Some(data_dir.into()); - self - } - - /// The maximum size of a segment file in bytes. - /// Once a segment file reaches this size, it can be closed, rotated and compacted. - /// Note that this is not a hard limit: if `db.do_maintenance_tasks()` is not called, - /// the segment file may continue to grow. - pub fn segment_size(mut self, segment_size: usize) -> Self { - self.segment_size = Some(segment_size); - self - } - - /// The write durability policy for the database. - /// This determines how writes are persisted to disk. - /// The default is WriteDurability::Flush. - pub fn write_durability(mut self, write_durability: WriteDurability) -> Self { - self.write_durability = Some(write_durability); - self - } - - /// The read consistency policy for the database. - /// This determines how recent writes are visible when reading. - /// See individual `ReadConsistency` enum values for more information. - /// The default is ReadConsistency::Strong. - pub fn read_consistency(mut self, read_consistency: ReadConsistency) -> Self { - self.read_consistency = Some(read_consistency); - self - } - - pub fn fields(mut self, schema: Vec>) -> Self { - self.fields = Some(schema.into_iter().map(|s| s.into()).collect()); - self - } - - pub fn primary_key(mut self, primary_key: impl Into) -> Self { - self.primary_key = Some(primary_key.into()); - self - } - - pub fn secondary_keys(mut self, secondary_keys: Vec>) -> Self { - self.secondary_keys = Some(secondary_keys.into_iter().map(|s| s.into()).collect()); - self - } - - pub fn initialize(self) -> DBResult { - let schema = self - .fields - .ok_or_else(|| DBError::ValidationError("Schema not set".to_string()))?; - let primary_key = self - .primary_key - .ok_or_else(|| DBError::ValidationError("Primary key not set".to_string()))?; - - let config = Config { - schema, - primary_key, - secondary_keys: self.secondary_keys.unwrap_or_default(), - - data_dir: self.data_dir.clone().unwrap_or("db_data".to_string()), - segment_size: self.segment_size.unwrap_or(4 * 1024 * 1024), // 4MB - write_durability: self - .write_durability - .clone() - .unwrap_or(WriteDurability::Flush), - read_consistency: self - .read_consistency - .clone() - .unwrap_or(ReadConsistency::Strong), - }; - - DB::initialize(config) - } -} - -#[derive(Clone)] -pub struct Config { - pub schema: Vec, - pub primary_key: String, - pub secondary_keys: Vec, - pub data_dir: String, - pub segment_size: usize, - pub write_durability: WriteDurability, - pub read_consistency: ReadConsistency, -} - -#[derive(Debug, Clone, Eq, PartialEq)] -pub enum ReadConsistency { - /// Reads by client A are guaranteed to see writes by themselves and any writes by other clients B - /// that were done before last index refresh. You must call `refresh_indexes()` manually to refresh indexes. - Eventual, - /// Reads by client A are guaranteed to see all writes. This is slower: all reads must first - /// refresh indexes. - Strong, -} - -#[derive(Debug, Clone, Eq, PartialEq)] -pub enum WriteDurability { - /// Changes are written to the OS write buffer but not immediately synced to disk. - /// This is generally recommended. Most OSes will sync the write buffer to disk within a few seconds. - Flush, - /// Changes are written to the OS write buffer and synced to disk immediately. - /// Offers the best durability guarantees but is a lot slower. - FlushSync, -} - -impl Display for WriteDurability { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> { - write!(f, "{:?}", self)?; - Ok(()) - } -} diff --git a/log_db/src/engine.rs b/log_db/src/engine.rs deleted file mode 100644 index 6fb9edf..0000000 --- a/log_db/src/engine.rs +++ /dev/null @@ -1,852 +0,0 @@ -use super::*; - -pub struct Engine { - pub config: Config, - pub lock_manager: LockManager, - - data_dir_path: PathBuf, - primary_key_index: usize, - refresh_next_logkey: LogKey, - - pub tx_active: bool, - pub tx_log: Vec, - - active_metadata_file: fs::File, - active_data_file: fs::File, - - // TODO: these could be made private. Currently they are public for testing in lib.rs. - pub primary_memtable: PrimaryMemtable, - pub secondary_memtables: Vec, -} - -impl Engine { - pub fn initialize(config: Config) -> DBResult { - info!("Initializing DB..."); - // If data_dir does not exist or is empty, create it and any necessary files. - // After creation, the directory should always be in a complete state without missing files. - - // Ensure the data directory exists - let data_dir_path = Path::new(&config.data_dir).to_path_buf(); - match fs::create_dir(&data_dir_path) { - Ok(_) => {} - Err(e) => { - if e.kind() != io::ErrorKind::AlreadyExists { - return Err(DBError::IOError(e)); - } - } - } - - // Create the lock file first to prevent multiple concurrent initializations - let mut lock_manager = LockManager::new(data_dir_path.clone())?; - lock_manager.lock_exclusive()?; - - // We have acquired the lock, check if the data directory is in a complete state - // If not, initialize it, otherwise skip. - if !fs::exists(data_dir_path.join(INITIALIZED_FILENAME))? { - // Delete all files except the lock files to ensure a clean state - for entry in fs::read_dir(&data_dir_path)? { - let entry = entry?; - let path = entry.path(); - if path.is_file() - && path.file_name().unwrap() != LOCK_FILENAME - && path.file_name().unwrap() != EXCL_LOCK_REQ_FILENAME - { - fs::remove_file(&path)?; - } - } - - // Create the initial segment files - let (segment_uuid, _) = create_segment_data_file(&data_dir_path)?; - let (segment_num, _) = create_segment_metadata_file(&data_dir_path, &segment_uuid)?; - set_active_segment(&data_dir_path, segment_num)?; - - // Create the initialized file to indicate that the directory is in a complete state - fs::File::create(data_dir_path.join(INITIALIZED_FILENAME))?; - } - - // Calculate the index of the primary value in a record - let primary_key_index = config - .schema - .iter() - .position(|field| field == &config.primary_key) - .ok_or(DBError::ValidationError( - "Primary key not found in schema after initialize".to_owned(), - ))?; - - // Join primary key and secondary keys vec into a single vec - let mut all_keys = vec![&config.primary_key]; - all_keys.extend(&config.secondary_keys); - - // If any of the keys is not in the schema or - // is not an IndexableValue, return an error - for &key in &all_keys { - let _ = config.schema.iter().find(|&field| field == key).ok_or( - DBError::ValidationError("Key must be present in the field schema".to_owned()), - )?; - } - let primary_memtable = PrimaryMemtable::new(); - let secondary_memtables = config - .secondary_keys - .iter() - .map(|_| SecondaryMemtable::new()) - .collect(); - - let active_symlink = Path::new(&config.data_dir).join(ACTIVE_SYMLINK_FILENAME); - - let active_target = fs::read_link(&active_symlink)?; - let active_metadata_path = Path::new(&config.data_dir).join(active_target); - let mut active_metadata_file = APPEND_MODE.open(&active_metadata_path)?; - - let active_metadata_header = read_metadata_header(&mut active_metadata_file)?; - validate_metadata_header(&active_metadata_header)?; - - let active_data_path = - Path::new(&config.data_dir).join(active_metadata_header.uuid.to_string()); - let active_data_file = APPEND_MODE.open(&active_data_path)?; - - let mut engine = Engine { - config, - lock_manager, - data_dir_path, - primary_key_index, - primary_memtable, - secondary_memtables, - active_metadata_file, - active_data_file, - refresh_next_logkey: LogKey::new(1, 0), - tx_active: false, - tx_log: vec![], - }; - - info!("Rebuilding memtable indexes..."); - engine.refresh_indexes()?; - - info!("Database ready."); - - engine.lock_manager.unlock()?; - Ok(engine) - } - - pub fn refresh_indexes(&mut self) -> DBResult<()> { - let active_symlink_path = self.data_dir_path.join(ACTIVE_SYMLINK_FILENAME); - let active_target = fs::read_link(active_symlink_path)?; - let active_metadata_path = self.data_dir_path.join(active_target); - - let to_segnum = parse_segment_number(&active_metadata_path)?; - let from_segnum = self.refresh_next_logkey.segment_num(); - let mut from_index = self.refresh_next_logkey.index(); - - for segnum in from_segnum..=to_segnum { - let metadata_path = self.data_dir_path.join(metadata_filename(segnum)); - let mut metadata_file = READ_MODE.open(&metadata_path)?; - - let metadata_len = metadata_file.seek(SeekFrom::End(0))?; - if (metadata_len - METADATA_FILE_HEADER_SIZE as u64) % METADATA_ROW_LENGTH as u64 != 0 { - return Err(DBError::ConsistencyError(format!( - "Metadata file {} has invalid size: {}", - metadata_path.display(), - metadata_len - ))); - } - - let metadata_header = read_metadata_header(&mut metadata_file)?; - validate_metadata_header(&metadata_header)?; - - let data_path = self.data_dir_path.join(metadata_header.uuid.to_string()); - let data_file = READ_MODE.open(data_path)?; - - for ForwardLogReaderItem { row, index } in - ForwardLogReader::new_with_index(metadata_file, data_file, from_index) - { - let log_key = LogKey::new(segnum, index); - - if row.tombstone { - self.remove_row_from_memtables(&row.values); - } else { - self.insert_row_to_memtables(log_key, row.values); - } - - // Update from_index in case this is the last iteration: we need to know the next - // index that should be read on later invocations of refresh_indexes. - from_index = index + 1 - } - - // If there are still segments to read, set from_index to zero to read them - // from beginning. Otherwise we leave from_index as the index of the next record to read. - if segnum != to_segnum { - from_index = 0 - } - } - - self.refresh_next_logkey = LogKey::new(to_segnum, from_index); - - Ok(()) - } - - fn insert_row_to_memtables(&mut self, log_key: LogKey, row_values: Vec) { - let pk = row_values[self.primary_key_index].as_indexable().unwrap(); - - for (sk_index, sk_field) in self.config.secondary_keys.iter().enumerate() { - let secondary_memtable = &mut self.secondary_memtables[sk_index]; - let sk_field_index = self - .config - .schema - .iter() - .position(|f| sk_field == f) - .unwrap(); - let sk = row_values[sk_field_index].as_indexable().unwrap(); - - secondary_memtable.set(pk.clone(), sk, log_key.clone()); - } - - // Doing this last because this moves log_key - self.primary_memtable.set(pk, log_key); - } - - fn remove_row_from_memtables(&mut self, row_values: &Vec) { - let pk = row_values[self.primary_key_index].as_indexable().unwrap(); - - if let Some(_) = self.primary_memtable.remove(&pk) { - for (sk_index, sk_field) in self.config.secondary_keys.iter_mut().enumerate() { - let secondary_memtable = &mut self.secondary_memtables[sk_index]; - let sk_field_index = self - .config - .schema - .iter() - .position(|f| sk_field == f) - .unwrap(); - let sk = row_values[sk_field_index].as_indexable().unwrap(); - - secondary_memtable.remove(&pk, &sk); - } - } - } - - pub fn upsert_record(&mut self, record: Row) -> DBResult<()> { - debug!("Opening file in append mode..."); - - if !self.ensure_metadata_file_is_active()? - || !ensure_active_metadata_is_valid( - &self.data_dir_path, - &mut self.active_metadata_file, - )? - { - // The log file has been rotated, so we must try again - return self.upsert_record(record); - } - - self.tx_log.push(TxEntry::Upsert { row: record }); - - if !self.tx_active { - self.commit_transaction()?; - self.tx_log.clear(); - } - - Ok(()) - } - - pub fn batch_find_by_records<'a>( - &mut self, - field: &str, - values: impl Iterator, - params: &QueryParams, - ) -> DBResult> { - let indexables = values - .map(|value| { - value.as_indexable().ok_or(DBError::ValidationError( - "Queried value must be indexable".to_owned(), - )) - }) - .collect::>>()?; - - // Otherwise, continue with querying secondary indexes. - debug!("Finding all records with matching fields"); - - if self.config.read_consistency == ReadConsistency::Strong { - self.refresh_indexes()?; - } - - let log_key_batches = indexables - .into_iter() - .map(|query_key| { - if field == &self.config.primary_key { - let opt = self.primary_memtable.get(&query_key); - let log_keys = match opt { - Some(log_key) => vec![log_key], - None => vec![], - }; - Ok(log_keys) - } else { - let smemtable_index = match get_secondary_memtable_index_by_field( - &self.config.secondary_keys, - field, - ) { - Some(index) => index, - None => { - return Err(DBError::ValidationError( - "Cannot find_by by non-indexed key".to_owned(), - )) - } - }; - - let log_keys = self.secondary_memtables[smemtable_index] - .find_by(&query_key) - .into_iter() - .collect(); - Ok(log_keys) - } - }) - .collect::>>>()?; - - debug!("Found log keys in memtable: {:?}", log_key_batches); - - let mut tagged = vec![]; - for (tag, batch) in log_key_batches.into_iter().enumerate() { - let mapped = batch.into_iter().map(|log_key| (tag, log_key)); - tagged.extend(mapped); - } - - if !params.sort_asc { - tagged.reverse(); - } - let bound_low = params.offset; - let bound_high = (params.offset + params.limit).min(tagged.len()); - let sliced = &tagged[bound_low..bound_high]; - let mut tagged_records = self.read_tagged_log_keys(sliced.into_iter())?; - - debug!("Read {} records", tagged_records.len()); - - if !params.sort_asc { - tagged_records.reverse(); - } - Ok(tagged_records) - } - - /// Read records from segment files based on log keys. - /// The log keys are accompanied by an integer tag that can be used to identify and group them later. - fn read_tagged_log_keys<'a>( - &self, - log_keys: impl Iterator, - ) -> DBResult> { - let mut records = vec![]; - let mut log_keys_map = BTreeMap::new(); - - for (tag, log_key) in log_keys { - if !log_keys_map.contains_key(&log_key.segment_num()) { - log_keys_map.insert(log_key.segment_num(), vec![(tag, log_key.index())]); - } else { - log_keys_map - .get_mut(&log_key.segment_num()) - .unwrap() - .push((tag, log_key.index())); - } - } - - for (segment_num, mut segment_indexes) in log_keys_map { - segment_indexes.sort_unstable(); - - let metadata_path = &self.data_dir_path.join(metadata_filename(segment_num)); - let mut metadata_file = READ_MODE.open(&metadata_path)?; - - let metadata_header = read_metadata_header(&mut metadata_file)?; - - let data_path = &self.data_dir_path.join(metadata_header.uuid.to_string()); - let mut data_file = READ_MODE.open(&data_path)?; - - let header_size = METADATA_FILE_HEADER_SIZE as i64; - let row_length = METADATA_ROW_LENGTH as i64; - let mut current_metadata_offset = header_size; - for (tag, segment_index) in segment_indexes { - let new_metadata_offset = header_size + segment_index as i64 * row_length; - metadata_file.seek_relative(new_metadata_offset - current_metadata_offset)?; - - let mut metadata_buf = [0; METADATA_ROW_LENGTH]; - metadata_file.read_exact(&mut metadata_buf)?; - - let data_offset = u64::from_be_bytes(metadata_buf[0..8].try_into().unwrap()); - let data_length = u64::from_be_bytes(metadata_buf[8..16].try_into().unwrap()); - assert!(data_length > 0); - - data_file.seek(SeekFrom::Start(data_offset))?; - - let mut data_buf = vec![0; data_length as usize]; - data_file.read_exact(&mut data_buf)?; - - let record = Row::deserialize(&data_buf); - records.push((*tag, record)); - - current_metadata_offset = new_metadata_offset + row_length; - } - } - - Ok(records) - } - - pub fn range_by_records>( - &mut self, - field: &str, - range: B, - params: &QueryParams, - ) -> DBResult> { - fn range_bound_to_indexable(bound: Bound<&Value>) -> DBResult> { - match bound { - Bound::Included(value) => value - .as_indexable() - .ok_or(DBError::ValidationError( - "Queried value must be indexable".to_owned(), - )) - .map(Bound::Included), - Bound::Excluded(value) => value - .as_indexable() - .ok_or(DBError::ValidationError( - "Queried value must be indexable".to_owned(), - )) - .map(Bound::Excluded), - Bound::Unbounded => Ok(Bound::Unbounded), - } - } - - let start_indexable = range_bound_to_indexable(range.start_bound())?; - let end_indexable = range_bound_to_indexable(range.end_bound())?; - - let indexable_bounds = OwnedBounds::new(start_indexable, end_indexable); - - if self.config.read_consistency == ReadConsistency::Strong { - self.refresh_indexes()?; - } - - let log_keys = if field == &self.config.primary_key { - self.primary_memtable.range(indexable_bounds) - } else { - let index = get_secondary_memtable_index_by_field(&self.config.secondary_keys, field) - .ok_or_else(|| { - DBError::ValidationError("Cannot range_by by non-indexed key".to_owned()) - })?; - - self.secondary_memtables[index].range(indexable_bounds) - }; - - let mut log_key_batches: Vec<(usize, &LogKey)> = - log_keys.into_iter().map(|log_key| (0, log_key)).collect(); - - if !params.sort_asc { - log_key_batches.reverse(); - } - - let bound_low = params.offset; - let bound_high = (params.offset + params.limit).min(log_key_batches.len()); - let sliced = &log_key_batches[bound_low..bound_high]; - let tagged_records = self.read_tagged_log_keys(sliced.into_iter()); - - let mut result_records: Vec = - tagged_records?.into_iter().map(|(_, rec)| rec).collect(); - - if !params.sort_asc { - result_records.reverse(); - } - - Ok(result_records) - } - - /// Ensures that the `self.metadata_file` and `self.data_file` handles are still pointing to the correct files. - /// If the segment has been rotated, the handle will be closed and reopened. - /// Returns `false` if the file has been rotated and the handle has been reopened, `true` otherwise. - fn ensure_metadata_file_is_active(&mut self) -> DBResult { - let active_target = fs::read_link(&self.data_dir_path.join(ACTIVE_SYMLINK_FILENAME))?; - let active_metadata_path = &self.data_dir_path.join(active_target); - - let correct = is_file_same_as_path(&self.active_metadata_file, &active_metadata_path)?; - if !correct { - debug!("Metadata file has been rotated. Reopening..."); - let metadata_file = APPEND_MODE.open(&active_metadata_path)?; - - let metadata_header = read_metadata_header(&mut self.active_metadata_file)?; - - validate_metadata_header(&metadata_header)?; - - let data_file_path = &self.data_dir_path.join(metadata_header.uuid.to_string()); - - self.active_metadata_file = metadata_file; - self.active_data_file = APPEND_MODE.open(&data_file_path)?; - - return Ok(false); - } else { - return Ok(true); - } - } - - pub fn delete_by_field(&mut self, field: &str, value: &Value) -> DBResult> { - let recs: Vec = self - .batch_find_by_records(field, std::iter::once(value), &DEFAULT_QUERY_PARAMS)? - .into_iter() - .map(|(_, mut rec)| { - rec.tombstone = true; - rec - }) - .collect(); - - // TODO: refactor the clone out of here - for record in &recs { - self.tx_log.push(TxEntry::Delete { - row: record.clone(), - }); - } - - if !self.tx_active { - self.commit_transaction()?; - self.tx_log.clear(); - } - - debug!("Records deleted"); - - Ok(recs) - } - - pub fn commit_transaction(&mut self) -> DBResult<()> { - let active_symlink_path = self.data_dir_path.join(ACTIVE_SYMLINK_FILENAME); - let active_target = fs::read_link(active_symlink_path)?; - let segment_num = parse_segment_number(&active_target)?; - - let initial_data_offset = self.active_data_file.seek(SeekFrom::End(0))?; - let initial_metadata_offset = self.active_metadata_file.seek(SeekFrom::End(0))?; - let mut serialized_data: Vec = vec![]; - let mut serialized_metadata: Vec = Vec::with_capacity(self.tx_log.len() * 16); - let mut pending_memtable_ops: Vec<(LogKey, TxEntry)> = vec![]; - - let mut metadata_buf = [0u8; 16]; - - debug!("Serializing tx_log to byte arrays"); - for tx_entry in &self.tx_log { - let record = match tx_entry { - TxEntry::Upsert { row: record } => record, - TxEntry::Delete { row: record } => record, - }; - - let serialized = record.serialize(); - let record_offset = initial_data_offset + serialized_data.len() as u64; - let record_length = serialized.len() as u64; - assert!(record_length > 0); - - serialized_data.extend(serialized); - - let metadata_pos = initial_metadata_offset + serialized_metadata.len() as u64; - let metadata_index = - (metadata_pos - METADATA_FILE_HEADER_SIZE as u64) / METADATA_ROW_LENGTH as u64; - - // Write the record metadata to the fixed-size metadata buffer - metadata_buf[..8].copy_from_slice(&record_offset.to_be_bytes()); - metadata_buf[8..].copy_from_slice(&record_length.to_be_bytes()); - - serialized_metadata.extend_from_slice(&metadata_buf); - - let log_key = LogKey::new(segment_num, metadata_index); - pending_memtable_ops.push((log_key, tx_entry.clone())); - } - - debug!("Writing serialized bytearrays to log files"); - self.active_data_file.write_all(&serialized_data)?; - self.active_metadata_file.write_all(&serialized_metadata)?; - - // Flush and sync data and metadata to disk - if self.config.write_durability == WriteDurability::Flush { - self.active_data_file.flush()?; - self.active_metadata_file.flush()?; - } else if self.config.write_durability == WriteDurability::FlushSync { - self.active_data_file.flush()?; - self.active_data_file.sync_all()?; - self.active_metadata_file.flush()?; - self.active_metadata_file.sync_all()?; - } - - debug!("Updating memtables"); - for (log_key, tx_entry) in pending_memtable_ops { - match tx_entry { - TxEntry::Upsert { row } => self.insert_row_to_memtables(log_key, row.values), - TxEntry::Delete { row } => self.remove_row_from_memtables(&row.values), - } - } - debug!("Commit done"); - - Ok(()) - } - - pub fn do_maintenance_tasks(&mut self) -> DBResult<()> { - ensure_active_metadata_is_valid(&self.data_dir_path, &mut self.active_metadata_file)?; - - let metadata_size = self.active_metadata_file.seek(SeekFrom::End(0))?; - if metadata_size >= self.config.segment_size as u64 { - self.rotate_and_compact()?; - } - - Ok(()) - } - - fn rotate_and_compact(&mut self) -> DBResult<()> { - debug!("Active log size exceeds threshold, starting rotation and compaction..."); - - let original_data_len = self.active_data_file.seek(SeekFrom::End(0))?; - - let active_target = fs::read_link(&self.data_dir_path.join(ACTIVE_SYMLINK_FILENAME))?; - let active_num = parse_segment_number(&active_target)?; - - debug!("Reading segment data into a BTreeMap"); - let mut pk_to_item_map: BTreeMap<&IndexableValue, &Row> = BTreeMap::new(); - let forward_read_items: Vec<(IndexableValue, Row)> = ForwardLogReader::new( - self.active_metadata_file.try_clone()?, - self.active_data_file.try_clone()?, - ) - .map(|item| { - ( - item.row.values[self.primary_key_index] - .as_indexable() - .expect("Primary key was not indexable"), - item.row, - ) - }) - .collect(); - - for (pk, record) in forward_read_items.iter() { - pk_to_item_map.insert(pk, record); - } - - debug!( - "Read {} records, out of which {} were unique", - forward_read_items.len(), - pk_to_item_map.len() - ); - - // Create a new log data file and write it - debug!("Opening new data file and writing compacted data"); - let (new_data_uuid, new_data_path) = create_segment_data_file(&self.data_dir_path)?; - let mut new_data_file = APPEND_MODE.open(&new_data_path)?; - - let mut pk_to_data_map = BTreeMap::new(); - let mut offset = 0u64; - for (pk, record) in pk_to_item_map.into_iter() { - let serialized = record.serialize(); - let len = serialized.len() as u64; - new_data_file.write_all(&serialized)?; - - pk_to_data_map.insert(pk, (offset, len)); - offset += len; - } - - // Sync the data file to disk. - // This is fine to do without consulting WriteDurability because this is a one-off - // operation that is not part of the normal write path. - new_data_file.flush()?; - new_data_file.sync_all()?; - - let final_data_len = new_data_file.seek(io::SeekFrom::End(0))?; - debug!( - "Wrote compacted data, reduced data size: {} -> {}", - original_data_len, final_data_len - ); - - // Create a new log metadata file and write it - debug!("Opening temp metadata file and writing pointers to compacted data file"); - let temp_metadata_file = tempfile::NamedTempFile::new()?; - let temp_metadata_path = temp_metadata_file.as_ref(); - let mut temp_metadata_file = WRITE_MODE.open(temp_metadata_path)?; - - let metadata_header = MetadataHeader { - version: 1, - uuid: new_data_uuid, - }; - - temp_metadata_file.write_all(&metadata_header.serialize())?; - - let mut metadata_buf = [0u8; 16]; - for (pk, _) in forward_read_items.iter() { - let (offset, len) = pk_to_data_map.get(&pk).unwrap(); - - metadata_buf[..8].copy_from_slice(&offset.to_be_bytes()); - metadata_buf[8..].copy_from_slice(&len.to_be_bytes()); - - temp_metadata_file.write_all(&metadata_buf)?; - } - - // Sync the metadata file to disk, see comment above about sync. - temp_metadata_file.flush()?; - temp_metadata_file.sync_all()?; - - debug!("Moving temporary files to their final locations"); - let new_data_path = &self.data_dir_path.join(new_data_uuid.to_string()); - let active_metadata_path = &self.data_dir_path.join(metadata_filename(active_num)); // overwrite active - - fs::rename(&temp_metadata_path, &active_metadata_path)?; - - debug!("Compaction complete, creating new segment"); - - let new_segment_num = active_num + 1; - let new_metadata_path = self.data_dir_path.join(metadata_filename(new_segment_num)); - let mut new_metadata_file = APPEND_MODE.clone().create(true).open(&new_metadata_path)?; - - let new_metadata_header = MetadataHeader { - version: 1, - uuid: new_data_uuid, - }; - - new_metadata_file.write_all(&new_metadata_header.serialize())?; - - set_active_segment(&self.data_dir_path, new_segment_num)?; - - self.active_metadata_file = APPEND_MODE.open(&new_metadata_path)?; - self.active_data_file = APPEND_MODE.open(&new_data_path)?; - - debug!( - "Active log file {} rotated and compacted, new segment: {}", - active_num, new_segment_num - ); - - Ok(()) - } - - #[inline] - pub fn with_exclusive_lock( - &mut self, - f: impl FnOnce(&mut Self) -> DBResult, - ) -> DBResult { - // No need to acquire a lock if a transaction is already active - // because the lock is already held. - if !self.tx_active { - self.lock_manager.lock_exclusive()?; - } - let result = f(self); - if !self.tx_active { - self.lock_manager.unlock()?; - } - result - } - - #[inline] - pub fn with_shared_lock(&mut self, f: impl FnOnce(&mut Self) -> DBResult) -> DBResult { - // No need to acquire a lock if a transaction is already active - // because the lock is already held. - if !self.tx_active { - self.lock_manager.lock_shared()?; - } - let result = f(self); - if !self.tx_active { - self.lock_manager.unlock()?; - } - - result - } -} - -#[cfg(test)] -mod tests { - use ctor::ctor; - use env_logger; - - use super::*; - - #[ctor] - fn init_logger() { - let _ = env_logger::builder().is_test(true).try_init(); - } - - #[derive(Eq, PartialEq, Clone, Debug)] - enum Field { - Id, - Name, - } - - impl Into for Field { - fn into(self) -> String { - match self { - Field::Id => "id".to_owned(), - Field::Name => "name".to_owned(), - } - } - } - - #[derive(PartialEq, Eq, Debug, Clone)] - struct TestInst2 { - id: i64, - name: String, - } - - impl From for Vec { - fn from(inst: TestInst2) -> Self { - vec![Value::Int(inst.id), Value::String(inst.name)] - } - } - - impl From> for TestInst2 { - fn from(record: Vec) -> Self { - let mut it = record.into_iter(); - TestInst2 { - id: match it.next().unwrap() { - Value::Int(i) => i, - _ => panic!("Expected int"), - }, - name: match it.next().unwrap() { - Value::String(s) => s, - _ => panic!("Expected string"), - }, - } - } - } - - #[test] - fn test_memtable_insert_and_delete() { - let temp_dir = tempfile::tempdir().unwrap(); - let data_dir = temp_dir.path(); - - let capacity = 5; - let segment_size = capacity * 2 * 8 + METADATA_FILE_HEADER_SIZE; - - let mut db = DB::configure() - .data_dir(data_dir.to_str().unwrap()) - .fields(vec![Field::Id, Field::Name]) - .primary_key(Field::Id) - .secondary_keys(vec![Field::Name]) - .segment_size(segment_size) - .initialize() - .expect("Failed to create DB"); - - let engine = &mut db.engine; - - let inst = TestInst2 { - id: 0, - name: "foo".to_owned(), - }; - let id = IndexableValue::Int(0); - - assert_eq!(engine.primary_memtable.get(&id), None); - assert_eq!( - engine.secondary_memtables[0] - .find_by(&IndexableValue::String("foo".to_owned())) - .len(), - 0 - ); - engine.insert_row_to_memtables(LogKey::new(1, 0), inst.clone().into()); - assert_eq!(engine.primary_memtable.get(&id), Some(&LogKey::new(1, 0))); - assert_eq!( - engine.secondary_memtables[0] - .find_by(&IndexableValue::String("foo".to_owned())) - .len(), - 1 - ); - - engine.insert_row_to_memtables(LogKey::new(1, 1), inst.clone().into()); - assert_eq!(engine.primary_memtable.get(&id), Some(&LogKey::new(1, 1))); - assert_eq!( - engine.secondary_memtables[0] - .find_by(&IndexableValue::String("foo".to_owned())) - .len(), - 1 - ); - - engine.remove_row_from_memtables(&inst.into()); - assert_eq!(engine.primary_memtable.get(&id), None); - assert_eq!( - engine.secondary_memtables[0] - .find_by(&IndexableValue::String("foo".to_owned())) - .len(), - 0 - ); - } -} diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs deleted file mode 100644 index 006056c..0000000 --- a/log_db/src/lib.rs +++ /dev/null @@ -1,549 +0,0 @@ -#[macro_use] -extern crate log; - -use once_cell::sync::Lazy; -use rust_decimal::Decimal; -use std::collections::BTreeMap; -use std::fmt::Debug; -use std::fmt::Display; -use std::fs::{self, metadata, File}; -use std::io::{self, Read, Seek, SeekFrom, Write}; -use std::ops::*; -use std::path::{Path, PathBuf}; -use std::thread; -use thiserror::Error; -use uuid::Uuid; - -#[macro_use] -mod common; -mod config; -mod engine; -mod lock; -mod log_reader_forward; -mod memtable_primary; -mod memtable_secondary; -mod record; -mod row; -mod schema; - -pub use common::{DBError, DBResult, OwnedBounds, QueryParams, Value, DEFAULT_QUERY_PARAMS}; -pub use config::{ReadConsistency, WriteDurability}; -pub use record::Record; -pub use schema::Schema; - -use common::*; -use config::*; -use engine::*; -use lock::*; -use log_reader_forward::*; -use memtable_primary::PrimaryMemtable; -use memtable_secondary::SecondaryMemtable; -use row::*; - -pub struct DB { - engine: Engine, -} - -impl DB { - /// Create a new database configuration builder. - pub fn configure() -> ConfigBuilder { - ConfigBuilder::new() - } - - fn initialize(config: Config) -> DBResult { - let engine = Engine::initialize(config)?; - Ok(DB { engine }) - } - - /// Insert a record into the database. If the primary key value already exists, - /// the existing record will be replaced by the supplied one. - pub fn upsert(&mut self, record: impl Into) -> DBResult<()> { - let row = Row { - values: record.into().into(), - tombstone: false, - }; - debug!("Upserting record: {:?}", row); - - self.engine - .with_exclusive_lock(move |engine| engine.upsert_record(row))?; - - Ok(()) - } - - /// Get a record by its primary index value. - /// E.g. `db.get(Value::Int(10))`. - pub fn get(&mut self, value: &Value) -> DBResult> { - let tagged_rows = self.engine.with_shared_lock(|engine| { - engine.batch_find_by_records( - // TODO: This clone is only here to appease the borrow checker - &engine.config.primary_key.clone(), - std::iter::once(value), - &DEFAULT_QUERY_PARAMS, - ) - })?; - - assert!(tagged_rows.len() <= 1); - - Ok(tagged_rows - .into_iter() - .next() - .map(|(_, row)| Record::from(row))) - } - - /// Get a collection of records based on an indexed field value. - pub fn find_by(&mut self, field: impl AsRef, value: &Value) -> DBResult> { - let tagged_rows = self.engine.with_shared_lock(|engine| { - engine.batch_find_by_records( - field.as_ref(), - std::iter::once(value), - &DEFAULT_QUERY_PARAMS, - ) - })?; - - Ok(tagged_rows - .into_iter() - .map(|(_, row)| Record::from(row)) - .collect()) - } - - /// Get a collection of records based on an indexed field value, with additional parameters. - pub fn find_by_with_params( - &mut self, - field: impl AsRef, - value: &Value, - params: &QueryParams, - ) -> DBResult> { - let recs = self.engine.with_shared_lock(|engine| { - engine.batch_find_by_records(field.as_ref(), std::iter::once(value), params) - })?; - - Ok(recs.into_iter().map(|(_, row)| Record::from(row)).collect()) - } - - /// Get a collection of records based on a sequence of indexed field values. - /// Returns a vector of pairs where the first value is an index into the given sequence of values, - /// and the second value is the record. - pub fn batch_find_by( - &mut self, - field: impl Into, - values: &[Value], - ) -> DBResult> { - let recs = self.engine.with_shared_lock(|engine| { - engine.batch_find_by_records(&field.into(), values.iter(), &DEFAULT_QUERY_PARAMS) - })?; - - Ok(recs - .into_iter() - .map(|(tag, row)| (tag, Record::from(row))) - .collect()) - } - - /// Get a collection of records based on a sequence of indexed field values, with additional parameters. - /// Returns a vector of pairs where the first value is an index into the given sequence of values, - /// and the second value is the record. - pub fn batch_find_by_with_params( - &mut self, - field: impl AsRef, - values: &[Value], - params: &QueryParams, - ) -> DBResult> { - let recs = self.engine.with_shared_lock(|engine| { - engine.batch_find_by_records(field.as_ref(), values.iter(), params) - })?; - - Ok(recs - .into_iter() - .map(|(tag, row)| (tag, Record::from(row))) - .collect()) - } - - /// Get a collection of records based on a range of indexed field values. - /// This method can be used to run comparison-like queries, e.g. `field >= 10` - /// could be expressed as `db.range_by(Field::Id, 10..)`. - pub fn range_by>( - &mut self, - field: impl AsRef, - range: B, - ) -> DBResult> { - let recs = self.engine.with_shared_lock(|engine| { - engine.range_by_records(field.as_ref(), range, &DEFAULT_QUERY_PARAMS) - })?; - - Ok(recs.into_iter().map(|row| Record::from(row)).collect()) - } - - /// Get a collection of records based on a range of indexed field values, with additional parameters. - /// This method can be used to run comparison-like queries, e.g. `field >= 10` - /// could be expressed as `db.range_by(Field::Id, 10..)`. - pub fn range_by_with_params>( - &mut self, - field: impl AsRef, - range: B, - params: &QueryParams, - ) -> DBResult> { - let recs = self - .engine - .with_shared_lock(|engine| engine.range_by_records(field.as_ref(), range, params))?; - - Ok(recs.into_iter().map(|row| Record::from(row)).collect()) - } - - /// Delete records by a field value. - /// E.g. `db.delete_by(Field::Name, "John")`, assuming `Field` is the DB field type and `Field::Name` is secondary indexed. - /// Returns a vector of deleted records. If no records were deleted, the vector will be empty. - /// - /// Deletion is done by marking the record as a tombstone. The record will still be present in the log file, - /// but will be ignored by reads. Upon compaction, tombstoned records will be removed. - pub fn delete_by(&mut self, field: impl AsRef, value: &Value) -> DBResult> { - let recs = self - .engine - .with_exclusive_lock(|engine| engine.delete_by_field(field.as_ref(), value))?; - - Ok(recs - .into_iter() - .map(|row| Record::from(row.values)) - .collect()) - } - - /// Delete record by primary key. - pub fn delete(&mut self, pk: &Value) -> DBResult> { - let recs = self.engine.with_exclusive_lock(|engine| { - engine - // TODO: This clone is only here to appease the borrow checker - .delete_by_field(&engine.config.primary_key.clone(), pk) - })?; - - assert!(recs.len() <= 1); - - Ok(recs.into_iter().next().map(|row| Record::from(row.values))) - } - - /// Check if there are any pending tasks and do them. Tasks include: - /// - Rotating the active log file if it has reached capacity and compacting it. - /// - /// This function should be called periodically to ensure that the database remains in an optimal state. - /// Note that this function is synchronous and may block for a relatively long time. - /// You may call this function in a separate thread or process to avoid blocking the main thread. - /// However, the database will be exclusively locked, so all writes and reads will be blocked during the tasks. - pub fn do_maintenance_tasks(&mut self) -> DBResult<()> { - self.engine - .with_exclusive_lock(|engine| engine.do_maintenance_tasks()) - } - - /// Refresh the in-memory indexes from the log files. - /// This needs to only be called if the read consistency is set to `ReadConsistency::Eventual`. - pub fn refresh_indexes(&mut self) -> DBResult<()> { - self.engine - .with_exclusive_lock(|engine| engine.refresh_indexes()) - } - - /// Begin a transaction. This will acquire an exclusive lock on the database, - /// preventing other clients from using the database until the transaction is committed or rolled back. - pub fn tx_begin(&mut self) -> DBResult<()> { - if self.engine.tx_active { - return Err(DBError::TransactionError( - "Transaction already active".to_string(), - )); - } - - self.engine.lock_manager.lock_exclusive()?; - self.engine.tx_active = true; - Ok(()) - } - - /// Commit the active transaction. A transaction must be active, otherwise - /// a `DBError::TransactionError` will be returned. - pub fn tx_commit(&mut self) -> DBResult<()> { - if !self.engine.tx_active { - return Err(DBError::TransactionError( - "No active transaction to commit".to_string(), - )); - } - - self.engine.commit_transaction()?; - self.engine.tx_log.clear(); - self.engine.tx_active = false; - self.engine.lock_manager.unlock()?; - Ok(()) - } - - /// Rollback the active transaction. A transaction must be active, otherwise - /// a `DBError::TransactionError` will be returned. - pub fn tx_rollback(&mut self) -> DBResult<()> { - if !self.engine.tx_active { - return Err(DBError::TransactionError( - "No active transaction to roll back".to_string(), - )); - } - - self.engine.tx_log.clear(); - self.engine.tx_active = false; - self.engine.lock_manager.unlock()?; - Ok(()) - } -} - -#[cfg(test)] -mod tests { - use ctor::ctor; - use env_logger; - - use super::*; - - #[ctor] - fn init_logger() { - let _ = env_logger::builder().is_test(true).try_init(); - } - - #[derive(Eq, PartialEq, Clone, Debug)] - enum Field { - Id, - Name, - } - - impl Into for Field { - fn into(self) -> String { - match self { - Field::Id => "id".to_string(), - Field::Name => "name".to_string(), - } - } - } - - struct TestInst1 { - id: i64, - } - - impl From for Record { - fn from(inst: TestInst1) -> Self { - vec![Value::Int(inst.id)].into() - } - } - - impl From for TestInst1 { - fn from(record: Record) -> Self { - let mut it = record.into_iter(); - TestInst1 { - id: match it.next().unwrap() { - Value::Int(i) => i, - _ => panic!("Expected int"), - }, - } - } - } - - struct TestInst2 { - id: i64, - name: String, - } - - impl From for Record { - fn from(inst: TestInst2) -> Self { - vec![Value::Int(inst.id), Value::String(inst.name)].into() - } - } - - impl From for TestInst2 { - fn from(record: Record) -> Self { - let mut it = record.into_iter(); - TestInst2 { - id: match it.next().unwrap() { - Value::Int(i) => i, - _ => panic!("Expected int"), - }, - name: match it.next().unwrap() { - Value::String(s) => s, - _ => panic!("Expected string"), - }, - } - } - } - - #[test] - fn test_compaction() { - let temp_dir = tempfile::tempdir().unwrap(); - let data_dir = temp_dir.path(); - - let capacity = 5; - let segment_size = capacity * 2 * 8 + METADATA_FILE_HEADER_SIZE; - - let mut db = DB::configure() - .data_dir(data_dir.to_str().unwrap()) - .fields(vec![Field::Id]) - .primary_key(Field::Id) - .segment_size(segment_size) - .initialize() - .expect("Failed to create DB"); - - // Insert records with same value until we reach the capacity - for _ in 0..capacity { - db.upsert(TestInst1 { id: 0 }) - .expect("Failed to insert record"); - } - - let mut segment1_file = READ_MODE.open(data_dir.join(metadata_filename(1))).unwrap(); - let segment1_metadata_size_original = segment1_file.seek(io::SeekFrom::End(0)).unwrap(); - - let segment1_header = read_metadata_header(&mut segment1_file).unwrap(); - let mut segment1_data_file = READ_MODE - .open(data_dir.join(segment1_header.uuid.to_string())) - .unwrap(); - let segment1_data_size_original = segment1_data_file.seek(io::SeekFrom::End(0)).unwrap(); - - // Rotate and compact - db.do_maintenance_tasks() - .expect("Failed to do maintenance tasks"); - - // Insert one extra with different value, this goes into another segment - db.upsert(TestInst1 { id: 1 }) - .expect("Failed to insert record"); - - // Check that rotation resulted in 2 segments - assert!(fs::exists(data_dir.join(metadata_filename(1))).unwrap()); - assert!(fs::exists(data_dir.join(metadata_filename(2))).unwrap()); - // Note negation here - assert!(!fs::exists(data_dir.join(metadata_filename(3))).unwrap()); - - // Check that the compacted metadata file has the same size - let mut segment1_metadata_file_compacted = - READ_MODE.open(data_dir.join(metadata_filename(1))).unwrap(); - let segment1_metadata_size_compacted = segment1_metadata_file_compacted - .seek(io::SeekFrom::End(0)) - .unwrap(); - assert_eq!( - segment1_metadata_size_compacted, - segment1_metadata_size_original - ); - - // Check that the compacted data file is smaller - let segment1_header_compacted = - read_metadata_header(&mut segment1_metadata_file_compacted).unwrap(); - let mut segment1_data_file_compacted = READ_MODE - .open(data_dir.join(segment1_header_compacted.uuid.to_string())) - .unwrap(); - let segment1_data_size_compacted = segment1_data_file_compacted - .seek(io::SeekFrom::End(0)) - .unwrap(); - assert!( - segment1_data_size_compacted < segment1_data_size_original, - "Original: {}, Compacted: {}", - segment1_data_size_original, - segment1_data_size_compacted - ); - - // Check that the records can be read - let inst0: TestInst1 = db - .get(&Value::Int(0 as i64)) - .expect("Failed to get record") - .expect("Record not found") - .into(); - - assert!(inst0.id == 0); - - let inst1: TestInst1 = db - .get(&Value::Int(1 as i64)) - .expect("Failed to get record") - .expect("Record not found") - .into(); - - assert!(inst1.id == 1); - } - - #[test] - fn test_repair() { - let temp_dir = tempfile::tempdir().unwrap(); - let data_dir = temp_dir.path(); - - let mut db = DB::configure() - .data_dir(data_dir.to_str().unwrap()) - .fields(vec![Field::Id]) - .primary_key(Field::Id) - .initialize() - .expect("Failed to create DB"); - - // Insert records - let n_recs: u64 = 100; - for i in 0..n_recs { - db.upsert(TestInst1 { id: i as i64 }) - .expect("Failed to insert record"); - } - - // Open the segment file and write garbage to it to simulate corruption - let segment_metadata_path = data_dir.join(metadata_filename(1)); - let mut file = APPEND_MODE - .open(&segment_metadata_path) - .expect("Failed to open file"); - - file.write_all(&[1, 0, 0, 0]) // A partially written integer value ([1] + some bytes) - .expect("Failed to write garbage"); - file.flush().unwrap(); - - let len = file.seek(SeekFrom::End(0)).expect("Failed to seek"); - assert_ne!(len, METADATA_FILE_HEADER_SIZE as u64 + n_recs * 16); - - // Try to refresh indexes, reading the file from beginning to end: should lead to error - db.refresh_indexes() - .expect_err("refresh_indexes should fail because of partial write"); - - // Trigger autorepair - db.do_maintenance_tasks() - .expect("Failed to run maintenance tasks"); - - // Try to refresh indexes, reading the file from beginning to end: should work now - db.refresh_indexes() - .expect("refresh_indexes should succeed"); - - // Reopen file and check that it has the correct size - let mut file = READ_MODE - .open(&segment_metadata_path) - .expect("Failed to open file"); - let len = file.seek(SeekFrom::End(0)).expect("Failed to seek"); - assert_eq!(len, METADATA_FILE_HEADER_SIZE as u64 + n_recs * 16); - } - - #[test] - fn test_memtables_updated_on_write() { - let temp_dir = tempfile::tempdir().unwrap(); - let data_dir = temp_dir.path(); - - let mut db = DB::configure() - .data_dir(data_dir.to_str().unwrap()) - .fields(vec![Field::Id, Field::Name]) - .primary_key(Field::Id) - .secondary_keys(vec![Field::Name]) - .initialize() - .expect("Failed to create DB"); - - // Check that the key is not indexed before write - assert_eq!( - db.engine.primary_memtable.get(&IndexableValue::Int(0)), - None - ); - assert_eq!( - db.engine.secondary_memtables[0] - .find_by(&IndexableValue::String("John".to_string())) - .len(), - 0 - ); - - // Insert record - db.upsert(TestInst2 { - id: 0, - name: "John".to_owned(), - }) - .expect("Failed to insert record"); - - // Check that the key is now indexed - let expected_log_key = LogKey::new(1, 0); - let expected_pk = IndexableValue::Int(0); - assert_eq!( - db.engine.primary_memtable.get(&expected_pk), - Some(&expected_log_key) - ); - let expected_vals = vec![&expected_log_key]; - let actual_vals = db.engine.secondary_memtables[0] - .find_by(&IndexableValue::String("John".to_string())) - .collect::>(); - assert_eq!(actual_vals, expected_vals); - } -} diff --git a/log_db/src/lock.rs b/log_db/src/lock.rs deleted file mode 100644 index f00429f..0000000 --- a/log_db/src/lock.rs +++ /dev/null @@ -1,117 +0,0 @@ -use super::*; - -pub struct LockManager { - lock_file: fs::File, - excl_lock_file: fs::File, - - state: LockState, -} - -#[derive(Debug, PartialEq, Eq)] -enum LockState { - NotLocked, - Shared, - Exclusive, -} - -impl LockManager { - pub fn new(data_dir_path: PathBuf) -> DBResult { - let lock_file = fs::File::create(data_dir_path.join(LOCK_FILENAME))?; - let excl_lock_file = fs::File::create(data_dir_path.join(EXCL_LOCK_REQ_FILENAME))?; - - Ok(LockManager { - lock_file, - excl_lock_file, - state: LockState::NotLocked, - }) - } - - fn is_exclusive_lock_requested(&self) -> DBResult { - // Attempt to acquire a shared lock on the lock request file - // If the file is already locked, return false - match fs2::FileExt::try_lock_shared(&self.excl_lock_file) { - Err(e) => { - if e.kind() == fs2::lock_contended_error().kind() { - return Ok(true); - } - return Err(DBError::IOError(e)); - } - - Ok(_) => { - fs2::FileExt::unlock(&self.excl_lock_file)?; - return Ok(false); - } - } - } - - pub fn lock_shared(&mut self) -> DBResult<()> { - if self.state == LockState::Shared { - return Err(DBError::LockRequestError( - "Already holding a shared lock".to_owned(), - )); - } else if self.state == LockState::Exclusive { - return Err(DBError::LockRequestError( - "Cannot acquire shared lock while holding an exclusive lock".to_owned(), - )); - } - - let mut timeout = 5; - loop { - if self.is_exclusive_lock_requested()? { - debug!( - "Exclusive lock requested, waiting for {}ms before requesting a shared lock again", - timeout - ); - thread::sleep(std::time::Duration::from_millis(timeout)); - timeout *= 2; - - if timeout > LOCK_WAIT_MAX_MS { - return Err(DBError::LockRequestError( - "Acquisition of shared lock timed out after {LOCK_WAIT_MAX_MS}".to_owned(), - )); - } - } else { - fs2::FileExt::lock_shared(&self.lock_file)?; - self.state = LockState::Shared; - return Ok(()); - } - } - } - - pub fn lock_exclusive(&mut self) -> DBResult<()> { - if self.state == LockState::Exclusive { - return Err(DBError::LockRequestError( - "Already holding an exclusive lock".to_owned(), - )); - } else if self.state == LockState::Shared { - return Err(DBError::LockRequestError( - "Cannot acquire exclusive lock while holding a shared lock".to_owned(), - )); - } - - // Create a lock on the exclusive lock request file to signal to readers that they should wait - // This will block until the lock is acquired - fs2::FileExt::lock_exclusive(&self.excl_lock_file)?; - - // Acquire an exclusive lock on the actual lock files - fs2::FileExt::lock_exclusive(&self.lock_file)?; - self.state = LockState::Exclusive; - - // Unlock the request file - fs2::FileExt::unlock(&self.excl_lock_file)?; - - Ok(()) - } - - pub fn unlock(&mut self) -> DBResult<()> { - if self.state == LockState::NotLocked { - return Err(DBError::LockRequestError( - "Not holding any locks".to_owned(), - )); - } - - fs2::FileExt::unlock(&self.lock_file)?; - self.state = LockState::NotLocked; - Ok(()) - } -} diff --git a/log_db/src/log_reader_forward.rs b/log_db/src/log_reader_forward.rs deleted file mode 100644 index f3fc16c..0000000 --- a/log_db/src/log_reader_forward.rs +++ /dev/null @@ -1,134 +0,0 @@ -use super::*; - -pub struct ForwardLogReader { - metadata_reader: io::BufReader, - data_reader: io::BufReader, -} - -pub struct ForwardLogReaderItem { - pub row: Row, - pub index: u64, -} - -impl ForwardLogReader { - pub fn new(metadata_file: fs::File, data_file: fs::File) -> ForwardLogReader { - let mut ret = ForwardLogReader { - metadata_reader: io::BufReader::new(metadata_file), - data_reader: io::BufReader::new(data_file), - }; - - ret.metadata_reader - .seek(io::SeekFrom::Start(METADATA_FILE_HEADER_SIZE as u64)) - .expect("Seek failed"); - - ret - } - - pub fn new_with_index( - metadata_file: fs::File, - data_file: fs::File, - index: u64, - ) -> ForwardLogReader { - let mut ret = ForwardLogReader { - metadata_reader: io::BufReader::new(metadata_file), - data_reader: io::BufReader::new(data_file), - }; - - ret.metadata_reader - .seek(io::SeekFrom::Start( - METADATA_FILE_HEADER_SIZE as u64 + METADATA_ROW_LENGTH as u64 * index, - )) - .expect("Seek failed"); - - ret - } - - fn read_record(&mut self) -> Result, io::Error> { - loop { - let pos = self.metadata_reader.stream_position()?; - let index = (pos - METADATA_FILE_HEADER_SIZE as u64) / METADATA_ROW_LENGTH as u64; - - let mut metadata_entry_buf = vec![0; 16]; // 2x u64 - if let Err(e) = self.metadata_reader.read_exact(&mut metadata_entry_buf) { - if e.kind() == io::ErrorKind::UnexpectedEof { - return Ok(None); - } else { - return Err(e); - } - } - - // First u64 is the offset of the record in the data file, second is the length of the record - let entry_offset = u64::from_be_bytes(metadata_entry_buf[0..8].try_into().unwrap()); - let entry_length = u64::from_be_bytes(metadata_entry_buf[8..16].try_into().unwrap()); - - if entry_offset == 0 && entry_length == 0 { - // This is an unused entry in the metadata file, skip - continue; - } - - // Use .seek_relative instead of .seek to avoid dropping the BufReader internal buffer when - // the seek distance is small - let seek_distance = entry_offset as i64 - self.data_reader.stream_position()? as i64; - self.data_reader.seek_relative(seek_distance)?; - - let mut result_buf = vec![0; entry_length as usize]; - self.data_reader.read_exact(&mut result_buf)?; - - let row = Row::deserialize(&result_buf); - return Ok(Some(ForwardLogReaderItem { row, index })); - } - } -} - -impl Iterator for ForwardLogReader { - type Item = ForwardLogReaderItem; - - fn next(&mut self) -> Option { - self.read_record().unwrap_or_else(|err| { - panic!("Error reading record: {:?}", err); - }) - } -} - -#[cfg(test)] -mod tests { - use ctor::ctor; - use env_logger; - - use super::*; - - #[ctor] - fn init_logger() { - let _ = env_logger::builder().is_test(true).try_init(); - } - - const TEST_RESOURCES_DIR: &str = "tests/resources"; - - #[test] - fn test_forward_log_reader_fixture_db1() { - let metadata_path = Path::new(TEST_RESOURCES_DIR).join("test_metadata_1"); - let data_path = Path::new(TEST_RESOURCES_DIR).join("test_data_1"); - let metadata_file = fs::OpenOptions::new() - .read(true) - .open(&metadata_path) - .expect("Failed to open metadata file"); - let data_file = fs::OpenOptions::new() - .read(true) - .open(&data_path) - .expect("Failed to open data file"); - - let mut forward_log_reader = ForwardLogReader::new(metadata_file, data_file); - - // There are two records in the log with "schema" with one field: Bytes - - let ForwardLogReaderItem { row, index: _ } = forward_log_reader - .next() - .expect("Failed to read the first record"); - assert!(match &row.values[..] { - [Value::Bytes(bytes)] => bytes.len() == 256, - _ => false, - }); - - assert!(forward_log_reader.next().is_none()); - } -} diff --git a/log_db/src/memtable_primary.rs b/log_db/src/memtable_primary.rs deleted file mode 100644 index 573592e..0000000 --- a/log_db/src/memtable_primary.rs +++ /dev/null @@ -1,40 +0,0 @@ -use super::*; -use std::collections::BTreeMap; - -pub struct PrimaryMemtable { - /// Map of records indexed by key. Used as a shared heap of records - /// for all secondary memtables also. Secondary memtables store an - /// IndexableValue as their record value, which is used to get - /// the actual record from the primary memtable `records` map. - /// - /// Note: it must be invariant that all memtables (primary and secondary) - /// contain the same keys. - records: BTreeMap, -} - -impl PrimaryMemtable { - pub fn new() -> PrimaryMemtable { - PrimaryMemtable { - records: BTreeMap::new(), - } - } - - pub fn set(&mut self, key: IndexableValue, value: LogKey) { - self.records.insert(key, value); - } - - pub fn get(&self, key: &IndexableValue) -> Option<&LogKey> { - self.records.get(key) - } - - pub fn remove(&mut self, key: &IndexableValue) -> Option { - self.records.remove(key) - } - - pub fn range>(&self, range: B) -> Vec<&LogKey> { - self.records - .range(range) - .map(|(_, log_key)| log_key) - .collect() - } -} diff --git a/log_db/src/memtable_secondary.rs b/log_db/src/memtable_secondary.rs deleted file mode 100644 index e1ca835..0000000 --- a/log_db/src/memtable_secondary.rs +++ /dev/null @@ -1,66 +0,0 @@ -use once_cell::sync::Lazy; - -use super::*; -use std::collections::{btree_map::Values, BTreeMap}; - -pub struct SecondaryMemtable { - /// A 2-layer map of records indexed by SK => PK => LogKey. - /// The PK information is required to tell two records apart. - records: BTreeMap, -} - -static EMPTY_MAP: Lazy> = Lazy::new(|| BTreeMap::new()); - -impl SecondaryMemtable { - pub fn new() -> SecondaryMemtable { - SecondaryMemtable { - records: BTreeMap::new(), - } - } - - pub fn set(&mut self, pk: IndexableValue, sk: IndexableValue, value: LogKey) { - match self.records.get_mut(&sk) { - Some(map) => { - map.insert(pk, value); - } - None => { - self.records - .insert(sk, LogKeyMap::new_with_initial(pk, value)); - } - }; - } - - pub fn find_by(&self, key: &IndexableValue) -> Values { - match self.records.get(key) { - Some(set) => set.log_keys(), - None => EMPTY_MAP.values(), - } - } - - // Remove a single mapping associated with the given PK and SK. Returns `true` - // if the log key existed and was removed, `false` otherwise. - pub fn remove(&mut self, pk: &IndexableValue, sk: &IndexableValue) -> bool { - let map = match self.records.get_mut(sk) { - Some(set) => set, - None => return false, - }; - if map.len() == 1 && map.contains_pk(pk) { - self.records.remove(sk); - true - } else { - return match map.remove_pk(pk) { - Ok(_) => true, - Err(LogKeyMapError::NotFoundError) => false, - Err(e) => panic!("{:?}", e), - }; - } - } - - pub fn range>(&self, range: B) -> Vec<&LogKey> { - let mut keys = Vec::new(); - for (_, map) in self.records.range(range) { - keys.extend(map.log_keys()); - } - keys - } -} diff --git a/log_db/src/record.rs b/log_db/src/record.rs deleted file mode 100644 index b349853..0000000 --- a/log_db/src/record.rs +++ /dev/null @@ -1,38 +0,0 @@ -use super::*; - -pub struct Record { - values: Vec, -} - -impl Record { - pub fn values(&self) -> &[Value] { - &self.values - } -} - -impl IntoIterator for Record { - type Item = Value; - type IntoIter = std::vec::IntoIter; - - fn into_iter(self) -> Self::IntoIter { - self.values.into_iter() - } -} - -impl From> for Record { - fn from(values: Vec) -> Self { - Record { values } - } -} - -impl From for Vec { - fn from(record: Record) -> Self { - record.values - } -} - -impl From for Record { - fn from(row: Row) -> Self { - Record { values: row.values } - } -} diff --git a/log_db/src/row.rs b/log_db/src/row.rs deleted file mode 100644 index d8140d2..0000000 --- a/log_db/src/row.rs +++ /dev/null @@ -1,70 +0,0 @@ -use super::*; - -#[derive(Debug, Clone)] -pub struct Row { - pub values: Vec, - pub tombstone: bool, -} - -impl Row { - pub fn serialize(&self) -> Vec { - let mut bytes = Vec::new(); - - if self.tombstone { - bytes.extend(&[B_TOMBSTONE]); - } else { - bytes.extend(&[B_LIVE]); - } - - for value in &self.values { - bytes.extend(value.serialize()); - } - bytes - } - - pub fn deserialize(bytes: &[u8]) -> Row { - assert!(bytes.len() > 0); - - let mut values = Vec::new(); - - let tombstone = bytes[0] == B_TOMBSTONE; - - let mut start = 1; - while start < bytes.len() { - let (rv, consumed) = Value::deserialize(&bytes[start..]); - values.push(rv); - start += consumed; - } - Row { values, tombstone } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_record_serialize_deserialize() { - let record = Row { - values: vec![ - Value::Int(1), - Value::String("hello".to_string()), - Value::Bytes(vec![0, 1, 2, 3]), - ], - tombstone: true, - }; - - let serialized = record.serialize(); - let deserialized = Row::deserialize(&serialized); - let reserialized = deserialized.serialize(); - - assert_eq!(serialized.len(), reserialized.len()); - assert_eq!(record.values, deserialized.values); - } -} - -#[derive(Clone, Debug)] -pub enum TxEntry { - Upsert { row: Row }, - Delete { row: Row }, -} diff --git a/log_db/src/schema.rs b/log_db/src/schema.rs deleted file mode 100644 index 98d0df7..0000000 --- a/log_db/src/schema.rs +++ /dev/null @@ -1,7 +0,0 @@ -use super::*; - -pub struct Schema { - pub fields: Vec, - pub primary_key: String, - pub secondary_keys: Vec, -} diff --git a/log_db/tests/integration.rs b/log_db/tests/integration.rs deleted file mode 100644 index 2012faf..0000000 --- a/log_db/tests/integration.rs +++ /dev/null @@ -1,1012 +0,0 @@ -#[macro_use] -extern crate log; -extern crate ctor; -extern crate tempfile; - -use ctor::ctor; -use env_logger; -use log_db::*; -use serial_test::serial; -use std::fs::{self}; -use std::path::Path; -use std::thread; -use std::time::Duration; -use tempfile::tempdir; - -pub fn tmp_dir() -> String { - let dir = tempdir() - .expect("Failed to create temporary directory") - .path() - .to_str() - .expect("Failed to convert temporary directory path to string") - .to_string(); - fs::create_dir_all(&dir).expect("Failed to create temporary directory"); - dir -} - -#[ctor] -fn init_logger() { - let _ = env_logger::builder().is_test(true).try_init(); -} - -#[derive(Eq, PartialEq, Clone, Debug)] -enum Field { - Id, - Name, - Data, -} - -impl AsRef for Field { - fn as_ref(&self) -> &str { - match self { - Field::Id => "id", - Field::Name => "name", - Field::Data => "data", - } - } -} - -impl From for String { - fn from(field: Field) -> String { - field.as_ref().to_owned() - } -} - -#[derive(Debug)] -struct Inst { - pub id: i64, - pub name: Option, - pub data: Vec, -} - -impl From for Record { - fn from(inst: Inst) -> Record { - vec![ - Value::Int(inst.id), - match inst.name { - Some(name) => Value::String(name), - None => Value::Null, - }, - Value::Bytes(inst.data), - ] - .into() - } -} - -impl From for Inst { - fn from(record: Record) -> Self { - let mut it = record.into_iter(); - Inst { - id: match it.next().unwrap() { - Value::Int(id) => id, - other => panic!("Invalid value type: {:?}", other), - }, - name: match it.next().unwrap() { - Value::String(name) => Some(name), - Value::Null => None, - other => panic!("Invalid value type: {:?}", other), - }, - data: match it.next().unwrap() { - Value::Bytes(data) => data, - other => panic!("Invalid value type: {:?}", other), - }, - } - } -} - -impl Inst { - fn schema() -> Vec { - vec![Field::Id, Field::Name, Field::Data] - } - fn primary_key() -> Field { - Field::Id - } - fn secondary_keys() -> Vec { - vec![Field::Name] - } -} - -#[test] -fn test_initialize_only() { - let data_dir = tmp_dir(); - let _db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); -} - -#[test] -fn test_upsert_and_get_with_primary_memtable() { - let data_dir = tmp_dir(); - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - let id = 1; - let inst = Inst { - id, - name: Some("Alice".to_string()), - data: vec![0, 1, 2], - }; - db.upsert(inst).unwrap(); - - let result: Inst = db.get(&Value::Int(1)).unwrap().unwrap().into(); - - // Check that the IDs match - assert!(result.id == id); -} - -#[test] -fn test_upsert_and_get() { - let data_dir = tmp_dir(); - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - // Insert some records - db.upsert(Inst { - id: 0, - name: None, - data: vec![3, 4, 5], - }) - .unwrap(); - - db.upsert(Inst { - id: 1, - name: Some("Alice".to_string()), - data: vec![0, 1, 2], - }) - .unwrap(); - - db.upsert(Inst { - id: 1, - name: Some("Bob".to_string()), - data: vec![0, 1, 2], - }) - .unwrap(); - - db.upsert(Inst { - id: 2, - name: Some("George".to_string()), - data: vec![], - }) - .unwrap(); - - // Get with ID = 0 - let result: Inst = db.get(&Value::Int(0)).unwrap().unwrap().into(); - - // Should match id == 0 - assert!(result.id == 0); - assert!(result.name == None); - - // Get with ID = 1 - let result: Inst = db.get(&Value::Int(1)).unwrap().unwrap().into(); - - // Should match newest inst with id == 1 - assert!(result.id == 1); - assert!(result.name == Some("Bob".to_owned())); -} - -#[test] -fn test_get_nonexistant() { - let data_dir = tmp_dir(); - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - let result = db.get(&Value::Int(0)).unwrap(); - assert!(result.is_none()); -} - -#[test] -fn test_upsert_and_find_by() { - let data_dir = tmp_dir(); - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - // Insert some records - db.upsert(Inst { - id: 0, - name: Some("John".to_string()), - data: vec![3, 4, 5], - }) - .unwrap(); - - db.upsert(Inst { - id: 1, - name: Some("John".to_string()), - data: vec![1, 2, 3], - }) - .unwrap(); - - db.upsert(Inst { - id: 2, - name: Some("George".to_string()), - data: vec![1, 2, 3], - }) - .unwrap(); - - // There should be 2 Johns - let johns = db - .find_by(&Field::Name, &Value::String("John".to_string())) - .expect("Failed to find all Johns"); - - assert_eq!(johns.len(), 2); -} - -struct InstSingleId { - pub id: i64, -} - -impl From for Record { - fn from(inst: InstSingleId) -> Record { - vec![Value::Int(inst.id)].into() - } -} - -impl From for InstSingleId { - fn from(record: Record) -> Self { - let mut it = record.into_iter(); - InstSingleId { - id: match it.next().unwrap() { - Value::Int(id) => id, - other => panic!("Invalid value type: {:?}", other), - }, - } - } -} - -impl InstSingleId { - fn schema() -> Vec { - vec![Field::Id] - } - fn primary_key() -> Field { - Field::Id - } - fn secondary_keys() -> Vec { - vec![] - } -} - -#[test] -#[serial] -fn test_multiple_writing_threads() { - let data_dir = tmp_dir(); - debug!("Data dir: {:?}", data_dir); - let mut threads = vec![]; - let threads_n = 100; - - for i in 0..threads_n { - let data_dir = data_dir.clone(); - threads.push(thread::spawn(move || { - let mut db = DB::configure() - .fields(InstSingleId::schema()) - .primary_key(InstSingleId::primary_key()) - .secondary_keys(InstSingleId::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - db.upsert(InstSingleId { id: i }) - .expect("Failed to upsert record"); - })); - } - - for thread in threads { - thread.join().expect("Failed to join thread"); - } - - // Read the records - let mut db = DB::configure() - .fields(InstSingleId::schema()) - .primary_key(InstSingleId::primary_key()) - .secondary_keys(InstSingleId::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - for i in 0..threads_n { - let result: InstSingleId = db - .get(&Value::Int(i)) - .expect("Failed to get record") - .expect("Record not found") - .into(); - - assert!(result.id == i); - } -} - -#[test] -#[serial] -fn test_one_writer_and_multiple_reading_threads() { - let data_dir = tmp_dir(); - let mut threads = vec![]; - let threads_n = 100; - - // Add readers that poll for the records - for i in 0..threads_n { - let data_dir = data_dir.clone(); - threads.push(thread::spawn(move || { - let mut db = DB::configure() - .fields(InstSingleId::schema()) - .primary_key(InstSingleId::primary_key()) - .secondary_keys(InstSingleId::secondary_keys()) - .data_dir(&data_dir) - .segment_size(1000) // should cause rotations - .initialize() - .expect("Failed to initialize DB instance"); - - let mut timeout = 5; - loop { - let result = db.get(&Value::Int(i)).expect("Failed to get record"); - match result { - None => { - thread::sleep(Duration::from_millis(timeout)); - timeout = std::cmp::min(timeout * 2, 100); - continue; - } - Some(result) => { - let result: InstSingleId = result.into(); - assert!(result.id == i); - break; - } - }; - } - })); - } - - // Add a writer that inserts the records - threads.push(thread::spawn(move || { - let mut db = DB::configure() - .fields(InstSingleId::schema()) - .primary_key(InstSingleId::primary_key()) - .secondary_keys(InstSingleId::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - for i in 0..threads_n { - db.upsert(InstSingleId { id: i }) - .expect("Failed to upsert record"); - - db.do_maintenance_tasks() // Run maintenance tasks after every write, just to test it - .expect("Failed to do maintenance tasks"); - } - })); - - for thread in threads { - thread.join().expect("Failed to join thread"); - } -} - -#[test] -fn test_log_is_rotated_when_capacity_reached() { - let data_dir = tmp_dir(); - let data_dir_path = Path::new(&data_dir); - - // Hand-calculated record length, find record below - let record_len = 1 // tombstone tag - + (1 + 8) // int tag + i64 - + (1 + 8 + 4) // string tag + string length + string data - + (1 + 8 + 3); // bytes tag + bytes length + bytes data - - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .segment_size(10 * record_len) // small log segment size - .initialize() - .expect("Failed to initialize DB instance"); - - // Insert more records than fits the capacity - for _ in 0..25 { - db.upsert(Inst { - id: 0, - name: Some("John".to_string()), - data: vec![3, 4, 5], - }) - .expect("Failed to upsert record"); - - db.do_maintenance_tasks() - .expect("Failed to do maintenance tasks"); - } - - // Check that the rotated segments exist - assert!(data_dir_path.join("metadata").with_extension("1").exists()); - assert!(data_dir_path.join("metadata").with_extension("2").exists()); - - // 3rd segment should not exist (note negation) - assert!(!data_dir_path.join("metadata").with_extension("3").exists()); -} - -#[test] -fn test_delete() { - let data_dir = tmp_dir(); - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - // Insert some records - db.upsert(Inst { - id: 0, - name: Some("John".to_string()), - data: vec![3, 4, 5], - }) - .unwrap(); - - db.upsert(Inst { - id: 1, - name: Some("John".to_string()), - data: vec![1, 2, 3], - }) - .unwrap(); - - db.delete(&Value::Int(0)).unwrap(); - - // Check that the record is deleted - assert!(db.get(&Value::Int(0)).unwrap().is_none()); - - // Check that the secondary index is updated - assert_eq!( - db.find_by(&Field::Name, &Value::String("John".to_string())) - .unwrap() - .len(), - 1 - ); -} - -#[test] -fn test_delete_by() { - let data_dir = tmp_dir(); - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - // Insert some records - db.upsert(Inst { - id: 0, - name: Some("John".to_string()), - data: vec![3, 4, 5], - }) - .unwrap(); - - db.upsert(Inst { - id: 1, - name: Some("John".to_string()), - data: vec![1, 2, 3], - }) - .unwrap(); - - db.upsert(Inst { - id: 2, - name: Some("Bob".to_string()), - data: vec![1, 2, 3], - }) - .unwrap(); - - db.delete_by(&Field::Name, &Value::String("John".to_string())) - .unwrap(); - - // Check that the record is deleted - assert!(db.get(&Value::Int(0)).unwrap().is_none()); - assert!(db.get(&Value::Int(1)).unwrap().is_none()); - - // Check that the secondary index is updated - assert_eq!( - db.find_by(&Field::Name, &Value::String("John".to_string())) - .unwrap() - .len(), - 0 - ); - assert_eq!( - db.find_by(&Field::Name, &Value::String("Bob".to_string())) - .unwrap() - .len(), - 1 - ); -} - -#[test] -fn test_range_by_id() { - let data_dir = tmp_dir(); - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - let inserted_ids = vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9]; - - for id in inserted_ids.iter() { - db.upsert(Inst { - id: *id, - name: Some("Foobar".to_string()), - data: vec![], - }) - .unwrap(); - } - - // Test range [3, 7) - let received = db - .range_by(&Field::Id, &Value::Int(3)..&Value::Int(7)) - .unwrap(); - let received_ids: Vec = received.into_iter().map(|rec| Inst::from(rec).id).collect(); - - assert_eq!(received_ids, vec![3, 4, 5, 6]); - - // Test range [3, 7] - let received = db - .range_by(&Field::Id, &Value::Int(3)..=&Value::Int(7)) - .unwrap(); - let received_ids: Vec = received.into_iter().map(|rec| Inst::from(rec).id).collect(); - - assert_eq!(received_ids, vec![3, 4, 5, 6, 7]); - - // Test range (-inf, 7] - let received = db.range_by(&Field::Id, ..=&Value::Int(7)).unwrap(); - let received_ids: Vec = received.into_iter().map(|rec| Inst::from(rec).id).collect(); - - assert_eq!(received_ids, vec![0, 1, 2, 3, 4, 5, 6, 7]); - - // Test range (3, inf) - let received = db.range_by(&Field::Id, &Value::Int(3)..).unwrap(); - let received_ids: Vec = received.into_iter().map(|rec| Inst::from(rec).id).collect(); - - assert_eq!(received_ids, vec![3, 4, 5, 6, 7, 8, 9]); -} - -#[test] -fn test_batch_find_by() { - let data_dir = tmp_dir(); - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - let inserted_ids = vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9]; - - for id in inserted_ids.iter() { - db.upsert(Inst { - id: *id, - name: Some("Foobar".to_string()), - data: vec![], - }) - .unwrap(); - } - - let batch: Vec = (2..5).map(Value::Int).collect(); - let result = db.batch_find_by(Field::Id, &batch).unwrap(); - - assert_eq!(result.len(), batch.len()); - assert_eq!( - result.iter().map(|(tag, _)| *tag).collect::>(), - vec![0, 1, 2] - ); - assert_eq!( - result - .into_iter() - .map(|(_, rec)| Inst::from(rec).id) - .collect::>(), - vec![2, 3, 4] - ); -} - -#[test] -fn test_commit_transaction() { - let data_dir = tmp_dir(); - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - db.tx_begin().expect("Failed to begin transaction"); - - db.upsert(Inst { - id: 0, - name: Some("John".to_string()), - data: vec![3, 4, 5], - }) - .unwrap(); - - db.upsert(Inst { - id: 1, - name: Some("John".to_string()), - data: vec![1, 2, 3], - }) - .unwrap(); - - db.tx_commit().expect("Failed to commit transaction"); - - let johns = db - .find_by(&Field::Name, &Value::String("John".to_string())) - .unwrap(); - - assert_eq!(johns.len(), 2); -} - -#[test] -fn test_rollback_transaction() { - let data_dir = tmp_dir(); - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - db.tx_begin().expect("Failed to begin transaction"); - - db.upsert(Inst { - id: 0, - name: Some("John".to_string()), - data: vec![3, 4, 5], - }) - .unwrap(); - - db.upsert(Inst { - id: 1, - name: Some("John".to_string()), - data: vec![1, 2, 3], - }) - .unwrap(); - - db.tx_rollback().expect("Failed to roll back transaction"); - - let johns = db - .find_by(&Field::Name, &Value::String("John".to_string())) - .unwrap(); - - assert_eq!(johns.len(), 0); -} - -#[derive(Eq, PartialEq, Clone, Debug)] -enum FieldWithNewNullableField { - Id, - Name, - Data, - MaybeStr, -} - -impl AsRef for FieldWithNewNullableField { - fn as_ref(&self) -> &str { - match self { - FieldWithNewNullableField::Id => "id", - FieldWithNewNullableField::Name => "name", - FieldWithNewNullableField::Data => "data", - FieldWithNewNullableField::MaybeStr => "maybe_str", - } - } -} - -impl Into for FieldWithNewNullableField { - fn into(self) -> String { - self.as_ref().to_owned() - } -} - -struct InstWithNewNullableField { - pub id: i64, - pub name: Option, - pub data: Vec, - pub maybe_str: Option, -} - -impl From for Record { - fn from(inst: InstWithNewNullableField) -> Record { - vec![ - Value::Int(inst.id), - match inst.name { - Some(name) => Value::String(name), - None => Value::Null, - }, - Value::Bytes(inst.data), - match inst.maybe_str { - Some(maybe_str) => Value::String(maybe_str), - None => Value::Null, - }, - ] - .into() - } -} - -impl From for InstWithNewNullableField { - fn from(record: Record) -> Self { - let mut it = record.into_iter(); - InstWithNewNullableField { - id: match it.next().unwrap() { - Value::Int(id) => id, - other => panic!("Invalid value type: {:?}", other), - }, - name: match it.next().unwrap() { - Value::String(name) => Some(name), - Value::Null => None, - other => panic!("Invalid value type: {:?}", other), - }, - data: match it.next().unwrap() { - Value::Bytes(data) => data, - other => panic!("Invalid value type: {:?}", other), - }, - maybe_str: match it.next() { - Some(Value::String(maybe_str)) => Some(maybe_str), - Some(Value::Null) => None, - None => None, - other => panic!("Invalid value type: {:?}", other), - }, - } - } -} - -impl InstWithNewNullableField { - fn schema() -> Vec { - vec![ - FieldWithNewNullableField::Id, - FieldWithNewNullableField::Name, - FieldWithNewNullableField::Data, - FieldWithNewNullableField::MaybeStr, - ] - } - - fn primary_key() -> FieldWithNewNullableField { - FieldWithNewNullableField::Id - } - - fn secondary_keys() -> Vec { - vec![FieldWithNewNullableField::Name] - } -} - -#[test] -fn test_add_nullable_field() { - let data_dir = tmp_dir(); - - // Insert a record with 3 fields - { - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - db.upsert(Inst { - id: 0, - name: Some("John".to_string()), - data: vec![3, 4, 5], - }) - .unwrap(); - } - - // Insert a record with 4 fields (last is nullable) - let mut db = DB::configure() - .fields(InstWithNewNullableField::schema()) - .primary_key(InstWithNewNullableField::primary_key()) - .secondary_keys(InstWithNewNullableField::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - db.upsert(InstWithNewNullableField { - id: 1, - name: Some("John".to_string()), - data: vec![3, 4, 5], - maybe_str: None, - }) - .unwrap(); - - let johns = db - .find_by( - &FieldWithNewNullableField::Name, - &Value::String("John".to_string()), - ) - .unwrap(); - - assert_eq!(johns.len(), 2); -} - -#[test] -fn test_delete_by_multiple_indexes() { - let data_dir = tmp_dir(); - - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - // Insert some identical records - for _ in 0..10 { - db.upsert(Inst { - id: 0, - name: Some("foo".to_string()), - data: vec![], - }) - .unwrap(); - } - - // Delete by name - db.delete_by(&Field::Name, &Value::String("foo".to_string())) - .unwrap(); - - // Check that the records are deleted by finding by name - let result = db - .find_by(&Field::Name, &Value::String("foo".to_string())) - .unwrap(); - assert_eq!(result.len(), 0); - - // Double check with find by id - let result = db.find_by(&Field::Id, &Value::Int(0)).unwrap(); - assert_eq!(result.len(), 0); -} - -#[test] -fn test_find_by_with_offset_and_limit() { - let data_dir = tmp_dir(); - - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - // Insert some records - for i in 0..10 { - db.upsert(Inst { - id: i, - name: Some("foo".to_string()), - data: vec![], - }) - .unwrap(); - } - - // Find by name with offset and limit - let result = db - .find_by_with_params( - &Field::Name, - &Value::String("foo".to_string()), - &QueryParams { - offset: 2, - limit: 3, - sort_asc: true, - }, - ) - .unwrap() - .into_iter() - .map(|rec| Inst::from(rec)) - .collect::>(); - - assert_eq!(result.len(), 3); - assert_eq!(result[0].id, 2); - assert_eq!(result[1].id, 3); - assert_eq!(result[2].id, 4); -} - -#[test] -fn test_batch_find_by_with_offset_and_limit() { - let data_dir = tmp_dir(); - - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - // Insert some records - for i in 0..10 { - db.upsert(Inst { - id: i, - name: Some("foo".to_string()), - data: vec![], - }) - .unwrap(); - } - - // Batch find by id with offset and limit - let batch: Vec = (2..5).map(Value::Int).collect(); - let result = db - .batch_find_by_with_params( - &Field::Id, - &batch, - &QueryParams { - offset: 1, - limit: 2, - sort_asc: true, - }, - ) - .unwrap() - .into_iter() - .map(|(_, rec)| Inst::from(rec)) - .collect::>(); - - assert_eq!(result.len(), 2); - assert_eq!(result[0].id, 3); - assert_eq!(result[1].id, 4); -} - -#[test] -fn test_range_by_with_offset_and_limit() { - let data_dir = tmp_dir(); - - let mut db = DB::configure() - .fields(Inst::schema()) - .primary_key(Inst::primary_key()) - .secondary_keys(Inst::secondary_keys()) - .data_dir(&data_dir) - .initialize() - .expect("Failed to initialize DB instance"); - - // Insert some records - for i in 0..10 { - db.upsert(Inst { - id: i, - name: Some("foo".to_string()), - data: vec![], - }) - .unwrap(); - } - - // Range by id with offset and limit - let result = db - .range_by_with_params( - &Field::Id, - &Value::Int(2)..&Value::Int(8), - &QueryParams { - offset: 1, - limit: 3, - sort_asc: true, - }, - ) - .unwrap() - .into_iter() - .map(|rec| Inst::from(rec)) - .collect::>(); - - assert_eq!(result.len(), 3); - assert_eq!(result[0].id, 3); - assert_eq!(result[1].id, 4); - assert_eq!(result[2].id, 5); -} diff --git a/log_db/tests/resources/test_data_1 b/log_db/tests/resources/test_data_1 deleted file mode 100644 index 5365384..0000000 Binary files a/log_db/tests/resources/test_data_1 and /dev/null differ diff --git a/log_db/tests/resources/test_metadata_1 b/log_db/tests/resources/test_metadata_1 deleted file mode 100644 index 98dd35e..0000000 Binary files a/log_db/tests/resources/test_metadata_1 and /dev/null differ diff --git a/py_bindings/Cargo.toml b/py_bindings/Cargo.toml index a84ccd6..1ddc264 100644 --- a/py_bindings/Cargo.toml +++ b/py_bindings/Cargo.toml @@ -7,6 +7,6 @@ edition = "2021" crate-type = ["lib"] [dependencies] -log_db = { path = "../log_db" } +autere_db = { path = "../autere_db" } pyo3 = "0.22.3" rust_decimal = { version = "1.36.0", features = [] } diff --git a/py_bindings/example.py b/py_bindings/example.py index 233d111..96040a9 100644 --- a/py_bindings/example.py +++ b/py_bindings/example.py @@ -1,8 +1,8 @@ import tempfile from pprint import pformat, pprint -from log_db import Value, Bound, DB +from autere_db import Value, Bound, DB -# TODO: This should be imported from the log_db module +# TODO: This should be imported from the autere_db module # but exporting type aliases does not work automatically Record = list[Value] @@ -40,7 +40,7 @@ db = DB \ db.upsert(Inst(1, "foo").into_record()) -#res = db.find_by("name", log_db.Value.string("foo")) +#res = db.find_by("name", autere_db.Value.string("foo")) #insts = [Inst.from_record(r) for r in res] res = db.range_by("name", Bound.unbounded(), diff --git a/py_bindings/pyproject.toml b/py_bindings/pyproject.toml index 6f5b437..113464a 100644 --- a/py_bindings/pyproject.toml +++ b/py_bindings/pyproject.toml @@ -3,7 +3,7 @@ requires = ["maturin>=1.7,<2.0"] build-backend = "maturin" [project] -name = "log_db" +name = "autere_db" requires-python = ">=3.8" classifiers = [ "Programming Language :: Rust", diff --git a/py_bindings/src/lib.rs b/py_bindings/src/lib.rs index 4e125ec..5a1c26a 100644 --- a/py_bindings/src/lib.rs +++ b/py_bindings/src/lib.rs @@ -1,7 +1,7 @@ use std::str::FromStr; use std::usize; -use log_db::{self, OwnedBounds, QueryParams, Record, DEFAULT_QUERY_PARAMS}; +use autere_db::{self, OwnedBounds, QueryParams, Record, DEFAULT_QUERY_PARAMS}; use pyo3::exceptions::PyException; use pyo3::prelude::*; use rust_decimal::Decimal; @@ -20,8 +20,8 @@ pub const READ_CONSISTENCY_STRONG: u8 = 1; struct Config { data_dir: Option, segment_size: Option, - write_durability: Option, - read_consistency: Option, + write_durability: Option, + read_consistency: Option, fields: Option>, primary_key: Option, secondary_keys: Option>, @@ -50,8 +50,8 @@ impl Config { write_durability: u8, ) -> PyResult> { slf.write_durability = Some(match write_durability { - WRITE_DURABILITY_FLUSH => log_db::WriteDurability::Flush, - WRITE_DURABILITY_FLUSH_SYNC => log_db::WriteDurability::FlushSync, + WRITE_DURABILITY_FLUSH => autere_db::WriteDurability::Flush, + WRITE_DURABILITY_FLUSH_SYNC => autere_db::WriteDurability::FlushSync, _ => { return Err(PyException::new_err(format!( "Invalid write_durability value: {}", @@ -67,8 +67,8 @@ impl Config { read_consistency: u8, ) -> PyResult> { slf.read_consistency = Some(match read_consistency { - READ_CONSISTENCY_EVENTUAL => log_db::ReadConsistency::Eventual, - READ_CONSISTENCY_STRONG => log_db::ReadConsistency::Strong, + READ_CONSISTENCY_EVENTUAL => autere_db::ReadConsistency::Eventual, + READ_CONSISTENCY_STRONG => autere_db::ReadConsistency::Strong, _ => { return Err(PyException::new_err(format!( "Invalid read_consistency value: {}", @@ -104,7 +104,7 @@ impl Config { } pub fn initialize(&self) -> PyResult { - let mut config = log_db::DB::configure(); + let mut config = autere_db::DB::configure(); if self.data_dir.is_some() { config = config.data_dir(&self.data_dir.as_ref().unwrap().to_string()); } @@ -157,7 +157,7 @@ fn py_into_record(record: Vec) -> Record { record .into_iter() .map(|value| value.record_value) - .collect::>() + .collect::>() .into() } @@ -170,34 +170,34 @@ const VALUE_NULL: u8 = 4; #[pyclass] #[derive(Clone, PartialEq, Eq)] pub struct Value { - record_value: log_db::Value, + record_value: autere_db::Value, } #[pymethods] impl Value { fn __repr__(&self) -> String { match &self.record_value { - log_db::Value::Int(value) => format!("Value.int({})", value), - log_db::Value::Decimal(value) => format!("Value.decimal({})", value), - log_db::Value::String(value) => { + autere_db::Value::Int(value) => format!("Value.int({})", value), + autere_db::Value::Decimal(value) => format!("Value.decimal({})", value), + autere_db::Value::String(value) => { format!("Value.string(\"{}\")", value.replace("\"", "\\\"")) } - log_db::Value::Bytes(value) => format!("Value.bytes({:?})", value), - log_db::Value::Null => "Value.null()".to_string(), + autere_db::Value::Bytes(value) => format!("Value.bytes({:?})", value), + autere_db::Value::Null => "Value.null()".to_string(), } } #[staticmethod] fn int(value: i64) -> Self { Value { - record_value: log_db::Value::Int(value), + record_value: autere_db::Value::Int(value), } } #[staticmethod] fn decimal(value: String) -> Self { Value { - record_value: log_db::Value::Decimal( + record_value: autere_db::Value::Decimal( Decimal::from_str(&value).expect(&format!("Invalid Decimal: {}", value)), ), } @@ -206,65 +206,65 @@ impl Value { #[staticmethod] fn string(value: String) -> Self { Value { - record_value: log_db::Value::String(value), + record_value: autere_db::Value::String(value), } } #[staticmethod] fn bytes(value: &[u8]) -> Self { Value { - record_value: log_db::Value::Bytes(value.to_vec()), + record_value: autere_db::Value::Bytes(value.to_vec()), } } #[staticmethod] fn null() -> Self { Value { - record_value: log_db::Value::Null, + record_value: autere_db::Value::Null, } } pub fn kind(&self) -> u8 { match &self.record_value { - log_db::Value::Int(_) => VALUE_INT, - log_db::Value::Decimal(_) => VALUE_DECIMAL, - log_db::Value::String(_) => VALUE_STRING, - log_db::Value::Bytes(_) => VALUE_BYTES, - log_db::Value::Null => VALUE_NULL, + autere_db::Value::Int(_) => VALUE_INT, + autere_db::Value::Decimal(_) => VALUE_DECIMAL, + autere_db::Value::String(_) => VALUE_STRING, + autere_db::Value::Bytes(_) => VALUE_BYTES, + autere_db::Value::Null => VALUE_NULL, } } pub fn as_int(&self) -> PyResult { match &self.record_value { - log_db::Value::Int(value) => Ok(*value), + autere_db::Value::Int(value) => Ok(*value), _ => Err(PyException::new_err("Value is not an Int")), } } pub fn as_decimal(&self) -> PyResult { match &self.record_value { - log_db::Value::Decimal(value) => Ok(value.to_string()), + autere_db::Value::Decimal(value) => Ok(value.to_string()), _ => Err(PyException::new_err("Value is not a Decimal")), } } pub fn as_string(&self) -> PyResult { match &self.record_value { - log_db::Value::String(value) => Ok(value.clone()), + autere_db::Value::String(value) => Ok(value.clone()), _ => Err(PyException::new_err("Value is not a String")), } } pub fn as_bytes(&self) -> PyResult> { match &self.record_value { - log_db::Value::Bytes(value) => Ok(value.clone()), + autere_db::Value::Bytes(value) => Ok(value.clone()), _ => Err(PyException::new_err("Value is not Bytes")), } } pub fn as_null(&self) -> PyResult<()> { match &self.record_value { - log_db::Value::Null => Ok(()), + autere_db::Value::Null => Ok(()), _ => Err(PyException::new_err("Value is not Null")), } } @@ -272,7 +272,7 @@ impl Value { #[pyclass] struct DB { - db: log_db::DB, + db: autere_db::DB, } #[pymethods] @@ -349,7 +349,7 @@ impl DB { limit, sort_asc, }; - let keys: Vec = keys.into_iter().map(|key| key.record_value).collect(); + let keys: Vec = keys.into_iter().map(|key| key.record_value).collect(); let recs = self .db .batch_find_by_with_params(&field, &keys, ¶ms) @@ -480,7 +480,7 @@ impl PyRangeBound { } } -#[pymodule(name = "log_db")] +#[pymodule(name = "autere_db")] fn log_db_py(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; -- cgit v1.3