From d422ffe3d48d2061b4d3ddeb0c08796e4d71a5fa Mon Sep 17 00:00:00 2001 From: Jan Tuomi Date: Wed, 5 Feb 2025 17:50:02 +0200 Subject: Remove Recordable trait, implement Python bindings, fixes --- log_db/src/config.rs | 97 ++++++++++++++++++++++++++++++++++++++++--------- log_db/src/engine.rs | 44 +++++++++++------------ log_db/src/lib.rs | 100 +++++++++++++++++++++++---------------------------- log_db/src/record.rs | 19 ---------- 4 files changed, 145 insertions(+), 115 deletions(-) (limited to 'log_db/src') diff --git a/log_db/src/config.rs b/log_db/src/config.rs index 40e0f3c..2ed11c9 100644 --- a/log_db/src/config.rs +++ b/log_db/src/config.rs @@ -1,27 +1,47 @@ use super::*; -pub struct ConfigBuilder { +pub struct Schema { + pub fields: Vec<(F, Type)>, + pub primary_key: F, + pub secondary_keys: Vec, +} + +pub struct ConfigBuilder { data_dir: Option, segment_size: Option, write_durability: Option, read_consistency: Option, - _marker: PhantomData, + + schema: Option>, + primary_key: Option, + secondary_keys: Option>, + from_record: Option) -> T>, + into_record: Option Vec>, + + _marker: PhantomData, } -impl ConfigBuilder { - pub fn new() -> ConfigBuilder { +impl ConfigBuilder { + pub fn new() -> ConfigBuilder { ConfigBuilder { data_dir: None, segment_size: None, write_durability: None, read_consistency: None, + + schema: None, + primary_key: None, + secondary_keys: None, + from_record: None, + into_record: None, + _marker: PhantomData, } } /// The directory where the database will store its data. - pub fn data_dir(&mut self, data_dir: &str) -> &mut Self { - self.data_dir = Some(data_dir.to_string()); + pub fn data_dir(mut self, data_dir: impl Into) -> Self { + self.data_dir = Some(data_dir.into()); self } @@ -29,7 +49,7 @@ impl ConfigBuilder { /// 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) -> &mut Self { + pub fn segment_size(mut self, segment_size: usize) -> Self { self.segment_size = Some(segment_size); self } @@ -37,7 +57,7 @@ impl ConfigBuilder { /// 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) -> &mut Self { + pub fn write_durability(mut self, write_durability: WriteDurability) -> Self { self.write_durability = Some(write_durability); self } @@ -46,16 +66,57 @@ impl ConfigBuilder { /// 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) -> &mut Self { + pub fn read_consistency(mut self, read_consistency: ReadConsistency) -> Self { self.read_consistency = Some(read_consistency); self } - pub fn initialize(&self) -> DBResult> { + pub fn schema(mut self, schema: Vec<(F, Type)>) -> Self { + self.schema = Some(schema); + self + } + + pub fn primary_key(mut self, primary_key: F) -> Self { + self.primary_key = Some(primary_key); + self + } + + pub fn secondary_keys(mut self, secondary_keys: Vec) -> Self { + self.secondary_keys = Some(secondary_keys); + self + } + + pub fn from_record(mut self, from_record: fn(Vec) -> T) -> Self { + self.from_record = Some(from_record); + self + } + + pub fn into_record(mut self, into_record: fn(T) -> Vec) -> Self { + self.into_record = Some(into_record); + self + } + + pub fn initialize(self) -> DBResult> { + let schema = self + .schema + .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 from_record = self + .from_record + .ok_or_else(|| DBError::ValidationError("Callback from_record not set".to_string()))?; + let into_record = self + .into_record + .ok_or_else(|| DBError::ValidationError("Callback into_record not set".to_string()))?; + let config = Config { - fields: R::schema(), - primary_key: R::primary_key(), - secondary_keys: R::secondary_keys(), + schema, + primary_key, + secondary_keys: self.secondary_keys.unwrap_or_default(), + from_record, + into_record, + 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 @@ -73,10 +134,12 @@ impl ConfigBuilder { } #[derive(Clone)] -pub struct Config { - pub fields: Vec<(R::Field, Type)>, - pub primary_key: R::Field, - pub secondary_keys: Vec, +pub struct Config { + pub schema: Vec<(F, Type)>, + pub primary_key: F, + pub secondary_keys: Vec, + pub from_record: fn(Vec) -> T, + pub into_record: fn(T) -> Vec, pub data_dir: String, pub segment_size: usize, pub write_durability: WriteDurability, diff --git a/log_db/src/engine.rs b/log_db/src/engine.rs index d028ee0..b8ac6b9 100644 --- a/log_db/src/engine.rs +++ b/log_db/src/engine.rs @@ -1,7 +1,7 @@ use super::*; -pub struct Engine { - pub config: Config, +pub struct Engine { + pub config: Config, pub lock_manager: LockManager, data_dir_path: PathBuf, @@ -19,8 +19,8 @@ pub struct Engine { pub secondary_memtables: Vec, } -impl Engine { - pub fn initialize(config: Config) -> DBResult> { +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. @@ -66,7 +66,7 @@ impl Engine { // Calculate the index of the primary value in a record let primary_key_index = config - .fields + .schema .iter() .position(|(field, _)| field == &config.primary_key) .ok_or(DBError::ValidationError( @@ -80,7 +80,7 @@ impl Engine { // If any of the keys is not in the schema or // is not an IndexableValue, return an error for &key in &all_keys { - let (_, value_type) = config.fields.iter().find(|(field, _)| field == key).ok_or( + let (_, value_type) = config.schema.iter().find(|(field, _)| field == key).ok_or( DBError::ValidationError("Key must be present in the field schema".to_owned()), )?; @@ -109,7 +109,7 @@ impl Engine { 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:: { + let mut engine = Engine:: { config, lock_manager, data_dir_path, @@ -164,7 +164,7 @@ impl Engine { ForwardLogReader::new_with_index(metadata_file, data_file, from_index) { // Validate that the values in the record are compatible with the schema - record.validate(&self.config.fields)?; + record.validate(&self.config.schema)?; let log_key = LogKey::new(segnum, index); @@ -196,7 +196,7 @@ impl Engine { let secondary_memtable = &mut self.secondary_memtables[sk_index]; let sk_field_index = self .config - .fields + .schema .iter() .position(|(f, _)| sk_field == f) .unwrap(); @@ -214,11 +214,12 @@ impl Engine { let pk = record.at(self.primary_key_index).as_indexable().unwrap(); if let Some(plk) = self.primary_memtable.remove(&pk) { + // TODO this does not work (test_delete_by_multiple_indexes) 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 - .fields + .schema .iter() .position(|(f, _)| sk_field == f) .unwrap(); @@ -254,7 +255,7 @@ impl Engine { pub fn batch_find_by_records<'a>( &mut self, - field: &R::Field, + field: &F, values: impl Iterator, ) -> DBResult> { let field_type = self.get_field_type(field).ok_or(DBError::ValidationError( @@ -277,10 +278,7 @@ impl Engine { .collect::>>()?; // Otherwise, continue with querying secondary indexes. - debug!( - "Finding all records with fields {:?} = {:?}", - field, indexables - ); + debug!("Finding all records with matching fields"); if self.config.read_consistency == ReadConsistency::Strong { self.refresh_indexes()?; @@ -395,7 +393,7 @@ impl Engine { pub fn range_by_records>( &mut self, - field: &R::Field, + field: &F, range: B, ) -> DBResult> { fn range_bound_to_indexable( @@ -479,7 +477,7 @@ impl Engine { } } - pub fn delete_by_field(&mut self, field: &R::Field, value: &Value) -> DBResult> { + pub fn delete_by_field(&mut self, field: &F, value: &Value) -> DBResult> { let recs: Vec = self .batch_find_by_records(field, std::iter::once(value))? .into_iter() @@ -708,19 +706,19 @@ impl Engine { } #[inline] - fn get_field_type(&self, field: &R::Field) -> Option<&Type> { + fn get_field_type(&self, field: &F) -> Option<&Type> { self.config - .fields + .schema .iter() .find(|(f, _)| f == field) .map(|(_, t)| t) } #[inline] - pub fn with_exclusive_lock( + pub fn with_exclusive_lock( &mut self, - f: impl FnOnce(&mut Self) -> DBResult, - ) -> DBResult { + 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 { @@ -734,7 +732,7 @@ impl Engine { } #[inline] - pub fn with_shared_lock(&mut self, f: impl FnOnce(&mut Self) -> DBResult) -> DBResult { + 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 { diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs index 5796a7e..0317320 100644 --- a/log_db/src/lib.rs +++ b/log_db/src/lib.rs @@ -27,9 +27,8 @@ mod memtable_primary; mod memtable_secondary; mod record; -pub use common::{DBError, DBResult, Type, Value}; -pub use config::{ReadConsistency, WriteDurability}; -pub use record::Recordable; +pub use common::{DBError, DBResult, OwnedBounds, Type, Value}; +pub use config::{ReadConsistency, Schema, WriteDurability}; use common::*; use config::*; @@ -40,28 +39,28 @@ use memtable_primary::PrimaryMemtable; use memtable_secondary::SecondaryMemtable; use record::*; -pub struct DB { - engine: Engine, +pub struct DB { + engine: Engine, } -impl DB { +impl DB { /// Create a new database configuration builder. - pub fn configure() -> ConfigBuilder { + pub fn configure() -> ConfigBuilder { ConfigBuilder::new() } - fn initialize(config: Config) -> DBResult> { + 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, recordable: R) -> DBResult<()> { - let record = Record::from(&recordable.into_record()); + pub fn upsert(&mut self, recordable: T) -> DBResult<()> { + let record = Record::from(&(self.engine.config.into_record)(recordable)); debug!("Upserting record: {:?}", record); - record.validate(&self.engine.config.fields)?; + record.validate(&self.engine.config.schema)?; debug!("Record is valid"); self.engine @@ -72,7 +71,7 @@ impl DB { /// Get a record by its primary index value. /// E.g. `db.get(Value::Int(10))`. - pub fn get(&mut self, value: &Value) -> DBResult> { + pub fn get(&mut self, value: &Value) -> DBResult> { let recs = self.engine.with_shared_lock(|engine| { engine.batch_find_by_records( // TODO: This clone is only here to appease the borrow checker @@ -86,54 +85,46 @@ impl DB { Ok(recs .into_iter() .next() - .map(|(_, rec)| R::from_record(rec.values))) + .map(|(_, rec)| (self.engine.config.from_record)(rec.values))) } /// Get a collection of records based on an indexed field value. - pub fn find_by(&mut self, field: &R::Field, value: &Value) -> DBResult> { + pub fn find_by(&mut self, field: &F, value: &Value) -> DBResult> { let recs = self.engine.with_shared_lock(|engine| { engine.batch_find_by_records(field, std::iter::once(value)) })?; Ok(recs .into_iter() - .map(|(_, rec)| R::from_record(rec.values)) + .map(|(_, rec)| (self.engine.config.from_record)(rec.values)) .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: &R::Field, - values: &[Value], - ) -> DBResult> { + pub fn batch_find_by(&mut self, field: &F, values: &[Value]) -> DBResult> { let recs = self .engine .with_shared_lock(|engine| engine.batch_find_by_records(field, values.iter()))?; Ok(recs .into_iter() - .map(|(tag, rec)| (tag, R::from_record(rec.values))) + .map(|(tag, rec)| (tag, (self.engine.config.from_record)(rec.values))) .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: &R::Field, - range: B, - ) -> DBResult> { + pub fn range_by>(&mut self, field: &F, range: B) -> DBResult> { let recs = self .engine .with_shared_lock(|engine| engine.range_by_records(field, range))?; Ok(recs .into_iter() - .map(|rec| R::from_record(rec.values)) + .map(|rec| (self.engine.config.from_record)(rec.values)) .collect()) } @@ -143,19 +134,19 @@ impl DB { /// /// 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: &R::Field, value: &Value) -> DBResult> { + pub fn delete_by(&mut self, field: &F, value: &Value) -> DBResult> { let recs = self .engine .with_exclusive_lock(|engine| engine.delete_by_field(field, value))?; Ok(recs .into_iter() - .map(|rec| R::from_record(rec.values)) + .map(|rec| (self.engine.config.from_record)(rec.values)) .collect()) } /// Delete record by primary key. - pub fn delete(&mut self, pk: &Value) -> DBResult> { + 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 @@ -167,7 +158,7 @@ impl DB { Ok(recs .into_iter() .next() - .map(|rec| R::from_record(rec.values))) + .map(|rec| (self.engine.config.from_record)(rec.values))) } /// Check if there are any pending tasks and do them. Tasks include: @@ -258,15 +249,7 @@ mod tests { id: i64, } - impl Recordable for TestInst1 { - type Field = Field; - fn schema() -> Vec<(Field, Type)> { - vec![(Field::Id, Type::int())] - } - fn primary_key() -> Self::Field { - Field::Id - } - + impl TestInst1 { fn into_record(self) -> Vec { vec![Value::Int(self.id)] } @@ -287,15 +270,7 @@ mod tests { name: String, } - impl Recordable for TestInst2 { - type Field = Field; - fn primary_key() -> Self::Field { - Field::Id - } - fn secondary_keys() -> Vec { - vec![Field::Name] - } - + impl TestInst2 { fn into_record(self) -> Vec { vec![Value::Int(self.id), Value::String(self.name)] } @@ -313,10 +288,6 @@ mod tests { }, } } - - fn schema() -> Vec<(Field, Type)> { - vec![(Field::Id, Type::int()), (Field::Name, Type::string())] - } } #[test] @@ -327,8 +298,13 @@ 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()) + .schema(vec![(Field::Id, Type::int())]) + .primary_key(Field::Id) + .from_record(TestInst1::from_record) + .into_record(TestInst1::into_record) .segment_size(segment_size) .initialize() .expect("Failed to create DB"); @@ -411,8 +387,12 @@ 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()) + .schema(vec![(Field::Id, Type::int())]) + .primary_key(Field::Id) + .from_record(TestInst1::from_record) + .into_record(TestInst1::into_record) .initialize() .expect("Failed to create DB"); @@ -462,8 +442,16 @@ 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()) + .schema(vec![ + (Field::Id, Type::int()), + (Field::Name, Type::string()), + ]) + .primary_key(Field::Id) + .secondary_keys(vec![Field::Name]) + .from_record(TestInst2::from_record) + .into_record(TestInst2::into_record) .initialize() .expect("Failed to create DB"); diff --git a/log_db/src/record.rs b/log_db/src/record.rs index ab9f8dc..c373897 100644 --- a/log_db/src/record.rs +++ b/log_db/src/record.rs @@ -114,25 +114,6 @@ impl Record { } } -/// 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 instance implementing the `Recordable` trait. - fn schema() -> Vec<(Self::Field, Type)>; - /// Define the primary key of the instance implementing the `Recordable` trait. - fn primary_key() -> Self::Field; - /// Define the secondary keys of the instance implementing the `Recordable` trait. - fn secondary_keys() -> Vec { - Vec::new() - } - - /// 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; -} - #[cfg(test)] mod tests { use super::*; -- cgit v1.3