aboutsummaryrefslogtreecommitdiffstats
path: root/log_db
diff options
context:
space:
mode:
Diffstat (limited to 'log_db')
-rw-r--r--log_db/benches/benchmark.rs10
-rw-r--r--log_db/benches/utils.rs40
-rw-r--r--log_db/src/config.rs44
-rw-r--r--log_db/src/engine.rs61
-rw-r--r--log_db/src/lib.rs112
-rw-r--r--log_db/src/record.rs38
-rw-r--r--log_db/src/row.rs15
-rw-r--r--log_db/src/schema.rs7
-rw-r--r--log_db/tests/integration.rs217
9 files changed, 257 insertions, 287 deletions
diff --git a/log_db/benches/benchmark.rs b/log_db/benches/benchmark.rs
index aa4b1bb..d8e6280 100644
--- a/log_db/benches/benchmark.rs
+++ b/log_db/benches/benchmark.rs
@@ -18,8 +18,6 @@ pub fn upsert_compacted(c: &mut Criterion) {
.fields(Inst::fields())
.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");
@@ -50,8 +48,6 @@ pub fn delete_existing_compacted(c: &mut Criterion) {
.fields(Inst::fields())
.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");
@@ -89,8 +85,6 @@ pub fn upsert_write_durability(c: &mut Criterion) {
.fields(Inst::fields())
.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()
@@ -117,8 +111,6 @@ pub fn get_existing_compacted(c: &mut Criterion) {
.fields(Inst::fields())
.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");
@@ -153,8 +145,6 @@ pub fn find_by_existing_compacted(c: &mut Criterion) {
.fields(Inst::fields())
.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 d25a5af..33a45cf 100644
--- a/log_db/benches/utils.rs
+++ b/log_db/benches/utils.rs
@@ -33,25 +33,19 @@ impl Into<String> for Field {
}
}
-impl Inst {
- pub fn fields() -> Vec<Field> {
- vec![Field::Id, Field::Name, Field::Data]
- }
- pub fn primary_key() -> Field {
- Field::Id
- }
- pub fn secondary_keys() -> Vec<Field> {
- vec![Field::Name]
- }
-
- pub fn into_record(self) -> Vec<Value> {
+impl From<Inst> for Record {
+ fn from(inst: Inst) -> Self {
vec![
- Value::Int(self.id),
- Value::String(self.name),
- Value::Bytes(self.data),
+ Value::Int(inst.id),
+ Value::String(inst.name),
+ Value::Bytes(inst.data),
]
+ .into()
}
- pub fn from_record(record: Vec<Value>) -> Self {
+}
+
+impl From<Record> for Inst {
+ fn from(record: Record) -> Self {
let mut it = record.into_iter();
Inst {
id: match it.next().unwrap() {
@@ -70,6 +64,18 @@ impl Inst {
}
}
+impl Inst {
+ pub fn fields() -> Vec<Field> {
+ vec![Field::Id, Field::Name, Field::Data]
+ }
+ pub fn primary_key() -> Field {
+ Field::Id
+ }
+ pub fn secondary_keys() -> Vec<Field> {
+ vec![Field::Name]
+ }
+}
+
// Function to generate a random integer
pub fn random_int(from: i64, to: i64) -> i64 {
let mut rng = rand::thread_rng();
@@ -98,7 +104,7 @@ pub fn random_inst(from_id: i64, to_id: i64) -> Inst {
}
pub fn prefill_db(
- db: &mut DB<Inst>,
+ db: &mut DB,
insts: &mut Vec<Inst>,
n_records: usize,
compact: bool,
diff --git a/log_db/src/config.rs b/log_db/src/config.rs
index ec74164..e8c6fa1 100644
--- a/log_db/src/config.rs
+++ b/log_db/src/config.rs
@@ -1,12 +1,6 @@
use super::*;
-pub struct Schema<F> {
- pub fields: Vec<F>,
- pub primary_key: F,
- pub secondary_keys: Vec<F>,
-}
-
-pub struct ConfigBuilder<T> {
+pub struct ConfigBuilder {
data_dir: Option<String>,
segment_size: Option<usize>,
write_durability: Option<WriteDurability>,
@@ -15,14 +9,10 @@ pub struct ConfigBuilder<T> {
fields: Option<Vec<String>>,
primary_key: Option<String>,
secondary_keys: Option<Vec<String>>,
- from_record: Option<fn(Vec<Value>) -> T>,
- into_record: Option<fn(T) -> Vec<Value>>,
-
- _marker: PhantomData<T>,
}
-impl<T> ConfigBuilder<T> {
- pub fn new() -> ConfigBuilder<T> {
+impl ConfigBuilder {
+ pub fn new() -> ConfigBuilder {
ConfigBuilder {
data_dir: None,
segment_size: None,
@@ -32,10 +22,6 @@ impl<T> ConfigBuilder<T> {
fields: None,
primary_key: None,
secondary_keys: None,
- from_record: None,
- into_record: None,
-
- _marker: PhantomData,
}
}
@@ -86,36 +72,18 @@ impl<T> ConfigBuilder<T> {
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>> {
+ pub fn initialize(self) -> DBResult<DB> {
let schema = self
.fields
.ok_or_else(|| DBError::ValidationError("Schema not set".to_string()))?;
let primary_key = self
.primary_key
.ok_or_else(|| DBError::ValidationError("Primary key not set".to_string()))?;
- let 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 {
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
@@ -134,12 +102,10 @@ impl<T> ConfigBuilder<T> {
}
#[derive(Clone)]
-pub struct Config<T> {
+pub struct Config {
pub schema: Vec<String>,
pub primary_key: String,
pub secondary_keys: Vec<String>,
- 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 b5ece69..0701191 100644
--- a/log_db/src/engine.rs
+++ b/log_db/src/engine.rs
@@ -1,7 +1,7 @@
use super::*;
-pub struct Engine<T> {
- pub config: Config<T>,
+pub struct Engine {
+ pub config: Config,
pub lock_manager: LockManager,
data_dir_path: PathBuf,
@@ -19,8 +19,8 @@ pub struct Engine<T> {
pub secondary_memtables: Vec<SecondaryMemtable>,
}
-impl<T> Engine<T> {
- pub fn initialize(config: Config<T>) -> DBResult<Engine<T>> {
+impl Engine {
+ pub fn initialize(config: Config) -> DBResult<Engine> {
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.
@@ -104,7 +104,7 @@ impl<T> Engine<T> {
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::<T> {
+ let mut engine = Engine {
config,
lock_manager,
data_dir_path,
@@ -161,9 +161,9 @@ impl<T> Engine<T> {
let log_key = LogKey::new(segnum, index);
if row.tombstone {
- self.remove_record_from_memtables(&row);
+ self.remove_row_from_memtables(&row.values);
} else {
- self.insert_record_to_memtables(log_key, row);
+ self.insert_row_to_memtables(log_key, row.values);
}
// Update from_index in case this is the last iteration: we need to know the next
@@ -183,8 +183,8 @@ impl<T> Engine<T> {
Ok(())
}
- fn insert_record_to_memtables(&mut self, log_key: LogKey, record: Row) {
- let pk = record.at(self.primary_key_index).as_indexable().unwrap();
+ fn insert_row_to_memtables(&mut self, log_key: LogKey, row_values: Vec<Value>) {
+ let pk = row_values[self.primary_key_index].as_indexable().unwrap();
for (sk_index, sk_field) in self.config.secondary_keys.iter().enumerate() {
let secondary_memtable = &mut self.secondary_memtables[sk_index];
@@ -194,7 +194,7 @@ impl<T> Engine<T> {
.iter()
.position(|f| sk_field == f)
.unwrap();
- let sk = record.at(sk_field_index).as_indexable().unwrap();
+ let sk = row_values[sk_field_index].as_indexable().unwrap();
secondary_memtable.set(pk.clone(), sk, log_key.clone());
}
@@ -203,8 +203,8 @@ impl<T> Engine<T> {
self.primary_memtable.set(pk, log_key);
}
- fn remove_record_from_memtables(&mut self, record: &Row) {
- let pk = record.at(self.primary_key_index).as_indexable().unwrap();
+ fn remove_row_from_memtables(&mut self, row_values: &Vec<Value>) {
+ let pk = row_values[self.primary_key_index].as_indexable().unwrap();
if let Some(_) = self.primary_memtable.remove(&pk) {
for (sk_index, sk_field) in self.config.secondary_keys.iter_mut().enumerate() {
@@ -215,7 +215,7 @@ impl<T> Engine<T> {
.iter()
.position(|f| sk_field == f)
.unwrap();
- let sk = record.at(sk_field_index).as_indexable().unwrap();
+ let sk = row_values[sk_field_index].as_indexable().unwrap();
secondary_memtable.remove(&pk, &sk);
}
@@ -235,7 +235,7 @@ impl<T> Engine<T> {
return self.upsert_record(record);
}
- self.tx_log.push(TxEntry::Upsert { record });
+ self.tx_log.push(TxEntry::Upsert { row: record });
if !self.tx_active {
self.commit_transaction()?;
@@ -471,7 +471,7 @@ impl<T> Engine<T> {
// TODO: refactor the clone out of here
for record in &recs {
self.tx_log.push(TxEntry::Delete {
- record: record.clone(),
+ row: record.clone(),
});
}
@@ -501,8 +501,8 @@ impl<T> Engine<T> {
debug!("Serializing tx_log to byte arrays");
for tx_entry in &self.tx_log {
let record = match tx_entry {
- TxEntry::Upsert { record } => record,
- TxEntry::Delete { record } => record,
+ TxEntry::Upsert { row: record } => record,
+ TxEntry::Delete { row: record } => record,
};
let serialized = record.serialize();
@@ -544,8 +544,8 @@ impl<T> Engine<T> {
debug!("Updating memtables");
for (log_key, tx_entry) in pending_memtable_ops {
match tx_entry {
- TxEntry::Upsert { record } => self.insert_record_to_memtables(log_key, record),
- TxEntry::Delete { record } => self.remove_record_from_memtables(&record),
+ TxEntry::Upsert { row } => self.insert_row_to_memtables(log_key, row.values),
+ TxEntry::Delete { row } => self.remove_row_from_memtables(&row.values),
}
}
debug!("Commit done");
@@ -580,8 +580,7 @@ impl<T> Engine<T> {
)
.map(|item| {
(
- item.row
- .at(self.primary_key_index)
+ item.row.values[self.primary_key_index]
.as_indexable()
.expect("Primary key was not indexable"),
item.row,
@@ -752,12 +751,14 @@ mod tests {
name: String,
}
- impl TestInst2 {
- fn into_record(self) -> Vec<Value> {
- vec![Value::Int(self.id), Value::String(self.name)]
+ impl From<TestInst2> for Vec<Value> {
+ fn from(inst: TestInst2) -> Self {
+ vec![Value::Int(inst.id), Value::String(inst.name)]
}
+ }
- fn from_record(record: Vec<Value>) -> Self {
+ impl From<Vec<Value>> for TestInst2 {
+ fn from(record: Vec<Value>) -> Self {
let mut it = record.into_iter();
TestInst2 {
id: match it.next().unwrap() {
@@ -785,8 +786,6 @@ mod tests {
.fields(vec![Field::Id, Field::Name])
.primary_key(Field::Id)
.secondary_keys(vec![Field::Name])
- .from_record(TestInst2::from_record)
- .into_record(TestInst2::into_record)
.segment_size(segment_size)
.initialize()
.expect("Failed to create DB");
@@ -806,8 +805,7 @@ mod tests {
.len(),
0
);
- engine
- .insert_record_to_memtables(LogKey::new(1, 0), Row::from(&inst.clone().into_record()));
+ engine.insert_row_to_memtables(LogKey::new(1, 0), inst.clone().into());
assert_eq!(engine.primary_memtable.get(&id), Some(&LogKey::new(1, 0)));
assert_eq!(
engine.secondary_memtables[0]
@@ -816,8 +814,7 @@ mod tests {
1
);
- engine
- .insert_record_to_memtables(LogKey::new(1, 1), Row::from(&inst.clone().into_record()));
+ engine.insert_row_to_memtables(LogKey::new(1, 1), inst.clone().into());
assert_eq!(engine.primary_memtable.get(&id), Some(&LogKey::new(1, 1)));
assert_eq!(
engine.secondary_memtables[0]
@@ -826,7 +823,7 @@ mod tests {
1
);
- engine.remove_record_from_memtables(&Row::from(&inst.into_record()));
+ engine.remove_row_from_memtables(&inst.into());
assert_eq!(engine.primary_memtable.get(&id), None);
assert_eq!(
engine.secondary_memtables[0]
diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs
index b7229fe..006056c 100644
--- a/log_db/src/lib.rs
+++ b/log_db/src/lib.rs
@@ -8,7 +8,6 @@ use std::fmt::Debug;
use std::fmt::Display;
use std::fs::{self, metadata, File};
use std::io::{self, Read, Seek, SeekFrom, Write};
-use std::marker::PhantomData;
use std::ops::*;
use std::path::{Path, PathBuf};
use std::thread;
@@ -23,10 +22,14 @@ mod lock;
mod log_reader_forward;
mod memtable_primary;
mod memtable_secondary;
+mod record;
mod row;
+mod schema;
pub use common::{DBError, DBResult, OwnedBounds, QueryParams, Value, DEFAULT_QUERY_PARAMS};
-pub use config::{ReadConsistency, Schema, WriteDurability};
+pub use config::{ReadConsistency, WriteDurability};
+pub use record::Record;
+pub use schema::Schema;
use common::*;
use config::*;
@@ -37,25 +40,28 @@ use memtable_primary::PrimaryMemtable;
use memtable_secondary::SecondaryMemtable;
use row::*;
-pub struct DB<T> {
- engine: Engine<T>,
+pub struct DB {
+ engine: Engine,
}
-impl<T> DB<T> {
+impl DB {
/// Create a new database configuration builder.
- pub fn configure() -> ConfigBuilder<T> {
+ pub fn configure() -> ConfigBuilder {
ConfigBuilder::new()
}
- fn initialize(config: Config<T>) -> DBResult<DB<T>> {
+ fn initialize(config: Config) -> DBResult<DB> {
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: T) -> DBResult<()> {
- let row = Row::from(&(self.engine.config.into_record)(recordable));
+ pub fn upsert(&mut self, record: impl Into<Record>) -> DBResult<()> {
+ let row = Row {
+ values: record.into().into(),
+ tombstone: false,
+ };
debug!("Upserting record: {:?}", row);
self.engine
@@ -66,7 +72,7 @@ impl<T> DB<T> {
/// Get a record by its primary index value.
/// E.g. `db.get(Value::Int(10))`.
- pub fn get(&mut self, value: &Value) -> DBResult<Option<T>> {
+ pub fn get(&mut self, value: &Value) -> DBResult<Option<Record>> {
let tagged_rows = self.engine.with_shared_lock(|engine| {
engine.batch_find_by_records(
// TODO: This clone is only here to appease the borrow checker
@@ -81,12 +87,12 @@ impl<T> DB<T> {
Ok(tagged_rows
.into_iter()
.next()
- .map(|(_, row)| (self.engine.config.from_record)(row.values)))
+ .map(|(_, row)| Record::from(row)))
}
/// Get a collection of records based on an indexed field value.
- pub fn find_by(&mut self, field: impl AsRef<str>, value: &Value) -> DBResult<Vec<T>> {
- let recs = self.engine.with_shared_lock(|engine| {
+ pub fn find_by(&mut self, field: impl AsRef<str>, value: &Value) -> DBResult<Vec<Record>> {
+ let tagged_rows = self.engine.with_shared_lock(|engine| {
engine.batch_find_by_records(
field.as_ref(),
std::iter::once(value),
@@ -94,9 +100,9 @@ impl<T> DB<T> {
)
})?;
- Ok(recs
+ Ok(tagged_rows
.into_iter()
- .map(|(_, rec)| (self.engine.config.from_record)(rec.values))
+ .map(|(_, row)| Record::from(row))
.collect())
}
@@ -106,15 +112,12 @@ impl<T> DB<T> {
field: impl AsRef<str>,
value: &Value,
params: &QueryParams,
- ) -> DBResult<Vec<T>> {
+ ) -> DBResult<Vec<Record>> {
let recs = self.engine.with_shared_lock(|engine| {
engine.batch_find_by_records(field.as_ref(), std::iter::once(value), params)
})?;
- Ok(recs
- .into_iter()
- .map(|(_, rec)| (self.engine.config.from_record)(rec.values))
- .collect())
+ Ok(recs.into_iter().map(|(_, row)| Record::from(row)).collect())
}
/// Get a collection of records based on a sequence of indexed field values.
@@ -124,14 +127,14 @@ impl<T> DB<T> {
&mut self,
field: impl Into<String>,
values: &[Value],
- ) -> DBResult<Vec<(usize, T)>> {
+ ) -> DBResult<Vec<(usize, Record)>> {
let recs = self.engine.with_shared_lock(|engine| {
engine.batch_find_by_records(&field.into(), values.iter(), &DEFAULT_QUERY_PARAMS)
})?;
Ok(recs
.into_iter()
- .map(|(tag, rec)| (tag, (self.engine.config.from_record)(rec.values)))
+ .map(|(tag, row)| (tag, Record::from(row)))
.collect())
}
@@ -143,14 +146,14 @@ impl<T> DB<T> {
field: impl AsRef<str>,
values: &[Value],
params: &QueryParams,
- ) -> DBResult<Vec<(usize, T)>> {
+ ) -> DBResult<Vec<(usize, Record)>> {
let recs = self.engine.with_shared_lock(|engine| {
engine.batch_find_by_records(field.as_ref(), values.iter(), params)
})?;
Ok(recs
.into_iter()
- .map(|(tag, rec)| (tag, (self.engine.config.from_record)(rec.values)))
+ .map(|(tag, row)| (tag, Record::from(row)))
.collect())
}
@@ -161,15 +164,12 @@ impl<T> DB<T> {
&mut self,
field: impl AsRef<str>,
range: B,
- ) -> DBResult<Vec<T>> {
+ ) -> DBResult<Vec<Record>> {
let recs = self.engine.with_shared_lock(|engine| {
engine.range_by_records(field.as_ref(), range, &DEFAULT_QUERY_PARAMS)
})?;
- Ok(recs
- .into_iter()
- .map(|rec| (self.engine.config.from_record)(rec.values))
- .collect())
+ Ok(recs.into_iter().map(|row| Record::from(row)).collect())
}
/// Get a collection of records based on a range of indexed field values, with additional parameters.
@@ -180,15 +180,12 @@ impl<T> DB<T> {
field: impl AsRef<str>,
range: B,
params: &QueryParams,
- ) -> DBResult<Vec<T>> {
+ ) -> DBResult<Vec<Record>> {
let recs = self
.engine
.with_shared_lock(|engine| engine.range_by_records(field.as_ref(), range, params))?;
- Ok(recs
- .into_iter()
- .map(|rec| (self.engine.config.from_record)(rec.values))
- .collect())
+ Ok(recs.into_iter().map(|row| Record::from(row)).collect())
}
/// Delete records by a field value.
@@ -197,19 +194,19 @@ impl<T> DB<T> {
///
/// Deletion is done by marking the record as a tombstone. The record will still be present in the log file,
/// but will be ignored by reads. Upon compaction, tombstoned records will be removed.
- pub fn delete_by(&mut self, field: impl AsRef<str>, value: &Value) -> DBResult<Vec<T>> {
+ pub fn delete_by(&mut self, field: impl AsRef<str>, value: &Value) -> DBResult<Vec<Record>> {
let recs = self
.engine
.with_exclusive_lock(|engine| engine.delete_by_field(field.as_ref(), value))?;
Ok(recs
.into_iter()
- .map(|rec| (self.engine.config.from_record)(rec.values))
+ .map(|row| Record::from(row.values))
.collect())
}
/// Delete record by primary key.
- pub fn delete(&mut self, pk: &Value) -> DBResult<Option<T>> {
+ pub fn delete(&mut self, pk: &Value) -> DBResult<Option<Record>> {
let recs = self.engine.with_exclusive_lock(|engine| {
engine
// TODO: This clone is only here to appease the borrow checker
@@ -218,10 +215,7 @@ impl<T> DB<T> {
assert!(recs.len() <= 1);
- Ok(recs
- .into_iter()
- .next()
- .map(|rec| (self.engine.config.from_record)(rec.values)))
+ Ok(recs.into_iter().next().map(|row| Record::from(row.values)))
}
/// Check if there are any pending tasks and do them. Tasks include:
@@ -320,12 +314,14 @@ mod tests {
id: i64,
}
- impl TestInst1 {
- fn into_record(self) -> Vec<Value> {
- vec![Value::Int(self.id)]
+ impl From<TestInst1> for Record {
+ fn from(inst: TestInst1) -> Self {
+ vec![Value::Int(inst.id)].into()
}
+ }
- fn from_record(record: Vec<Value>) -> Self {
+ impl From<Record> for TestInst1 {
+ fn from(record: Record) -> Self {
let mut it = record.into_iter();
TestInst1 {
id: match it.next().unwrap() {
@@ -341,12 +337,14 @@ mod tests {
name: String,
}
- impl TestInst2 {
- fn into_record(self) -> Vec<Value> {
- vec![Value::Int(self.id), Value::String(self.name)]
+ impl From<TestInst2> for Record {
+ fn from(inst: TestInst2) -> Self {
+ vec![Value::Int(inst.id), Value::String(inst.name)].into()
}
+ }
- fn from_record(record: Vec<Value>) -> Self {
+ impl From<Record> for TestInst2 {
+ fn from(record: Record) -> Self {
let mut it = record.into_iter();
TestInst2 {
id: match it.next().unwrap() {
@@ -373,8 +371,6 @@ mod tests {
.data_dir(data_dir.to_str().unwrap())
.fields(vec![Field::Id])
.primary_key(Field::Id)
- .from_record(TestInst1::from_record)
- .into_record(TestInst1::into_record)
.segment_size(segment_size)
.initialize()
.expect("Failed to create DB");
@@ -436,17 +432,19 @@ mod tests {
);
// Check that the records can be read
- let inst0 = db
+ let inst0: TestInst1 = db
.get(&Value::Int(0 as i64))
.expect("Failed to get record")
- .expect("Record not found");
+ .expect("Record not found")
+ .into();
assert!(inst0.id == 0);
- let inst1 = db
+ let inst1: TestInst1 = db
.get(&Value::Int(1 as i64))
.expect("Failed to get record")
- .expect("Record not found");
+ .expect("Record not found")
+ .into();
assert!(inst1.id == 1);
}
@@ -460,8 +458,6 @@ mod tests {
.data_dir(data_dir.to_str().unwrap())
.fields(vec![Field::Id])
.primary_key(Field::Id)
- .from_record(TestInst1::from_record)
- .into_record(TestInst1::into_record)
.initialize()
.expect("Failed to create DB");
@@ -515,8 +511,6 @@ mod tests {
.fields(vec![Field::Id, Field::Name])
.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
new file mode 100644
index 0000000..b349853
--- /dev/null
+++ b/log_db/src/record.rs
@@ -0,0 +1,38 @@
+use super::*;
+
+pub struct Record {
+ values: Vec<Value>,
+}
+
+impl Record {
+ pub fn values(&self) -> &[Value] {
+ &self.values
+ }
+}
+
+impl IntoIterator for Record {
+ type Item = Value;
+ type IntoIter = std::vec::IntoIter<Self::Item>;
+
+ fn into_iter(self) -> Self::IntoIter {
+ self.values.into_iter()
+ }
+}
+
+impl From<Vec<Value>> for Record {
+ fn from(values: Vec<Value>) -> Self {
+ Record { values }
+ }
+}
+
+impl From<Record> for Vec<Value> {
+ fn from(record: Record) -> Self {
+ record.values
+ }
+}
+
+impl From<Row> for Record {
+ fn from(row: Row) -> Self {
+ Record { values: row.values }
+ }
+}
diff --git a/log_db/src/row.rs b/log_db/src/row.rs
index 5ebb069..d8140d2 100644
--- a/log_db/src/row.rs
+++ b/log_db/src/row.rs
@@ -37,17 +37,6 @@ impl Row {
}
Row { values, tombstone }
}
-
- pub fn from(values: &[Value]) -> Row {
- Row {
- values: values.to_vec(),
- tombstone: false,
- }
- }
-
- pub fn at(&self, index: usize) -> &Value {
- &self.values[index]
- }
}
#[cfg(test)]
@@ -76,6 +65,6 @@ mod tests {
#[derive(Clone, Debug)]
pub enum TxEntry {
- Upsert { record: Row },
- Delete { record: Row },
+ Upsert { row: Row },
+ Delete { row: Row },
}
diff --git a/log_db/src/schema.rs b/log_db/src/schema.rs
new file mode 100644
index 0000000..98d0df7
--- /dev/null
+++ b/log_db/src/schema.rs
@@ -0,0 +1,7 @@
+use super::*;
+
+pub struct Schema {
+ pub fields: Vec<String>,
+ pub primary_key: String,
+ pub secondary_keys: Vec<String>,
+}
diff --git a/log_db/tests/integration.rs b/log_db/tests/integration.rs
index af12fb4..e7c6d73 100644
--- a/log_db/tests/integration.rs
+++ b/log_db/tests/integration.rs
@@ -46,9 +46,9 @@ impl AsRef<str> for Field {
}
}
-impl Into<String> for Field {
- fn into(self) -> String {
- self.as_ref().to_owned()
+impl From<Field> for String {
+ fn from(field: Field) -> String {
+ field.as_ref().to_owned()
}
}
@@ -59,31 +59,23 @@ struct Inst {
pub data: Vec<u8>,
}
-impl Inst {
- fn schema() -> Vec<Field> {
- vec![Field::Id, Field::Name, Field::Data]
- }
- fn primary_key() -> Field {
- Field::Id
- }
- fn secondary_keys() -> Vec<Field> {
- vec![Field::Name]
- }
-
- fn into_record(self) -> Vec<Value> {
+impl From<Inst> for Record {
+ fn from(inst: Inst) -> Record {
vec![
- Value::Int(self.id),
- match self.name {
+ Value::Int(inst.id),
+ match inst.name {
Some(name) => Value::String(name),
None => Value::Null,
},
- Value::Bytes(self.data),
+ Value::Bytes(inst.data),
]
+ .into()
}
+}
- fn from_record(record: Vec<Value>) -> Self {
+impl From<Record> for Inst {
+ fn from(record: Record) -> Self {
let mut it = record.into_iter();
-
Inst {
id: match it.next().unwrap() {
Value::Int(id) => id,
@@ -102,6 +94,18 @@ impl Inst {
}
}
+impl Inst {
+ fn schema() -> Vec<Field> {
+ vec![Field::Id, Field::Name, Field::Data]
+ }
+ fn primary_key() -> Field {
+ Field::Id
+ }
+ fn secondary_keys() -> Vec<Field> {
+ vec![Field::Name]
+ }
+}
+
#[test]
fn test_initialize_only() {
let data_dir = tmp_dir();
@@ -109,8 +113,6 @@ fn test_initialize_only() {
.fields(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");
@@ -123,8 +125,6 @@ fn test_upsert_and_get_with_primary_memtable() {
.fields(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");
@@ -137,7 +137,7 @@ fn test_upsert_and_get_with_primary_memtable() {
};
db.upsert(inst).unwrap();
- let result = db.get(&Value::Int(1)).unwrap().unwrap();
+ let result: Inst = db.get(&Value::Int(1)).unwrap().unwrap().into();
// Check that the IDs match
assert!(result.id == id);
@@ -150,8 +150,6 @@ fn test_upsert_and_get() {
.fields(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");
@@ -186,14 +184,14 @@ fn test_upsert_and_get() {
.unwrap();
// Get with ID = 0
- let result = db.get(&Value::Int(0)).unwrap().unwrap();
+ let result: Inst = db.get(&Value::Int(0)).unwrap().unwrap().into();
// Should match id == 0
assert!(result.id == 0);
assert!(result.name == None);
// Get with ID = 1
- let result = db.get(&Value::Int(1)).unwrap().unwrap();
+ let result: Inst = db.get(&Value::Int(1)).unwrap().unwrap().into();
// Should match newest inst with id == 1
assert!(result.id == 1);
@@ -207,8 +205,6 @@ fn test_get_nonexistant() {
.fields(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");
@@ -224,8 +220,6 @@ fn test_upsert_and_find_by() {
.fields(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");
@@ -264,6 +258,24 @@ struct InstSingleId {
pub id: i64,
}
+impl From<InstSingleId> for Record {
+ fn from(inst: InstSingleId) -> Record {
+ vec![Value::Int(inst.id)].into()
+ }
+}
+
+impl From<Record> for InstSingleId {
+ fn from(record: Record) -> Self {
+ let mut it = record.into_iter();
+ InstSingleId {
+ id: match it.next().unwrap() {
+ Value::Int(id) => id,
+ other => panic!("Invalid value type: {:?}", other),
+ },
+ }
+ }
+}
+
impl InstSingleId {
fn schema() -> Vec<Field> {
vec![Field::Id]
@@ -274,19 +286,6 @@ impl InstSingleId {
fn secondary_keys() -> Vec<Field> {
vec![]
}
-
- fn into_record(self) -> Vec<Value> {
- vec![Value::Int(self.id)]
- }
-
- fn from_record(record: Vec<Value>) -> Self {
- Self {
- id: match record[0] {
- Value::Int(id) => id,
- _ => panic!("Invalid value type"),
- },
- }
- }
}
#[test]
@@ -304,8 +303,6 @@ fn test_multiple_writing_threads() {
.fields(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");
@@ -324,17 +321,16 @@ fn test_multiple_writing_threads() {
.fields(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");
for i in 0..threads_n {
- let result = db
+ let result: InstSingleId = db
.get(&Value::Int(i))
.expect("Failed to get record")
- .expect("Record not found");
+ .expect("Record not found")
+ .into();
assert!(result.id == i);
}
@@ -355,8 +351,6 @@ fn test_one_writer_and_multiple_reading_threads() {
.fields(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()
@@ -372,6 +366,7 @@ fn test_one_writer_and_multiple_reading_threads() {
continue;
}
Some(result) => {
+ let result: InstSingleId = result.into();
assert!(result.id == i);
break;
}
@@ -386,8 +381,6 @@ fn test_one_writer_and_multiple_reading_threads() {
.fields(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");
@@ -421,8 +414,6 @@ fn test_log_is_rotated_when_capacity_reached() {
.fields(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()
@@ -456,8 +447,6 @@ fn test_delete() {
.fields(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");
@@ -498,8 +487,6 @@ fn test_delete_by() {
.fields(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");
@@ -555,8 +542,6 @@ fn test_range_by_id() {
.fields(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");
@@ -576,7 +561,7 @@ fn test_range_by_id() {
let received = db
.range_by(&Field::Id, &Value::Int(3)..&Value::Int(7))
.unwrap();
- let received_ids: Vec<i64> = received.iter().map(|inst| inst.id).collect();
+ let received_ids: Vec<i64> = received.into_iter().map(|rec| Inst::from(rec).id).collect();
assert_eq!(received_ids, vec![3, 4, 5, 6]);
@@ -584,19 +569,19 @@ fn test_range_by_id() {
let received = db
.range_by(&Field::Id, &Value::Int(3)..=&Value::Int(7))
.unwrap();
- let received_ids: Vec<i64> = received.iter().map(|inst| inst.id).collect();
+ let received_ids: Vec<i64> = received.into_iter().map(|rec| Inst::from(rec).id).collect();
assert_eq!(received_ids, vec![3, 4, 5, 6, 7]);
// Test range (-inf, 7]
let received = db.range_by(&Field::Id, ..=&Value::Int(7)).unwrap();
- let received_ids: Vec<i64> = received.iter().map(|inst| inst.id).collect();
+ let received_ids: Vec<i64> = received.into_iter().map(|rec| Inst::from(rec).id).collect();
assert_eq!(received_ids, vec![0, 1, 2, 3, 4, 5, 6, 7]);
// Test range (3, inf)
let received = db.range_by(&Field::Id, &Value::Int(3)..).unwrap();
- let received_ids: Vec<i64> = received.iter().map(|inst| inst.id).collect();
+ let received_ids: Vec<i64> = received.into_iter().map(|rec| Inst::from(rec).id).collect();
assert_eq!(received_ids, vec![3, 4, 5, 6, 7, 8, 9]);
}
@@ -608,8 +593,6 @@ fn test_batch_find_by() {
.fields(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");
@@ -634,7 +617,10 @@ fn test_batch_find_by() {
vec![0, 1, 2]
);
assert_eq!(
- result.iter().map(|(_, inst)| inst.id).collect::<Vec<i64>>(),
+ result
+ .into_iter()
+ .map(|(_, rec)| Inst::from(rec).id)
+ .collect::<Vec<i64>>(),
vec![2, 3, 4]
);
}
@@ -646,8 +632,6 @@ fn test_commit_transaction() {
.fields(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");
@@ -684,8 +668,6 @@ fn test_rollback_transaction() {
.fields(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");
@@ -747,42 +729,27 @@ struct InstWithNewNullableField {
pub maybe_str: Option<String>,
}
-impl InstWithNewNullableField {
- fn schema() -> Vec<FieldWithNewNullableField> {
+impl From<InstWithNewNullableField> for Record {
+ fn from(inst: InstWithNewNullableField) -> Record {
vec![
- FieldWithNewNullableField::Id,
- FieldWithNewNullableField::Name,
- FieldWithNewNullableField::Data,
- FieldWithNewNullableField::MaybeStr,
- ]
- }
-
- fn primary_key() -> FieldWithNewNullableField {
- FieldWithNewNullableField::Id
- }
-
- fn secondary_keys() -> Vec<FieldWithNewNullableField> {
- vec![FieldWithNewNullableField::Name]
- }
-
- fn into_record(self) -> Vec<Value> {
- vec![
- Value::Int(self.id),
- match self.name {
+ Value::Int(inst.id),
+ match inst.name {
Some(name) => Value::String(name),
None => Value::Null,
},
- Value::Bytes(self.data),
- match self.maybe_str {
+ Value::Bytes(inst.data),
+ match inst.maybe_str {
Some(maybe_str) => Value::String(maybe_str),
None => Value::Null,
},
]
+ .into()
}
+}
- fn from_record(record: Vec<Value>) -> Self {
+impl From<Record> for InstWithNewNullableField {
+ fn from(record: Record) -> Self {
let mut it = record.into_iter();
-
InstWithNewNullableField {
id: match it.next().unwrap() {
Value::Int(id) => id,
@@ -807,6 +774,25 @@ impl InstWithNewNullableField {
}
}
+impl InstWithNewNullableField {
+ fn schema() -> Vec<FieldWithNewNullableField> {
+ vec![
+ FieldWithNewNullableField::Id,
+ FieldWithNewNullableField::Name,
+ FieldWithNewNullableField::Data,
+ FieldWithNewNullableField::MaybeStr,
+ ]
+ }
+
+ fn primary_key() -> FieldWithNewNullableField {
+ FieldWithNewNullableField::Id
+ }
+
+ fn secondary_keys() -> Vec<FieldWithNewNullableField> {
+ vec![FieldWithNewNullableField::Name]
+ }
+}
+
#[test]
fn test_add_nullable_field() {
let data_dir = tmp_dir();
@@ -817,8 +803,6 @@ fn test_add_nullable_field() {
.fields(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");
@@ -836,8 +820,6 @@ fn test_add_nullable_field() {
.fields(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");
@@ -868,8 +850,6 @@ fn test_delete_by_multiple_indexes() {
.fields(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");
@@ -907,8 +887,6 @@ fn test_find_by_with_offset_and_limit() {
.fields(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");
@@ -933,7 +911,10 @@ fn test_find_by_with_offset_and_limit() {
limit: 3,
},
)
- .unwrap();
+ .unwrap()
+ .into_iter()
+ .map(|rec| Inst::from(rec))
+ .collect::<Vec<Inst>>();
assert_eq!(result.len(), 3);
assert_eq!(result[0].id, 2);
@@ -949,8 +930,6 @@ fn test_batch_find_by_with_offset_and_limit() {
.fields(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");
@@ -976,11 +955,14 @@ fn test_batch_find_by_with_offset_and_limit() {
limit: 2,
},
)
- .unwrap();
+ .unwrap()
+ .into_iter()
+ .map(|(_, rec)| Inst::from(rec))
+ .collect::<Vec<Inst>>();
assert_eq!(result.len(), 2);
- assert_eq!(result[0].1.id, 3);
- assert_eq!(result[1].1.id, 4);
+ assert_eq!(result[0].id, 3);
+ assert_eq!(result[1].id, 4);
}
#[test]
@@ -991,8 +973,6 @@ fn test_range_by_with_offset_and_limit() {
.fields(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");
@@ -1017,7 +997,10 @@ fn test_range_by_with_offset_and_limit() {
limit: 3,
},
)
- .unwrap();
+ .unwrap()
+ .into_iter()
+ .map(|rec| Inst::from(rec))
+ .collect::<Vec<Inst>>();
assert_eq!(result.len(), 3);
assert_eq!(result[0].id, 3);

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