aboutsummaryrefslogtreecommitdiffstats
path: root/log_db
diff options
context:
space:
mode:
Diffstat (limited to 'log_db')
-rw-r--r--log_db/benches/benchmark.rs53
-rw-r--r--log_db/benches/utils.rs16
-rw-r--r--log_db/src/config.rs97
-rw-r--r--log_db/src/engine.rs44
-rw-r--r--log_db/src/lib.rs100
-rw-r--r--log_db/src/record.rs19
-rw-r--r--log_db/tests/integration.rs262
7 files changed, 404 insertions, 187 deletions
diff --git a/log_db/benches/benchmark.rs b/log_db/benches/benchmark.rs
index 148c0b4..4c9ad47 100644
--- a/log_db/benches/benchmark.rs
+++ b/log_db/benches/benchmark.rs
@@ -10,12 +10,17 @@ 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
+ let data_dir = data_dir_obj
.path()
.to_str()
.expect("Failed to convert tmpdir path to str");
- let mut db = DB::<Inst>::configure()
- .data_dir(&data_dir)
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
+ .data_dir(data_dir)
.initialize()
.expect("Failed to initialize DB");
@@ -37,12 +42,17 @@ pub fn upsert_compacted(c: &mut Criterion) {
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
+ let data_dir = data_dir_obj
.path()
.to_str()
.expect("Failed to convert tmpdir path to str");
- let mut db = DB::<Inst>::configure()
- .data_dir(&data_dir)
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
+ .data_dir(data_dir)
.initialize()
.expect("Failed to initialize DB");
@@ -71,12 +81,17 @@ pub fn upsert_write_durability(c: &mut Criterion) {
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
+ let data_dir = data_dir_obj
.path()
.to_str()
.expect("Failed to convert tmpdir path to str");
- let mut db = DB::<Inst>::configure()
- .data_dir(&data_dir)
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
+ .data_dir(data_dir)
.write_durability(mode.clone())
.initialize()
.expect("Failed to initialize DB");
@@ -94,12 +109,17 @@ 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
+ let data_dir = data_dir_obj
.path()
.to_str()
.expect("Failed to convert tmpdir path to str");
- let mut db = DB::<Inst>::configure()
- .data_dir(&data_dir)
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
+ .data_dir(data_dir)
.initialize()
.expect("Failed to initialize DB");
@@ -129,8 +149,13 @@ pub fn find_by_existing_compacted(c: &mut Criterion) {
let data_dir = data_dir_path
.to_str()
.expect("Failed to convert tmpdir path to str");
- let mut db = DB::<Inst>::configure()
- .data_dir(&data_dir)
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
+ .data_dir(data_dir)
.initialize()
.expect("Failed to initialize DB");
diff --git a/log_db/benches/utils.rs b/log_db/benches/utils.rs
index ed114ea..9f29cc6 100644
--- a/log_db/benches/utils.rs
+++ b/log_db/benches/utils.rs
@@ -16,30 +16,30 @@ pub struct Inst {
pub name: String,
pub data: Vec<u8>,
}
-impl Recordable for Inst {
- type Field = Field;
- fn schema() -> Vec<(Field, Type)> {
+
+impl Inst {
+ pub fn schema() -> Vec<(Field, Type)> {
vec![
(Field::Id, Type::int()),
(Field::Name, Type::string()),
(Field::Data, Type::bytes()),
]
}
- fn primary_key() -> Self::Field {
+ pub fn primary_key() -> Field {
Field::Id
}
- fn secondary_keys() -> Vec<Self::Field> {
+ pub fn secondary_keys() -> Vec<Field> {
vec![Field::Name]
}
- fn into_record(self) -> Vec<Value> {
+ pub fn into_record(self) -> Vec<Value> {
vec![
Value::Int(self.id),
Value::String(self.name),
Value::Bytes(self.data),
]
}
- fn from_record(record: Vec<Value>) -> Self {
+ pub fn from_record(record: Vec<Value>) -> Self {
let mut it = record.into_iter();
Inst {
id: match it.next().unwrap() {
@@ -86,7 +86,7 @@ pub fn random_inst(from_id: i64, to_id: i64) -> Inst {
}
pub fn prefill_db(
- db: &mut DB<Inst>,
+ db: &mut DB<Inst, Field>,
insts: &mut Vec<Inst>,
n_records: usize,
compact: bool,
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::*;
diff --git a/log_db/tests/integration.rs b/log_db/tests/integration.rs
index 5376850..4672c06 100644
--- a/log_db/tests/integration.rs
+++ b/log_db/tests/integration.rs
@@ -43,19 +43,18 @@ struct Inst {
pub data: Vec<u8>,
}
-impl Recordable for Inst {
- type Field = Field;
- fn schema() -> Vec<(Self::Field, Type)> {
+impl Inst {
+ fn schema() -> Vec<(Field, Type)> {
vec![
(Field::Id, Type::int()),
(Field::Name, Type::string().nullable()),
(Field::Data, Type::bytes()),
]
}
- fn primary_key() -> Self::Field {
+ fn primary_key() -> Field {
Field::Id
}
- fn secondary_keys() -> Vec<Self::Field> {
+ fn secondary_keys() -> Vec<Field> {
vec![Field::Name]
}
@@ -94,7 +93,12 @@ impl Recordable for Inst {
#[test]
fn test_initialize_only() {
let data_dir = tmp_dir();
- let _db = DB::<Inst>::configure()
+ let _db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -103,7 +107,12 @@ fn test_initialize_only() {
#[test]
fn test_upsert_and_get_with_primary_memtable() {
let data_dir = tmp_dir();
- let mut db = DB::<Inst>::configure()
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -125,7 +134,12 @@ fn test_upsert_and_get_with_primary_memtable() {
#[test]
fn test_upsert_and_get() {
let data_dir = tmp_dir();
- let mut db = DB::<Inst>::configure()
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -177,7 +191,12 @@ fn test_upsert_and_get() {
#[test]
fn test_get_nonexistant() {
let data_dir = tmp_dir();
- let mut db = DB::<Inst>::configure()
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -187,14 +206,16 @@ fn test_get_nonexistant() {
}
struct InstTestNullable {}
-impl Recordable for InstTestNullable {
- type Field = Field;
- fn schema() -> Vec<(Self::Field, Type)> {
+impl InstTestNullable {
+ fn schema() -> Vec<(Field, Type)> {
vec![(Field::Id, Type::int())]
}
- fn primary_key() -> Self::Field {
+ fn primary_key() -> Field {
Field::Id
}
+ fn secondary_keys() -> Vec<Field> {
+ vec![]
+ }
fn into_record(self) -> Vec<Value> {
vec![Value::Null]
@@ -208,7 +229,12 @@ impl Recordable for InstTestNullable {
#[test]
fn test_upsert_fails_on_null_in_non_nullable_field() {
let data_dir = tmp_dir();
- let mut db = DB::<InstTestNullable>::configure()
+ let mut db = DB::configure()
+ .schema(InstTestNullable::schema())
+ .primary_key(InstTestNullable::primary_key())
+ .secondary_keys(InstTestNullable::secondary_keys())
+ .from_record(InstTestNullable::from_record)
+ .into_record(InstTestNullable::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -218,14 +244,16 @@ fn test_upsert_fails_on_null_in_non_nullable_field() {
}
struct InstTestNumValues {}
-impl Recordable for InstTestNumValues {
- type Field = Field;
- fn schema() -> Vec<(Self::Field, Type)> {
+impl InstTestNumValues {
+ fn schema() -> Vec<(Field, Type)> {
vec![(Field::Id, Type::int()), (Field::Name, Type::string())]
}
- fn primary_key() -> Self::Field {
+ fn primary_key() -> Field {
Field::Id
}
+ fn secondary_keys() -> Vec<Field> {
+ vec![]
+ }
fn into_record(self) -> Vec<Value> {
vec![Value::Int(0)]
@@ -239,7 +267,12 @@ impl Recordable for InstTestNumValues {
#[test]
fn test_upsert_fails_on_invalid_number_of_values() {
let data_dir = tmp_dir();
- let mut db = DB::<InstTestNumValues>::configure()
+ let mut db = DB::configure()
+ .schema(InstTestNumValues::schema())
+ .primary_key(InstTestNumValues::primary_key())
+ .secondary_keys(InstTestNumValues::secondary_keys())
+ .from_record(InstTestNumValues::from_record)
+ .into_record(InstTestNumValues::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -249,14 +282,16 @@ fn test_upsert_fails_on_invalid_number_of_values() {
}
struct InstTestInvalidType {}
-impl Recordable for InstTestInvalidType {
- type Field = Field;
- fn schema() -> Vec<(Self::Field, Type)> {
+impl InstTestInvalidType {
+ fn schema() -> Vec<(Field, Type)> {
vec![(Field::Id, Type::int())]
}
- fn primary_key() -> Self::Field {
+ fn primary_key() -> Field {
Field::Id
}
+ fn secondary_keys() -> Vec<Field> {
+ vec![]
+ }
fn into_record(self) -> Vec<Value> {
vec![Value::String("foo".to_string())]
@@ -270,7 +305,12 @@ impl Recordable for InstTestInvalidType {
#[test]
fn test_upsert_fails_on_invalid_value_type() {
let data_dir = tmp_dir();
- let mut db = DB::<InstTestInvalidType>::configure()
+ let mut db = DB::configure()
+ .schema(InstTestInvalidType::schema())
+ .primary_key(InstTestInvalidType::primary_key())
+ .secondary_keys(InstTestInvalidType::secondary_keys())
+ .from_record(InstTestInvalidType::from_record)
+ .into_record(InstTestInvalidType::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -282,7 +322,12 @@ fn test_upsert_fails_on_invalid_value_type() {
#[test]
fn test_upsert_and_find_by() {
let data_dir = tmp_dir();
- let mut db = DB::<Inst>::configure()
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -321,14 +366,16 @@ struct InstSingleId {
pub id: i64,
}
-impl Recordable for InstSingleId {
- type Field = Field;
- fn schema() -> Vec<(Self::Field, Type)> {
+impl InstSingleId {
+ fn schema() -> Vec<(Field, Type)> {
vec![(Field::Id, Type::int())]
}
- fn primary_key() -> Self::Field {
+ fn primary_key() -> Field {
Field::Id
}
+ fn secondary_keys() -> Vec<Field> {
+ vec![]
+ }
fn into_record(self) -> Vec<Value> {
vec![Value::Int(self.id)]
@@ -355,7 +402,12 @@ 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::<InstSingleId>::configure()
+ let mut db = DB::configure()
+ .schema(InstSingleId::schema())
+ .primary_key(InstSingleId::primary_key())
+ .secondary_keys(InstSingleId::secondary_keys())
+ .from_record(InstSingleId::from_record)
+ .into_record(InstSingleId::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -370,7 +422,12 @@ fn test_multiple_writing_threads() {
}
// Read the records
- let mut db = DB::<InstSingleId>::configure()
+ let mut db = DB::configure()
+ .schema(InstSingleId::schema())
+ .primary_key(InstSingleId::primary_key())
+ .secondary_keys(InstSingleId::secondary_keys())
+ .from_record(InstSingleId::from_record)
+ .into_record(InstSingleId::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -396,7 +453,12 @@ 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::<InstSingleId>::configure()
+ let mut db = DB::configure()
+ .schema(InstSingleId::schema())
+ .primary_key(InstSingleId::primary_key())
+ .secondary_keys(InstSingleId::secondary_keys())
+ .from_record(InstSingleId::from_record)
+ .into_record(InstSingleId::into_record)
.data_dir(&data_dir)
.segment_size(1000) // should cause rotations
.initialize()
@@ -422,7 +484,12 @@ fn test_one_writer_and_multiple_reading_threads() {
// Add a writer that inserts the records
threads.push(thread::spawn(move || {
- let mut db = DB::<InstSingleId>::configure()
+ let mut db = DB::configure()
+ .schema(InstSingleId::schema())
+ .primary_key(InstSingleId::primary_key())
+ .secondary_keys(InstSingleId::secondary_keys())
+ .from_record(InstSingleId::from_record)
+ .into_record(InstSingleId::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -452,7 +519,12 @@ fn test_log_is_rotated_when_capacity_reached() {
+ (1 + 8 + 4) // string tag + string length + string data
+ (1 + 8 + 3); // bytes tag + bytes length + bytes data
- let mut db = DB::<Inst>::configure()
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
.data_dir(&data_dir)
.segment_size(10 * record_len) // small log segment size
.initialize()
@@ -482,7 +554,12 @@ fn test_log_is_rotated_when_capacity_reached() {
#[test]
fn test_delete() {
let data_dir = tmp_dir();
- let mut db = DB::<Inst>::configure()
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -519,7 +596,12 @@ fn test_delete() {
#[test]
fn test_delete_by() {
let data_dir = tmp_dir();
- let mut db = DB::<Inst>::configure()
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -571,7 +653,12 @@ fn test_delete_by() {
#[test]
fn test_range_by_id() {
let data_dir = tmp_dir();
- let mut db = DB::<Inst>::configure()
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -619,7 +706,12 @@ fn test_range_by_id() {
#[test]
fn test_batch_find_by() {
let data_dir = tmp_dir();
- let mut db = DB::<Inst>::configure()
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -652,7 +744,12 @@ fn test_batch_find_by() {
#[test]
fn test_commit_transaction() {
let data_dir = tmp_dir();
- let mut db = DB::<Inst>::configure()
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -685,7 +782,12 @@ fn test_commit_transaction() {
#[test]
fn test_rollback_transaction() {
let data_dir = tmp_dir();
- let mut db = DB::<Inst>::configure()
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -730,10 +832,8 @@ struct InstWithNewNullableField {
pub maybe_str: Option<String>,
}
-impl Recordable for InstWithNewNullableField {
- type Field = FieldWithNewNullableField;
-
- fn schema() -> Vec<(Self::Field, Type)> {
+impl InstWithNewNullableField {
+ fn schema() -> Vec<(FieldWithNewNullableField, Type)> {
vec![
(FieldWithNewNullableField::Id, Type::int()),
(FieldWithNewNullableField::Name, Type::string().nullable()),
@@ -745,11 +845,11 @@ impl Recordable for InstWithNewNullableField {
]
}
- fn primary_key() -> Self::Field {
+ fn primary_key() -> FieldWithNewNullableField {
FieldWithNewNullableField::Id
}
- fn secondary_keys() -> Vec<Self::Field> {
+ fn secondary_keys() -> Vec<FieldWithNewNullableField> {
vec![FieldWithNewNullableField::Name]
}
@@ -801,7 +901,12 @@ fn test_add_nullable_field() {
// Insert a record with 3 fields
{
- let mut db = DB::<Inst>::configure()
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -815,7 +920,12 @@ fn test_add_nullable_field() {
}
// Insert a record with 4 fields (last is nullable)
- let mut db = DB::<InstWithNewNullableField>::configure()
+ let mut db = DB::configure()
+ .schema(InstWithNewNullableField::schema())
+ .primary_key(InstWithNewNullableField::primary_key())
+ .secondary_keys(InstWithNewNullableField::secondary_keys())
+ .from_record(InstWithNewNullableField::from_record)
+ .into_record(InstWithNewNullableField::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -844,7 +954,12 @@ fn test_add_non_nullable_field() {
// Insert a record with just one field
{
- let mut db = DB::<InstSingleId>::configure()
+ let mut db = DB::configure()
+ .schema(InstSingleId::schema())
+ .primary_key(InstSingleId::primary_key())
+ .secondary_keys(InstSingleId::secondary_keys())
+ .from_record(InstSingleId::from_record)
+ .into_record(InstSingleId::into_record)
.data_dir(&data_dir)
.initialize()
.expect("Failed to initialize DB instance");
@@ -854,7 +969,15 @@ fn test_add_non_nullable_field() {
// Configure the DB with three fields, one of which is non-nullable
// This should fail
- match DB::<Inst>::configure().data_dir(&data_dir).initialize() {
+ match DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
+ .data_dir(&data_dir)
+ .initialize()
+ {
Ok(_) => panic!("Expected initialization to fail"),
Err(DBError::ValidationError(e)) => {
// Expected
@@ -863,3 +986,42 @@ fn test_add_non_nullable_field() {
Err(e) => panic!("Unexpected error: {:?}", e),
}
}
+
+#[test]
+fn test_delete_by_multiple_indexes() {
+ let data_dir = tmp_dir();
+
+ let mut db = DB::configure()
+ .schema(Inst::schema())
+ .primary_key(Inst::primary_key())
+ .secondary_keys(Inst::secondary_keys())
+ .from_record(Inst::from_record)
+ .into_record(Inst::into_record)
+ .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);
+}

Want to contribute? Host your changes somewhere and send me the link: jan (at) sydanviita.fi. Also, check out my website.


forge.jan.systems — powered by cgit