From 21d0b647d6916c4dbbde8a310a0760872d44f7a2 Mon Sep 17 00:00:00 2001 From: Jan Tuomi Date: Mon, 6 Jan 2025 18:52:22 +0200 Subject: Rework DB interface to use Recordable instances --- log_db/src/common.rs | 126 +----------- log_db/src/lib.rs | 230 +++++++++++++--------- log_db/src/log_reader_forward.rs | 3 +- log_db/src/log_reader_reverse.rs | 3 +- log_db/src/record.rs | 113 +++++++++++ log_db/tests/integration.rs | 409 ++++++++++++++++++++++----------------- 6 files changed, 502 insertions(+), 382 deletions(-) create mode 100644 log_db/src/record.rs (limited to 'log_db') diff --git a/log_db/src/common.rs b/log_db/src/common.rs index a4f253c..a3ff5c2 100644 --- a/log_db/src/common.rs +++ b/log_db/src/common.rs @@ -400,114 +400,6 @@ impl Value { } } -#[derive(Debug, Clone)] -pub struct Record { - values: Vec, - tombstone: bool, -} - -impl Record { - 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]) -> Record { - 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; - } - Record { values, tombstone } - } - - pub fn from(values: &[Value]) -> Record { - Record { - values: values.to_vec(), - tombstone: false, - } - } - - pub fn values(&self) -> &[Value] { - &self.values - } - - pub fn at(&self, index: usize) -> &Value { - &self.values[index] - } - - pub fn is_tombstone(&self) -> bool { - self.tombstone - } - - pub fn validate(&self, schema: &Vec<(Field, ValueType)>) -> Result<(), DBError> { - // Validate the record length - if self.values.len() != schema.len() { - return Err(DBError::ValidationError(format!( - "Record has an incorrect number of fields: {}, expected {}", - self.values.len(), - schema.len() - ))); - } - - // Validate that record fields match schema types - for (i, (_, field)) in schema.iter().enumerate() { - match (&self.values[i], field) { - ( - Value::Null, - ValueType { - nullable: true, - prim_value_type: _, - }, - ) => {} - ( - Value::Int(_), - ValueType { - prim_value_type: PrimValueType::Int, - .. - }, - ) => {} - ( - Value::String(_), - ValueType { - prim_value_type: PrimValueType::String, - .. - }, - ) => {} - ( - Value::Bytes(_), - ValueType { - prim_value_type: PrimValueType::Bytes, - .. - }, - ) => {} - _ => { - return Err(DBError::ValidationError(format!( - "Record field {} has incorrect type: {:?}, expected {:?}", - &i, &self.values[i], &field.prim_value_type - ))); - } - } - } - Ok(()) - } -} - pub fn type_check(value: &Value, value_type: &ValueType) -> bool { match (value, value_type) { ( @@ -543,14 +435,6 @@ pub fn type_check(value: &Value, value_type: &ValueType) -> bool { } } -/// A trait that describes how to convert a data structure into a database `Record` and vice versa. -pub trait Recordable { - /// Convert the data structure implementing the `Recordable` trait into a database `Record`. - fn to_record(&self) -> Record; - /// Convert a database `Record` into the data structure implementing the `Recordable` trait. - fn from_record(record: &Record) -> Self; -} - pub fn get_secondary_memtable_index_by_field( sks: &Vec, field: &Field, @@ -898,3 +782,13 @@ pub fn request_exclusive_lock(data_dir: &Path, file: &mut fs::File) -> Result<() Ok(()) } + +macro_rules! dbg_trace { + ($($args: expr),*) => { + print!("TRACE: file: {}, line: {}", file!(), line!()); + $( + print!(", {}: {:?}", stringify!($args), $args); + )* + println!(""); // to get a new line at the end + } +} diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs index 3c3f7bb..2cfcfbe 100644 --- a/log_db/src/lib.rs +++ b/log_db/src/lib.rs @@ -1,11 +1,13 @@ #[macro_use] extern crate log; +#[macro_use] mod common; mod log_reader_forward; mod log_reader_reverse; mod memtable_primary; mod memtable_secondary; +mod record; pub use common::*; use fs2::FileExt; @@ -14,28 +16,28 @@ use log_reader_forward::ForwardLogReaderItem; pub use log_reader_reverse::ReverseLogReader; use memtable_primary::PrimaryMemtable; use memtable_secondary::SecondaryMemtable; +pub use record::Recordable; +use record::*; use std::collections::BTreeMap; use std::fmt::Debug; use std::fs::{self}; use std::io::{self, Read, Seek, SeekFrom, Write}; use std::path::{Path, PathBuf}; -pub struct ConfigBuilder { +pub struct ConfigBuilder { data_dir: Option, segment_size: Option, - fields: Option>, - primary_key: Option, - secondary_keys: Option>, + primary_key: Option, + secondary_keys: Option>, write_durability: Option, read_consistency: Option, } -impl<'a, Field: Eq + Clone + Debug> ConfigBuilder { - pub fn new() -> ConfigBuilder { - ConfigBuilder:: { +impl<'a, R: Recordable> ConfigBuilder { + pub fn new() -> ConfigBuilder { + ConfigBuilder:: { data_dir: None, segment_size: None, - fields: None, primary_key: None, secondary_keys: None, write_durability: None, @@ -58,23 +60,17 @@ impl<'a, Field: Eq + Clone + Debug> ConfigBuilder { self } - /// The field schema of the database. - pub fn fields(&mut self, fields: &[(Field, ValueType)]) -> &mut Self { - self.fields = Some(fields.to_vec()); - self - } - /// The primary key of the database, used to construct /// the primary memtable index. This should be the field /// that is most frequently queried. - pub fn primary_key(&mut self, primary_key: Field) -> &mut Self { + pub fn primary_key(&mut self, primary_key: R::Field) -> &mut Self { self.primary_key = Some(primary_key); self } /// The secondary keys of the database, used to construct /// the secondary memtable indexes. - pub fn secondary_keys(&mut self, secondary_keys: &[Field]) -> &mut Self { + pub fn secondary_keys(&mut self, secondary_keys: &[R::Field]) -> &mut Self { self.secondary_keys = Some(secondary_keys.to_vec()); self } @@ -96,18 +92,11 @@ impl<'a, Field: Eq + Clone + Debug> ConfigBuilder { self } - pub fn initialize(&self) -> Result, DBError> { - let config = Config:: { + pub fn initialize(&self) -> Result, DBError> { + let config = Config:: { data_dir: self.data_dir.clone().unwrap_or("db_data".to_string()), segment_size: self.segment_size.unwrap_or(4 * 1024 * 1024), // 4MB - fields: self - .fields - .as_ref() - .ok_or(io::Error::new( - io::ErrorKind::InvalidInput, - "Required config value \"fields\" is not set", - ))? - .clone(), + fields: R::schema(), primary_key: self.primary_key.clone().ok_or(io::Error::new( io::ErrorKind::InvalidInput, "Required config value \"primary_key\" is not set", @@ -123,23 +112,23 @@ impl<'a, Field: Eq + Clone + Debug> ConfigBuilder { .unwrap_or(ReadConsistency::Strong), }; - DB::initialize(&config) + DB::initialize(config) } } #[derive(Clone)] -struct Config { +struct Config { pub data_dir: String, pub segment_size: usize, - pub fields: Vec<(Field, ValueType)>, - pub primary_key: Field, - pub secondary_keys: Vec, + pub fields: Vec<(R::Field, ValueType)>, + pub primary_key: R::Field, + pub secondary_keys: Vec, pub write_durability: WriteDurability, pub read_consistency: ReadConsistency, } -pub struct DB { - config: Config, +pub struct DB { + config: Config, data_dir: PathBuf, active_metadata_file: fs::File, active_data_file: fs::File, @@ -149,13 +138,13 @@ pub struct DB { refresh_next_logkey: LogKey, } -impl DB { +impl DB { /// Create a new database configuration builder. - pub fn configure() -> ConfigBuilder { + pub fn configure() -> ConfigBuilder { ConfigBuilder::new() } - fn initialize(config: &Config) -> Result, DBError> { + fn initialize(config: Config) -> Result, DBError> { 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 @@ -163,8 +152,8 @@ impl DB { // A tempdir-move strategy is used to achieve one-phase commit. // Ensure the data directory exists - let data_dir_path = Path::new(&config.data_dir); - match fs::create_dir(&data_dir_path) { + let data_dir = Path::new(&config.data_dir).to_path_buf(); + match fs::create_dir(&data_dir) { Ok(_) => {} Err(e) => { if e.kind() != io::ErrorKind::AlreadyExists { @@ -177,22 +166,22 @@ impl DB { let init_lock_file = fs::OpenOptions::new() .create(true) .write(true) - .open(&data_dir_path.join(INIT_LOCK_FILENAME))?; + .open(&data_dir.join(INIT_LOCK_FILENAME))?; init_lock_file.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(ACTIVE_SYMLINK_FILENAME))? { - 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)?; + if !fs::exists(data_dir.join(ACTIVE_SYMLINK_FILENAME))? { + let (segment_uuid, _) = create_segment_data_file(&data_dir)?; + let (segment_num, _) = create_segment_metadata_file(&data_dir, &segment_uuid)?; + set_active_segment(&data_dir, segment_num)?; // Create the exclusive lock request file fs::OpenOptions::new() .create(true) .write(true) - .open(data_dir_path.join(EXCL_LOCK_REQUEST_FILENAME))?; + .open(data_dir.join(EXCL_LOCK_REQUEST_FILENAME))?; } init_lock_file.unlock()?; @@ -257,9 +246,9 @@ impl DB { Path::new(&config.data_dir).join(active_metadata_header.uuid.to_string()); let active_data_file = APPEND_MODE.open(&active_data_path)?; - let mut db = DB:: { - config: config.clone(), - data_dir: data_dir_path.to_path_buf(), + let mut db = DB:: { + config, + data_dir, active_metadata_file, active_data_file, primary_key_index, @@ -312,7 +301,7 @@ impl DB { { let log_key = LogKey::new(segnum, index); - if record.is_tombstone() { + if record.tombstone { self.remove_record_from_memtables(&record); } else { self.insert_record_to_memtables(&log_key, &record); @@ -374,12 +363,17 @@ impl DB { /// 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: &Record) -> Result<(), DBError> { + pub fn upsert(&mut self, recordable: R) -> Result<(), DBError> { + let record = Record::from(&recordable.into_record()); debug!("Upserting record: {:?}", record); record.validate(&self.config.fields)?; - debug!("Record is valid"); + + self.upsert_record(&record) + } + + fn upsert_record(&mut self, record: &Record) -> Result<(), DBError> { debug!("Opening file in append mode and acquiring exclusive lock..."); // Acquire an exclusive lock for writing @@ -390,7 +384,7 @@ impl DB { { // The log file has been rotated, so we must try again self.active_metadata_file.unlock()?; - return self.upsert(record); + return self.upsert_record(record); } self.active_data_file.lock_exclusive()?; @@ -462,7 +456,13 @@ impl DB { /// Get a record by its primary index value. /// E.g. `db.get(Value::Int(10))`. - pub fn get(&mut self, query_key: &Value) -> Result, DBError> { + pub fn get(&mut self, query_key: &Value) -> Result, DBError> { + Ok(self + .get_record(query_key)? + .map(|rec| R::from_record(rec.values))) + } + + fn get_record(&mut self, query_key: &Value) -> Result, DBError> { let pk_type = &self.config.fields[self.primary_key_index].1; if !type_check(&query_key, &pk_type) { return Err(DBError::ValidationError(format!( @@ -530,16 +530,16 @@ impl DB { ); let record = Record::deserialize(&data_buf); - return Ok(Some(record)); + Ok(Some(record)) } /// Get a collection of records based on a field value. /// Indexes will be used if they contain the requested key. - pub fn find_all(&mut self, field: &Field, query_key: &Value) -> Result, DBError> { + pub fn find_all(&mut self, field: &R::Field, query_key: &Value) -> Result, DBError> { // If querying by primary key, return the result of `get` wrapped in a vec. if field == &self.config.primary_key { return match self.get(query_key)? { - Some(record) => Ok(vec![record.clone()]), + Some(record) => Ok(vec![record]), None => Ok(vec![]), }; } @@ -620,7 +620,10 @@ impl DB { records.push(record); } - Ok(records) + Ok(records + .into_iter() + .map(|rec| R::from_record(rec.values)) + .collect()) } /// Ensures that the `self.metadata_file` and `self.data_file` handles are still pointing to the correct files. @@ -653,8 +656,8 @@ impl DB { } /// Delete record by primary key. - pub fn delete(&mut self, pk: &Value) -> Result, DBError> { - let record = match self.get(pk)? { + pub fn delete(&mut self, pk: &Value) -> Result, DBError> { + let record = match self.get_record(pk)? { Some(record) => record, None => return Ok(None), }; @@ -703,7 +706,7 @@ impl DB { debug!("Record deleted, returning from delete"); - Ok(Some(record)) + Ok(Some(R::from_record(record.values))) } /// Check if there are any pending tasks and do them. Tasks include: @@ -755,7 +758,7 @@ impl DB { .expect("Primary key was not indexable"); // If the record is a tombstone, remove the PK from the map - if item.record.is_tombstone() { + if item.record.tombstone { map.remove(&pk); } else { map.insert(pk, (original_index, item.record)); @@ -870,6 +873,66 @@ mod tests { Name, } + struct TestInst1 { + id: i64, + } + + impl Recordable for TestInst1 { + type Field = Field; + + fn into_record(self) -> Vec { + vec![Value::Int(self.id)] + } + + fn from_record(record: Vec) -> Self { + let mut it = record.into_iter(); + TestInst1 { + id: match it.next().unwrap() { + Value::Int(i) => i, + _ => panic!("Expected int"), + }, + } + } + + fn schema() -> Vec<(Field, ValueType)> { + vec![(Field::Id, ValueType::int())] + } + } + + struct TestInst2 { + id: i64, + name: String, + } + + impl Recordable for TestInst2 { + type Field = Field; + + fn into_record(self) -> Vec { + vec![Value::Int(self.id), Value::String(self.name)] + } + + fn from_record(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"), + }, + } + } + + fn schema() -> Vec<(Field, ValueType)> { + vec![ + (Field::Id, ValueType::int()), + (Field::Name, ValueType::string()), + ] + } + } + #[test] fn test_compaction() { let _ = env_logger::builder().is_test(true).try_init(); @@ -878,18 +941,17 @@ mod tests { let capacity = 5; let segment_size = capacity * 2 * 8 + METADATA_FILE_HEADER_SIZE; - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(data_dir.to_str().unwrap()) .segment_size(segment_size) - .fields(&[(Field::Id, ValueType::int())]) .primary_key(Field::Id) .initialize() .expect("Failed to create DB"); // Insert records with same value until we reach the capacity for _ in 0..capacity { - let record = Record::from(&[Value::Int(0 as i64)]); - db.upsert(&record).expect("Failed to insert record"); + db.upsert(TestInst1 { id: 0 }) + .expect("Failed to insert record"); } let mut segment1_file = READ_MODE.open(data_dir.join(metadata_filename(1))).unwrap(); @@ -906,8 +968,8 @@ mod tests { .expect("Failed to do maintenance tasks"); // Insert one extra with different value, this goes into another segment - let record = Record::from(&[Value::Int(1 as i64)]); - db.upsert(&record).expect("Failed to insert record"); + 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()); @@ -943,25 +1005,19 @@ mod tests { ); // Check that the records can be read - let rec0 = db + let inst0 = db .get(&Value::Int(0 as i64)) .expect("Failed to get record") .expect("Record not found"); - assert!(match rec0.at(0) { - Value::Int(0) => true, - _ => false, - }); + assert!(inst0.id == 0); - let rec1 = db + let inst1 = db .get(&Value::Int(1 as i64)) .expect("Failed to get record") .expect("Record not found"); - assert!(match rec1.at(0) { - Value::Int(1) => true, - _ => false, - }); + assert!(inst1.id == 1); } #[test] @@ -970,18 +1026,17 @@ mod tests { let temp_dir = tempfile::tempdir().unwrap(); let data_dir = temp_dir.path(); - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(data_dir.to_str().unwrap()) - .fields(&[(Field::Id, ValueType::int())]) .primary_key(Field::Id) .initialize() .expect("Failed to create DB"); // Insert records - let n_recs = 100; + let n_recs: u64 = 100; for i in 0..n_recs { - let record = Record::from(&[Value::Int(i as i64)]); - db.upsert(&record).expect("Failed to insert record"); + db.upsert(TestInst1 { id: i as i64 }) + .expect("Failed to insert record"); } // Open the segment file and write garbage to it to simulate corruption @@ -1023,12 +1078,8 @@ mod tests { let temp_dir = tempfile::tempdir().unwrap(); let data_dir = temp_dir.path(); - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(data_dir.to_str().unwrap()) - .fields(&[ - (Field::Id, ValueType::int()), - (Field::Name, ValueType::string()), - ]) .primary_key(Field::Id) .secondary_keys(&[Field::Name]) .initialize() @@ -1042,8 +1093,11 @@ mod tests { ); // Insert record - let record = Record::from(&[Value::Int(0), Value::String("John".to_string())]); - db.upsert(&record).expect("Failed to 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); diff --git a/log_db/src/log_reader_forward.rs b/log_db/src/log_reader_forward.rs index 5f1d054..8063918 100644 --- a/log_db/src/log_reader_forward.rs +++ b/log_db/src/log_reader_forward.rs @@ -1,4 +1,5 @@ use super::common::*; +use super::record::*; use std::fs::{self}; use std::io::{self, Read, Seek}; @@ -120,7 +121,7 @@ mod tests { let first_record = forward_log_reader .next() .expect("Failed to read the first record"); - assert!(match first_record.record.values() { + assert!(match &first_record.record.values[..] { [Value::Bytes(bytes)] => bytes.len() == 256, _ => false, }); diff --git a/log_db/src/log_reader_reverse.rs b/log_db/src/log_reader_reverse.rs index bf093e8..0c95322 100644 --- a/log_db/src/log_reader_reverse.rs +++ b/log_db/src/log_reader_reverse.rs @@ -1,4 +1,5 @@ use super::common::*; +use super::record::*; use std::fs::{self}; use std::io::{self, Read, Seek}; @@ -97,7 +98,7 @@ mod tests { let last_record = reverse_log_reader .next() .expect("Failed to read the last record"); - assert!(match last_record.values() { + assert!(match &last_record.values[..] { [Value::Bytes(bytes)] => bytes.len() == 256, _ => false, }); diff --git a/log_db/src/record.rs b/log_db/src/record.rs new file mode 100644 index 0000000..1b882ed --- /dev/null +++ b/log_db/src/record.rs @@ -0,0 +1,113 @@ +use super::*; + +#[derive(Debug, Clone)] +pub struct Record { + pub values: Vec, + pub tombstone: bool, +} + +impl Record { + 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]) -> Record { + 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; + } + Record { values, tombstone } + } + + pub fn from(values: &[Value]) -> Record { + Record { + values: values.to_vec(), + tombstone: false, + } + } + + pub fn at(&self, index: usize) -> &Value { + &self.values[index] + } + + pub fn validate(&self, schema: &Vec<(Field, ValueType)>) -> Result<(), DBError> { + // Validate the record length + if self.values.len() != schema.len() { + return Err(DBError::ValidationError(format!( + "Record has an incorrect number of fields: {}, expected {}", + self.values.len(), + schema.len() + ))); + } + + // Validate that record fields match schema types + for (i, (_, field)) in schema.iter().enumerate() { + match (&self.values[i], field) { + ( + Value::Null, + ValueType { + nullable: true, + prim_value_type: _, + }, + ) => {} + ( + Value::Int(_), + ValueType { + prim_value_type: PrimValueType::Int, + .. + }, + ) => {} + ( + Value::String(_), + ValueType { + prim_value_type: PrimValueType::String, + .. + }, + ) => {} + ( + Value::Bytes(_), + ValueType { + prim_value_type: PrimValueType::Bytes, + .. + }, + ) => {} + _ => { + return Err(DBError::ValidationError(format!( + "Record field {} has incorrect type: {:?}, expected {:?}", + &i, &self.values[i], &field.prim_value_type + ))); + } + } + } + Ok(()) + } +} + +/// A trait that describes how to convert a data structure into a database record and vice versa. +pub trait Recordable { + /// The field type of the data structure implementing the `Recordable` trait. + type Field: Eq + Clone + Debug; + /// Define the schema of the data structure implementing the `Recordable` trait. + fn schema() -> Vec<(Self::Field, ValueType)>; + /// Convert the data structure implementing the `Recordable` trait into a vector of database values. + fn into_record(self) -> Vec; + /// Convert a vector of database values into the data structure implementing the `Recordable` trait. + fn from_record(record: Vec) -> Self; +} diff --git a/log_db/tests/integration.rs b/log_db/tests/integration.rs index daf2745..37b4182 100644 --- a/log_db/tests/integration.rs +++ b/log_db/tests/integration.rs @@ -36,16 +36,59 @@ enum Field { Data, } +struct Inst { + pub id: i64, + pub name: Option, + pub data: Vec, +} + +impl Recordable for Inst { + type Field = Field; + fn schema() -> Vec<(Self::Field, ValueType)> { + vec![ + (Field::Id, ValueType::int()), + (Field::Name, ValueType::string().nullable()), + (Field::Data, ValueType::bytes()), + ] + } + + fn into_record(self) -> Vec { + vec![ + Value::Int(self.id), + match self.name { + Some(name) => Value::String(name), + None => Value::Null, + }, + Value::Bytes(self.data), + ] + } + + fn from_record(record: Vec) -> 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), + }, + } + } +} + #[test] fn test_initialize_only() { let data_dir = tmp_dir(); - let _db = DB::configure() + let _db = DB::::configure() .data_dir(&data_dir) - .fields(&[ - (Field::Id, ValueType::int()), - (Field::Name, ValueType::string()), - (Field::Data, ValueType::bytes()), - ]) .primary_key(Field::Id) .initialize() .expect("Failed to initialize DB instance"); @@ -54,109 +97,84 @@ fn test_initialize_only() { #[test] fn test_upsert_and_get_with_primary_memtable() { let data_dir = tmp_dir(); - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(&data_dir) - .fields(&[ - (Field::Id, ValueType::int()), - (Field::Name, ValueType::string()), - (Field::Data, ValueType::bytes()), - ]) .primary_key(Field::Id) .initialize() .expect("Failed to initialize DB instance"); - let record = Record::from(&[ - Value::Int(1), - Value::String("Alice".to_string()), - Value::Bytes(vec![0, 1, 2]), - ]); - db.upsert(&record).unwrap(); + let id = 1; + let inst = Inst { + id, + name: Some("Alice".to_string()), + data: vec![0, 1, 2], + }; + db.upsert(inst).unwrap(); let result = db.get(&Value::Int(1)).unwrap().unwrap(); // Check that the IDs match - assert!(match (result.at(0), record.at(0)) { - (Value::Int(a), Value::Int(b)) => a == b, - _ => false, - }); + assert!(result.id == id); } #[test] fn test_upsert_and_get() { let data_dir = tmp_dir(); - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(&data_dir) - .fields(&[ - (Field::Id, ValueType::int()), - (Field::Name, ValueType::string().nullable()), - (Field::Data, ValueType::bytes()), - ]) .primary_key(Field::Id) .initialize() .expect("Failed to initialize DB instance"); // Insert some records - let record0 = Record::from(&[Value::Int(0), Value::Null, Value::Bytes(vec![3, 4, 5])]); - db.upsert(&record0).unwrap(); - - let record1 = Record::from(&[ - Value::Int(1), - Value::String("Alice".to_string()), - Value::Bytes(vec![0, 1, 2]), - ]); - db.upsert(&record1).unwrap(); - - let record2 = Record::from(&[ - Value::Int(1), - Value::String("Bob".to_string()), - Value::Bytes(vec![0, 1, 2]), - ]); - db.upsert(&record2).unwrap(); - - let record3 = Record::from(&[ - Value::Int(2), - Value::String("George".to_string()), - Value::Bytes(vec![]), - ]); - db.upsert(&record3).unwrap(); + 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 = db.get(&Value::Int(0)).unwrap().unwrap(); - // Should match record0 - assert!(match (result.at(0), record0.at(0)) { - (Value::Int(a), Value::Int(b)) => a == b, - _ => false, - }); - assert!(match (result.at(1), record0.at(1)) { - (Value::Null, Value::Null) => true, - _ => false, - }); + // Should match id == 0 + assert!(result.id == 0); + assert!(result.name == None); // Get with ID = 1 let result = db.get(&Value::Int(1)).unwrap().unwrap(); - // Should match record2 - assert!(match (result.at(0), record2.at(0)) { - (Value::Int(a), Value::Int(b)) => a == b, - _ => false, - }); - assert!(match (result.at(1), record2.at(1)) { - (Value::String(a), Value::String(b)) => a == b, - _ => false, - }); + // 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() + let mut db = DB::::configure() .data_dir(&data_dir) - .fields(&[ - (Field::Id, ValueType::int()), - (Field::Name, ValueType::string().nullable()), - (Field::Data, ValueType::bytes()), - ]) .primary_key(Field::Id) .initialize() .expect("Failed to initialize DB instance"); @@ -165,101 +183,127 @@ fn test_get_nonexistant() { assert!(result.is_none()); } +struct InstTestNullable {} +impl Recordable for InstTestNullable { + type Field = Field; + fn schema() -> Vec<(Self::Field, ValueType)> { + vec![(Field::Id, ValueType::int())] + } + + fn into_record(self) -> Vec { + vec![Value::Null] + } + + fn from_record(_record: Vec) -> Self { + Self {} + } +} + #[test] fn test_upsert_fails_on_null_in_non_nullable_field() { let data_dir = tmp_dir(); - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(&data_dir) - .fields(&[(Field::Id, ValueType::int())]) .primary_key(Field::Id) .initialize() .expect("Failed to initialize DB instance"); // Null value - let record = Record::from(&[Value::Null]); - assert!(db.upsert(&record).is_err()); + assert!(db.upsert(InstTestNullable {}).is_err()); +} + +struct InstTestNumValues {} +impl Recordable for InstTestNumValues { + type Field = Field; + fn schema() -> Vec<(Self::Field, ValueType)> { + vec![ + (Field::Id, ValueType::int()), + (Field::Name, ValueType::string()), + ] + } + + fn into_record(self) -> Vec { + vec![Value::Int(0)] + } + + fn from_record(_record: Vec) -> Self { + Self {} + } } #[test] fn test_upsert_fails_on_invalid_number_of_values() { let data_dir = tmp_dir(); - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(&data_dir) - .fields(&[ - (Field::Id, ValueType::int()), - (Field::Name, ValueType::string()), - (Field::Data, ValueType::bytes()), - ]) .primary_key(Field::Id) .initialize() .expect("Failed to initialize DB instance"); - // Missing primary key - let record = Record::from(&[ - Value::String("Alice".to_string()), - Value::Bytes(vec![0, 1, 2]), - ]); - assert!(db.upsert(&record).is_err()); + // Missing values + assert!(db.upsert(InstTestNumValues {}).is_err()); +} + +struct InstTestInvalidType {} +impl Recordable for InstTestInvalidType { + type Field = Field; + fn schema() -> Vec<(Self::Field, ValueType)> { + vec![(Field::Id, ValueType::int())] + } + + fn into_record(self) -> Vec { + vec![Value::String("foo".to_string())] + } + + fn from_record(_record: Vec) -> Self { + Self {} + } } #[test] fn test_upsert_fails_on_invalid_value_type() { let data_dir = tmp_dir(); - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(&data_dir) - .fields(&[ - (Field::Id, ValueType::int()), - (Field::Name, ValueType::string()), - (Field::Data, ValueType::bytes()), - ]) .primary_key(Field::Id) .initialize() .expect("Failed to initialize DB instance"); - let record = Record::from(&[ - Value::String("foo".to_string()), - Value::String("bar".to_string()), - Value::String("baz".to_string()), - ]); - assert!(db.upsert(&record).is_err()); + // Invalid type + assert!(db.upsert(InstTestInvalidType {}).is_err()); } #[test] fn test_upsert_and_find_all() { let data_dir = tmp_dir(); - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(&data_dir) - .fields(&[ - (Field::Id, ValueType::int()), - (Field::Name, ValueType::string()), - (Field::Data, ValueType::bytes()), - ]) .primary_key(Field::Id) .secondary_keys(&[Field::Name]) .initialize() .expect("Failed to initialize DB instance"); // Insert some records - let record0 = Record::from(&[ - Value::Int(0), - Value::String("John".to_string()), - Value::Bytes(vec![3, 4, 5]), - ]); - db.upsert(&record0).unwrap(); - - let record1 = Record::from(&[ - Value::Int(1), - Value::String("John".to_string()), - Value::Bytes(vec![1, 2, 3]), - ]); - db.upsert(&record1).unwrap(); - - let record2 = Record::from(&[ - Value::Int(2), - Value::String("George".to_string()), - Value::Bytes(vec![1, 2, 3]), - ]); - db.upsert(&record2).unwrap(); + 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 @@ -269,6 +313,30 @@ fn test_upsert_and_find_all() { assert_eq!(johns.len(), 2); } +struct InstSingleId { + pub id: i64, +} + +impl Recordable for InstSingleId { + type Field = Field; + fn schema() -> Vec<(Self::Field, ValueType)> { + vec![(Field::Id, ValueType::int())] + } + + fn into_record(self) -> Vec { + vec![Value::Int(self.id)] + } + + fn from_record(record: Vec) -> Self { + Self { + id: match record[0] { + Value::Int(id) => id, + _ => panic!("Invalid value type"), + }, + } + } +} + #[test] #[serial] fn test_multiple_writing_threads() { @@ -280,15 +348,14 @@ fn test_multiple_writing_threads() { for i in 0..threads_n { let data_dir = data_dir.clone(); threads.push(thread::spawn(move || { - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(&data_dir) - .fields(&[(Field::Id, ValueType::int())]) .primary_key(Field::Id) .initialize() .expect("Failed to initialize DB instance"); - let record = Record::from(&[Value::Int(i)]); - db.upsert(&record).expect("Failed to upsert record"); + db.upsert(InstSingleId { id: i }) + .expect("Failed to upsert record"); })); } @@ -297,9 +364,8 @@ fn test_multiple_writing_threads() { } // Read the records - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(&data_dir) - .fields(&[(Field::Id, ValueType::int())]) .primary_key(Field::Id) .initialize() .expect("Failed to initialize DB instance"); @@ -310,10 +376,7 @@ fn test_multiple_writing_threads() { .expect("Failed to get record") .expect("Record not found"); - assert!(match &result.values() { - [Value::Int(a)] => a == &i, - _ => false, - }); + assert!(result.id == i); } } @@ -328,10 +391,9 @@ fn test_one_writer_and_multiple_reading_threads() { for i in 0..threads_n { let data_dir = data_dir.clone(); threads.push(thread::spawn(move || { - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(&data_dir) .segment_size(1000) // should cause rotations - .fields(&[(Field::Id, ValueType::int())]) .primary_key(Field::Id) .initialize() .expect("Failed to initialize DB instance"); @@ -346,10 +408,7 @@ fn test_one_writer_and_multiple_reading_threads() { continue; } Some(result) => { - assert!(match &result.values() { - [Value::Int(a)] => a == &i, - _ => false, - }); + assert!(result.id == i); break; } }; @@ -359,16 +418,15 @@ fn test_one_writer_and_multiple_reading_threads() { // Add a writer that inserts the records threads.push(thread::spawn(move || { - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(&data_dir) - .fields(&[(Field::Id, ValueType::int())]) .primary_key(Field::Id) .initialize() .expect("Failed to initialize DB instance"); for i in 0..threads_n { - let record = Record::from(&[Value::Int(i)]); - db.upsert(&record).expect("Failed to upsert record"); + 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"); @@ -385,23 +443,27 @@ fn test_log_is_rotated_when_capacity_reached() { let data_dir = tmp_dir(); let data_dir_path = Path::new(&data_dir); - let record = Record::from(&[Value::Int(1), Value::Bytes(vec![1, 2, 3, 4])]); - let record_len = &record.serialize().len(); + // 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() + let mut db = DB::::configure() .data_dir(&data_dir) .segment_size(10 * record_len) // small log segment size - .fields(&[ - (Field::Id, ValueType::int()), - (Field::Data, ValueType::bytes()), - ]) .primary_key(Field::Id) .initialize() .expect("Failed to initialize DB instance"); // Insert more records than fits the capacity for _ in 0..25 { - db.upsert(&record).expect("Failed to upsert record"); + 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"); @@ -418,32 +480,27 @@ fn test_log_is_rotated_when_capacity_reached() { #[test] fn test_delete() { let data_dir = tmp_dir(); - let mut db = DB::configure() + let mut db = DB::::configure() .data_dir(&data_dir) - .fields(&[ - (Field::Id, ValueType::int()), - (Field::Name, ValueType::string()), - (Field::Data, ValueType::bytes()), - ]) .primary_key(Field::Id) .secondary_keys(&[Field::Name]) .initialize() .expect("Failed to initialize DB instance"); // Insert some records - let record0 = Record::from(&[ - Value::Int(0), - Value::String("John".to_string()), - Value::Bytes(vec![3, 4, 5]), - ]); - db.upsert(&record0).unwrap(); - - let record1 = Record::from(&[ - Value::Int(1), - Value::String("John".to_string()), - Value::Bytes(vec![1, 2, 3]), - ]); - db.upsert(&record1).unwrap(); + 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(); -- cgit v1.3