diff options
Diffstat (limited to 'log_db/src')
| -rw-r--r-- | log_db/src/config.rs | 97 | ||||
| -rw-r--r-- | log_db/src/engine.rs | 44 | ||||
| -rw-r--r-- | log_db/src/lib.rs | 100 | ||||
| -rw-r--r-- | log_db/src/record.rs | 19 |
4 files changed, 145 insertions, 115 deletions
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<R: Recordable> { +pub struct Schema<F> { + pub fields: Vec<(F, Type)>, + pub primary_key: F, + pub secondary_keys: Vec<F>, +} + +pub struct ConfigBuilder<T, F> { data_dir: Option<String>, segment_size: Option<usize>, write_durability: Option<WriteDurability>, read_consistency: Option<ReadConsistency>, - _marker: PhantomData<R>, + + schema: Option<Vec<(F, Type)>>, + primary_key: Option<F>, + secondary_keys: Option<Vec<F>>, + from_record: Option<fn(Vec<Value>) -> T>, + into_record: Option<fn(T) -> Vec<Value>>, + + _marker: PhantomData<T>, } -impl<R: Recordable> ConfigBuilder<R> { - pub fn new() -> ConfigBuilder<R> { +impl<T, F: Eq + Clone> ConfigBuilder<T, F> { + pub fn new() -> ConfigBuilder<T, F> { 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<String>) -> Self { + self.data_dir = Some(data_dir.into()); self } @@ -29,7 +49,7 @@ impl<R: Recordable> ConfigBuilder<R> { /// 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<R: Recordable> ConfigBuilder<R> { /// 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<R: Recordable> ConfigBuilder<R> { /// 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<DB<R>> { + 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<F>) -> Self { + self.secondary_keys = Some(secondary_keys); + self + } + + pub fn from_record(mut self, from_record: fn(Vec<Value>) -> T) -> Self { + self.from_record = Some(from_record); + self + } + + pub fn into_record(mut self, into_record: fn(T) -> Vec<Value>) -> Self { + self.into_record = Some(into_record); + self + } + + pub fn initialize(self) -> DBResult<DB<T, F>> { + 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<R: Recordable> ConfigBuilder<R> { } #[derive(Clone)] -pub struct Config<R: Recordable> { - pub fields: Vec<(R::Field, Type)>, - pub primary_key: R::Field, - pub secondary_keys: Vec<R::Field>, +pub struct Config<T, F> { + pub schema: Vec<(F, Type)>, + pub primary_key: F, + pub secondary_keys: Vec<F>, + pub from_record: fn(Vec<Value>) -> T, + pub into_record: fn(T) -> Vec<Value>, 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<R: Recordable> { - pub config: Config<R>, +pub struct Engine<T, F: Eq + Clone> { + pub config: Config<T, F>, pub lock_manager: LockManager, data_dir_path: PathBuf, @@ -19,8 +19,8 @@ pub struct Engine<R: Recordable> { pub secondary_memtables: Vec<SecondaryMemtable>, } -impl<R: Recordable> Engine<R> { - pub fn initialize(config: Config<R>) -> DBResult<Engine<R>> { +impl<T, F: Eq + Clone> Engine<T, F> { + pub fn initialize(config: Config<T, F>) -> DBResult<Engine<T, F>> { 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<R: Recordable> Engine<R> { // 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<R: Recordable> Engine<R> { // 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<R: Recordable> Engine<R> { 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::<R> { + let mut engine = Engine::<T, F> { config, lock_manager, data_dir_path, @@ -164,7 +164,7 @@ impl<R: Recordable> Engine<R> { 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<R: Recordable> Engine<R> { 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<R: Recordable> Engine<R> { 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<R: Recordable> Engine<R> { pub fn batch_find_by_records<'a>( &mut self, - field: &R::Field, + field: &F, values: impl Iterator<Item = &'a Value>, ) -> DBResult<Vec<(usize, Record)>> { let field_type = self.get_field_type(field).ok_or(DBError::ValidationError( @@ -277,10 +278,7 @@ impl<R: Recordable> Engine<R> { .collect::<DBResult<Vec<IndexableValue>>>()?; // 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<R: Recordable> Engine<R> { pub fn range_by_records<B: RangeBounds<Value>>( &mut self, - field: &R::Field, + field: &F, range: B, ) -> DBResult<Vec<Record>> { fn range_bound_to_indexable( @@ -479,7 +477,7 @@ impl<R: Recordable> Engine<R> { } } - pub fn delete_by_field(&mut self, field: &R::Field, value: &Value) -> DBResult<Vec<Record>> { + pub fn delete_by_field(&mut self, field: &F, value: &Value) -> DBResult<Vec<Record>> { let recs: Vec<Record> = self .batch_find_by_records(field, std::iter::once(value))? .into_iter() @@ -708,19 +706,19 @@ impl<R: Recordable> Engine<R> { } #[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<T>( + pub fn with_exclusive_lock<A>( &mut self, - f: impl FnOnce(&mut Self) -> DBResult<T>, - ) -> DBResult<T> { + f: impl FnOnce(&mut Self) -> DBResult<A>, + ) -> DBResult<A> { // 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<R: Recordable> Engine<R> { } #[inline] - pub fn with_shared_lock<T>(&mut self, f: impl FnOnce(&mut Self) -> DBResult<T>) -> DBResult<T> { + pub fn with_shared_lock<A>(&mut self, f: impl FnOnce(&mut Self) -> DBResult<A>) -> DBResult<A> { // 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<R: Recordable> { - engine: Engine<R>, +pub struct DB<T, F: Eq + Clone> { + engine: Engine<T, F>, } -impl<R: Recordable> DB<R> { +impl<T, F: Eq + Clone> DB<T, F> { /// Create a new database configuration builder. - pub fn configure() -> ConfigBuilder<R> { + pub fn configure() -> ConfigBuilder<T, F> { ConfigBuilder::new() } - fn initialize(config: Config<R>) -> DBResult<DB<R>> { + fn initialize(config: Config<T, F>) -> DBResult<DB<T, F>> { 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<R: Recordable> DB<R> { /// Get a record by its primary index value. /// E.g. `db.get(Value::Int(10))`. - pub fn get(&mut self, value: &Value) -> DBResult<Option<R>> { + pub fn get(&mut self, value: &Value) -> DBResult<Option<T>> { 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<R: Recordable> DB<R> { 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<Vec<R>> { + pub fn find_by(&mut self, field: &F, value: &Value) -> DBResult<Vec<T>> { 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<Vec<(usize, R)>> { + pub fn batch_find_by(&mut self, field: &F, values: &[Value]) -> DBResult<Vec<(usize, T)>> { 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<B: RangeBounds<Value>>( - &mut self, - field: &R::Field, - range: B, - ) -> DBResult<Vec<R>> { + pub fn range_by<B: RangeBounds<Value>>(&mut self, field: &F, range: B) -> DBResult<Vec<T>> { 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<R: Recordable> DB<R> { /// /// 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<Vec<R>> { + pub fn delete_by(&mut self, field: &F, value: &Value) -> DBResult<Vec<T>> { 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<Option<R>> { + pub fn delete(&mut self, pk: &Value) -> DBResult<Option<T>> { 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<R: Recordable> DB<R> { 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<Value> { 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<Self::Field> { - vec![Field::Name] - } - + impl TestInst2 { fn into_record(self) -> Vec<Value> { 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::<TestInst1>::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::<TestInst1>::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::<TestInst2>::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<Self::Field> { - Vec::new() - } - - /// Convert the data structure implementing the `Recordable` trait into a vector of database values. - fn into_record(self) -> Vec<Value>; - /// Convert a vector of database values into the data structure implementing the `Recordable` trait. - fn from_record(record: Vec<Value>) -> Self; -} - #[cfg(test)] mod tests { use super::*; |