aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--Cargo.lock3
-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
-rw-r--r--py_bindings/Cargo.toml1
-rw-r--r--py_bindings/example.py69
-rw-r--r--py_bindings/log_db.pyi76
-rw-r--r--py_bindings/pyproject.toml2
-rw-r--r--py_bindings/src/lib.rs630
13 files changed, 996 insertions, 376 deletions
diff --git a/Cargo.lock b/Cargo.lock
index 00fdf6f..6a3c89c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1,6 +1,6 @@
# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
-version = 3
+version = 4
[[package]]
name = "ahash"
@@ -784,6 +784,7 @@ version = "0.1.0"
dependencies = [
"log_db",
"pyo3",
+ "rust_decimal",
]
[[package]]
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);
+}
diff --git a/py_bindings/Cargo.toml b/py_bindings/Cargo.toml
index 2f7a83f..a84ccd6 100644
--- a/py_bindings/Cargo.toml
+++ b/py_bindings/Cargo.toml
@@ -9,3 +9,4 @@ crate-type = ["lib"]
[dependencies]
log_db = { path = "../log_db" }
pyo3 = "0.22.3"
+rust_decimal = { version = "1.36.0", features = [] }
diff --git a/py_bindings/example.py b/py_bindings/example.py
new file mode 100644
index 0000000..f175b8a
--- /dev/null
+++ b/py_bindings/example.py
@@ -0,0 +1,69 @@
+from pprint import pformat, pprint
+import log_db
+from log_db import Value, Type, Bound
+
+# TODO: This should be imported from the log_db module
+# but exporting type aliases does not work automatically
+Record = list[Value]
+
+class Inst:
+ def __init__(self, id: int, name: str):
+ self.id = id
+ self.name = name
+
+ def into_record(self) -> Record:
+ return [
+ Value.int(self.id),
+ Value.string(self.name),
+ ]
+
+ @staticmethod
+ def from_record(rec: Record):
+ return Inst(
+ rec[0].as_int(),
+ rec[1].as_string(),
+ )
+
+ def __repr__(self) -> str:
+ return pformat(self.__dict__)
+
+# Create a new database
+db = log_db.DB \
+ .configure() \
+ .data_dir("db") \
+ .schema([
+ ("id", Type.int()),
+ ("name", Type.string()),
+ ]) \
+ .primary_key("id") \
+ .secondary_keys(["name"]) \
+ .initialize()
+
+#db.upsert(Inst(1, "foo").into_record())
+
+#res = db.find_by("name", log_db.Value.string("foo"))
+#insts = [Inst.from_record(r) for r in res]
+res = db.range_by("name",
+ Bound.unbounded(),
+ Bound.unbounded(),
+)
+# insts = [Inst.from_record(r) for r in res]
+
+print("before delete:")
+res = db.find_by("name", Value.string("foo"))
+print(len(res))
+
+res = db.delete_by("name", Value.string("foo"))
+#res = db.delete_by("id", Value.int(1))
+
+print("deleted:")
+for r in res:
+ pprint(Inst.from_record(r))
+
+print("find name after delete:")
+res = db.find_by("name", Value.string("foo"))
+print(len(res))
+
+print("find id after delete:")
+res = db.find_by("id", Value.int(1))
+print(len(res))
diff --git a/py_bindings/log_db.pyi b/py_bindings/log_db.pyi
new file mode 100644
index 0000000..49c10f2
--- /dev/null
+++ b/py_bindings/log_db.pyi
@@ -0,0 +1,76 @@
+WRITE_DURABILITY_FLUSH: int
+WRITE_DURABILITY_FLUSH_SYNC: int
+READ_CONSISTENCY_EVENTUAL: int
+READ_CONSISTENCY_STRONG: int
+
+VALUE_INT: int
+VALUE_DECIMAL: int
+VALUE_STRING: int
+VALUE_BYTES: int
+VALUE_NULL: int
+
+Record = list["Value"]
+
+class Config:
+ def data_dir(self, data_dir: str) -> "Config": ...
+ def segment_size(self, segment_size: int) -> "Config": ...
+ def write_durability(self, write_durability: int) -> "Config": ...
+ def read_consistency(self, read_consistency: int) -> "Config": ...
+ def schema(self, schema: list[tuple[str, "Type"]]) -> "Config": ...
+ def primary_key(self, primary_key: str) -> "Config": ...
+ def secondary_keys(self, secondary_keys: list[str]) -> "Config": ...
+ def initialize(self) -> "DB": ...
+
+class DB:
+ @staticmethod
+ def configure() -> Config: ...
+ def upsert(self, record: Record) -> None: ...
+ def get(self, key: str) -> Record: ...
+ def find_by(self, key: str, value: "Value") -> list[Record]: ...
+ def batch_find_by(self, key: str, values: list["Value"]) -> list[tuple[int, Record]]: ...
+ def delete(self, key: str) -> list[Record]: ...
+ def delete_by(self, key: str, value: "Value") -> list[Record]: ...
+ def range_by(self, key: str, start: "Bound", end: "Bound") -> list[Record]: ...
+ def tx_begin(self) -> None: ...
+ def tx_commit(self) -> None: ...
+ def tx_rollback(self) -> None: ...
+
+class Value:
+ @staticmethod
+ def int(v: int) -> "Value": ...
+ @staticmethod
+ def decimal(v: str) -> "Value": ...
+ @staticmethod
+ def string(v: str) -> "Value": ...
+ @staticmethod
+ def bytes(v: bytes) -> "Value": ...
+ @staticmethod
+ def null() -> "Value": ...
+
+ def kind(self) -> int: ...
+
+ def as_int(self) -> int: ...
+ def as_decimal(self) -> str: ...
+ def as_string(self) -> str: ...
+ def as_bytes(self) -> bytes: ...
+ def as_null(self) -> None: ...
+
+class Type:
+ @staticmethod
+ def int() -> "Type": ...
+ @staticmethod
+ def decimal() -> "Type": ...
+ @staticmethod
+ def string() -> "Type": ...
+ @staticmethod
+ def bytes() -> "Type": ...
+
+ def nullable(self) -> "Type": ...
+
+class Bound:
+ @staticmethod
+ def unbounded() -> "Bound": ...
+ @staticmethod
+ def included(v: "Value") -> "Bound": ...
+ @staticmethod
+ def excluded(v: "Value") -> "Bound": ...
diff --git a/py_bindings/pyproject.toml b/py_bindings/pyproject.toml
index 01bdbce..6f5b437 100644
--- a/py_bindings/pyproject.toml
+++ b/py_bindings/pyproject.toml
@@ -3,7 +3,7 @@ requires = ["maturin>=1.7,<2.0"]
build-backend = "maturin"
[project]
-name = "log_db_py"
+name = "log_db"
requires-python = ">=3.8"
classifiers = [
"Programming Language :: Rust",
diff --git a/py_bindings/src/lib.rs b/py_bindings/src/lib.rs
index 498aaaf..7749839 100644
--- a/py_bindings/src/lib.rs
+++ b/py_bindings/src/lib.rs
@@ -1,212 +1,468 @@
-// use log_db;
-// use pyo3::exceptions::PyException;
-// use pyo3::prelude::*;
+use std::str::FromStr;
-// type Field = String;
+use log_db::{self, OwnedBounds};
+use pyo3::exceptions::PyException;
+use pyo3::prelude::*;
+use rust_decimal::Decimal;
+use std::ops::Bound as StdBound;
-// #[pyclass]
-// #[derive(Clone)]
-// struct ValueType {
-// record_field: log_db::ValueType,
-// }
+type PyRecord = Vec<Value>;
+type PyField = String;
-// #[pymethods]
-// impl ValueType {
-// #[staticmethod]
-// fn int() -> Self {
-// ValueType {
-// record_field: log_db::ValueType::int(),
-// }
-// }
+#[pyclass]
+#[derive(Clone)]
+struct Type {
+ typ: log_db::Type,
+}
-// #[staticmethod]
-// fn float() -> Self {
-// ValueType {
-// record_field: log_db::ValueType::float(),
-// }
-// }
+#[pymethods]
+impl Type {
+ #[staticmethod]
+ fn int() -> Self {
+ Type {
+ typ: log_db::Type::int(),
+ }
+ }
-// #[staticmethod]
-// fn string() -> Self {
-// ValueType {
-// record_field: log_db::ValueType::string(),
-// }
-// }
+ #[staticmethod]
+ fn decimal() -> Self {
+ Type {
+ typ: log_db::Type::decimal(),
+ }
+ }
-// #[staticmethod]
-// fn bytes() -> Self {
-// ValueType {
-// record_field: log_db::ValueType::bytes(),
-// }
-// }
+ #[staticmethod]
+ fn string() -> Self {
+ Type {
+ typ: log_db::Type::string(),
+ }
+ }
-// fn nullable(&self) -> Self {
-// ValueType {
-// record_field: self.record_field.clone().nullable(),
-// }
-// }
-// }
+ #[staticmethod]
+ fn bytes() -> Self {
+ Type {
+ typ: log_db::Type::bytes(),
+ }
+ }
-// #[pyclass]
-// #[derive(Clone)]
-// struct WriteDurability {
-// write_durability: log_db::WriteDurability,
-// }
+ fn nullable(&self) -> Self {
+ Type {
+ typ: self.typ.clone().nullable(),
+ }
+ }
+}
-// #[pyclass]
-// struct Config {
-// #[pyo3(get, set)]
-// data_dir: Option<String>,
-// #[pyo3(get, set)]
-// segment_size: Option<usize>,
-// #[pyo3(get, set)]
-// fields: Option<Vec<(Field, ValueType)>>,
-// #[pyo3(get, set)]
-// primary_key: Option<Field>,
-// #[pyo3(get, set)]
-// secondary_keys: Option<Vec<Field>>,
-// #[pyo3(get, set)]
-// write_durability: Option<WriteDurability>,
-// }
+pub const WRITE_DURABILITY_FLUSH: u8 = 0;
+pub const WRITE_DURABILITY_FLUSH_SYNC: u8 = 1;
-// #[pymethods]
-// impl Config {
-// pub fn initialize(&self) -> PyResult<DB> {
-// let mut config = log_db::DB::configure();
-// if self.data_dir.is_some() {
-// config.data_dir(&self.data_dir.as_ref().unwrap().to_string());
-// }
-// if self.segment_size.is_some() {
-// config.segment_size(self.segment_size.unwrap());
-// }
-// if self.fields.is_some() {
-// let mut fields = Vec::new();
-// for (field, record_field) in self.fields.as_ref().unwrap() {
-// fields.push((field.to_string(), record_field.record_field.clone()));
-// }
-// config.fields(&fields);
-// }
-// if self.primary_key.is_some() {
-// config.primary_key(self.primary_key.as_ref().unwrap().to_string());
-// }
-// if self.secondary_keys.is_some() {
-// let tmp = self.secondary_keys.as_ref().unwrap();
-// config.secondary_keys(&tmp);
-// }
-// if self.write_durability.is_some() {
-// let tmp = self.write_durability.as_ref().unwrap();
-// config.write_durability(tmp.write_durability.clone());
-// }
+pub const READ_CONSISTENCY_EVENTUAL: u8 = 0;
+pub const READ_CONSISTENCY_STRONG: u8 = 1;
-// let db = config
-// .initialize()
-// .map_err(|e| PyException::new_err(e.to_string()))?;
-// Ok(DB { db })
-// }
-// }
+#[pyclass]
+struct Config {
+ data_dir: Option<PyField>,
+ segment_size: Option<usize>,
+ write_durability: Option<log_db::WriteDurability>,
+ read_consistency: Option<log_db::ReadConsistency>,
+ schema: Option<Vec<(PyField, Type)>>,
+ primary_key: Option<PyField>,
+ secondary_keys: Option<Vec<PyField>>,
+}
-// #[pyclass]
-// #[derive(Clone)]
-// struct Value {
-// record_value: log_db::Value,
-// }
+#[pymethods]
+impl Config {
+ pub fn data_dir<'a>(
+ mut slf: PyRefMut<'a, Self>,
+ data_dir: &str,
+ ) -> PyResult<PyRefMut<'a, Self>> {
+ slf.data_dir = Some(data_dir.into());
+ Ok(slf)
+ }
-// #[pymethods]
-// impl Value {
-// #[staticmethod]
-// fn int(value: i64) -> Self {
-// Value {
-// record_value: log_db::Value::Int(value),
-// }
-// }
+ pub fn segment_size<'a>(
+ mut slf: PyRefMut<'a, Self>,
+ segment_size: usize,
+ ) -> PyResult<PyRefMut<'a, Self>> {
+ slf.segment_size = Some(segment_size);
+ Ok(slf)
+ }
-// #[staticmethod]
-// fn float(value: f64) -> Self {
-// Value {
-// record_value: log_db::Value::Float(value),
-// }
-// }
+ pub fn write_durability<'a>(
+ mut slf: PyRefMut<'a, Self>,
+ write_durability: u8,
+ ) -> PyResult<PyRefMut<'a, Self>> {
+ slf.write_durability = Some(match write_durability {
+ WRITE_DURABILITY_FLUSH => log_db::WriteDurability::Flush,
+ WRITE_DURABILITY_FLUSH_SYNC => log_db::WriteDurability::FlushSync,
+ _ => {
+ return Err(PyException::new_err(format!(
+ "Invalid write_durability value: {}",
+ write_durability,
+ )))
+ }
+ });
+ Ok(slf)
+ }
-// #[staticmethod]
-// fn string(value: &str) -> Self {
-// Value {
-// record_value: log_db::Value::String(value.to_string()),
-// }
-// }
+ pub fn read_consistency<'a>(
+ mut slf: PyRefMut<'a, Self>,
+ read_consistency: u8,
+ ) -> PyResult<PyRefMut<'a, Self>> {
+ slf.read_consistency = Some(match read_consistency {
+ READ_CONSISTENCY_EVENTUAL => log_db::ReadConsistency::Eventual,
+ READ_CONSISTENCY_STRONG => log_db::ReadConsistency::Strong,
+ _ => {
+ return Err(PyException::new_err(format!(
+ "Invalid read_consistency value: {}",
+ read_consistency,
+ )))
+ }
+ });
+ Ok(slf)
+ }
-// #[staticmethod]
-// fn bytes(value: &[u8]) -> Self {
-// Value {
-// record_value: log_db::Value::Bytes(value.to_vec()),
-// }
-// }
+ pub fn schema<'a>(
+ mut slf: PyRefMut<'a, Self>,
+ schema: Vec<(PyField, Type)>,
+ ) -> PyResult<PyRefMut<'a, Self>> {
+ slf.schema = Some(schema);
+ Ok(slf)
+ }
-// #[staticmethod]
-// fn null() -> Self {
-// Value {
-// record_value: log_db::Value::Null,
-// }
-// }
-// }
+ pub fn primary_key<'a>(
+ mut slf: PyRefMut<'a, Self>,
+ primary_key: PyField,
+ ) -> PyResult<PyRefMut<'a, Self>> {
+ slf.primary_key = Some(primary_key);
+ Ok(slf)
+ }
-// #[pyclass]
-// struct Record {
-// values: Vec<Value>,
-// }
+ pub fn secondary_keys<'a>(
+ mut slf: PyRefMut<'a, Self>,
+ secondary_keys: Vec<PyField>,
+ ) -> PyResult<PyRefMut<'a, Self>> {
+ slf.secondary_keys = Some(secondary_keys);
+ Ok(slf)
+ }
-// #[pymethods]
-// impl Record {
-// #[new]
-// #[pyo3(signature = (*py_args))]
-// fn new(py_args: Vec<Value>) -> Self {
-// Record { values: py_args }
-// }
-// }
+ pub fn initialize(&self) -> PyResult<DB> {
+ let mut config = log_db::DB::configure();
+ if self.data_dir.is_some() {
+ config = config.data_dir(&self.data_dir.as_ref().unwrap().to_string());
+ }
+ if self.segment_size.is_some() {
+ config = config.segment_size(self.segment_size.unwrap());
+ }
+ if self.write_durability.is_some() {
+ let tmp = self.write_durability.as_ref().unwrap();
+ config = config.write_durability(tmp.clone());
+ }
+ if self.read_consistency.is_some() {
+ let tmp = self.read_consistency.as_ref().unwrap();
+ config = config.read_consistency(tmp.clone());
+ }
+ if self.schema.is_some() {
+ let schema = self
+ .schema
+ .as_ref()
+ .unwrap()
+ .iter()
+ .map(|(name, typ)| (name.clone(), typ.typ.clone()))
+ .collect();
+ config = config.schema(schema);
+ }
+ if self.primary_key.is_some() {
+ config = config.primary_key(self.primary_key.as_ref().unwrap().to_string());
+ }
+ if self.secondary_keys.is_some() {
+ config = config.secondary_keys(self.secondary_keys.as_ref().unwrap().clone());
+ }
-// #[pyclass]
-// struct DB {
-// db: log_db::DB<Field>,
-// }
+ let db = config
+ .from_record(py_from_record)
+ .into_record(py_into_record)
+ .initialize()
+ .map_err(|e| PyException::new_err(e.to_string()))?;
+ Ok(DB { db })
+ }
+}
-// #[pymethods]
-// impl DB {
-// fn upsert(&mut self, record: &Record) -> PyResult<()> {
-// let values: Vec<log_db::Value> = record
-// .values
-// .iter()
-// .map(|v| v.record_value.clone())
-// .collect();
+fn py_from_record(record: Vec<log_db::Value>) -> Vec<Value> {
+ record
+ .into_iter()
+ .map(|value| Value {
+ record_value: value,
+ })
+ .collect()
+}
-// self.db
-// .upsert(&log_db::Record::from(&values))
-// .map_err(|e| PyException::new_err(e.to_string()))?;
-// Ok(())
-// }
+fn py_into_record(record: Vec<Value>) -> Vec<log_db::Value> {
+ record.into_iter().map(|value| value.record_value).collect()
+}
-// #[staticmethod]
-// pub fn configure() -> Config {
-// Config {
-// data_dir: None,
-// segment_size: None,
-// fields: None,
-// primary_key: None,
-// secondary_keys: None,
-// write_durability: None,
-// }
-// }
-// }
+const VALUE_INT: u8 = 0;
+const VALUE_DECIMAL: u8 = 1;
+const VALUE_STRING: u8 = 2;
+const VALUE_BYTES: u8 = 3;
+const VALUE_NULL: u8 = 4;
-// // #[pyfunction]
-// // fn sum_as_string(a: usize, b: usize) -> PyResult<String> {
-// // Ok((a + b).to_string())
-// // }
+#[pyclass]
+#[derive(Clone, PartialEq, Eq)]
+pub struct Value {
+ record_value: log_db::Value,
+}
-// #[pymodule]
-// fn log_db_py(m: &Bound<'_, PyModule>) -> PyResult<()> {
-// //m.add_function(wrap_pyfunction!(sum_as_string, m)?)?;
-// m.add_class::<DB>()?;
-// m.add_class::<ValueType>()?;
-// m.add_class::<Value>()?;
-// m.add_class::<Record>()?;
-// Ok(())
-// }
+#[pymethods]
+impl Value {
+ fn __repr__(&self) -> String {
+ match &self.record_value {
+ log_db::Value::Int(value) => format!("Value.int({})", value),
+ log_db::Value::Decimal(value) => format!("Value.decimal({})", value),
+ log_db::Value::String(value) => {
+ format!("Value.string(\"{}\")", value.replace("\"", "\\\""))
+ }
+ log_db::Value::Bytes(value) => format!("Value.bytes({:?})", value),
+ log_db::Value::Null => "Value.null()".to_string(),
+ }
+ }
+
+ #[staticmethod]
+ fn int(value: i64) -> Self {
+ Value {
+ record_value: log_db::Value::Int(value),
+ }
+ }
+
+ #[staticmethod]
+ fn decimal(value: String) -> Self {
+ Value {
+ record_value: log_db::Value::Decimal(
+ Decimal::from_str(&value).expect(&format!("Invalid Decimal: {}", value)),
+ ),
+ }
+ }
+
+ #[staticmethod]
+ fn string(value: String) -> Self {
+ Value {
+ record_value: log_db::Value::String(value),
+ }
+ }
+
+ #[staticmethod]
+ fn bytes(value: &[u8]) -> Self {
+ Value {
+ record_value: log_db::Value::Bytes(value.to_vec()),
+ }
+ }
+
+ #[staticmethod]
+ fn null() -> Self {
+ Value {
+ record_value: log_db::Value::Null,
+ }
+ }
+
+ pub fn kind(&self) -> u8 {
+ match &self.record_value {
+ log_db::Value::Int(_) => VALUE_INT,
+ log_db::Value::Decimal(_) => VALUE_DECIMAL,
+ log_db::Value::String(_) => VALUE_STRING,
+ log_db::Value::Bytes(_) => VALUE_BYTES,
+ log_db::Value::Null => VALUE_NULL,
+ }
+ }
+
+ pub fn as_int(&self) -> PyResult<i64> {
+ match &self.record_value {
+ log_db::Value::Int(value) => Ok(*value),
+ _ => Err(PyException::new_err("Value is not an Int")),
+ }
+ }
+
+ pub fn as_decimal(&self) -> PyResult<String> {
+ match &self.record_value {
+ log_db::Value::Decimal(value) => Ok(value.to_string()),
+ _ => Err(PyException::new_err("Value is not a Decimal")),
+ }
+ }
+
+ pub fn as_string(&self) -> PyResult<String> {
+ match &self.record_value {
+ log_db::Value::String(value) => Ok(value.clone()),
+ _ => Err(PyException::new_err("Value is not a String")),
+ }
+ }
+
+ pub fn as_bytes(&self) -> PyResult<Vec<u8>> {
+ match &self.record_value {
+ log_db::Value::Bytes(value) => Ok(value.clone()),
+ _ => Err(PyException::new_err("Value is not Bytes")),
+ }
+ }
+
+ pub fn as_null(&self) -> PyResult<()> {
+ match &self.record_value {
+ log_db::Value::Null => Ok(()),
+ _ => Err(PyException::new_err("Value is not Null")),
+ }
+ }
+}
+
+#[pyclass]
+struct DB {
+ db: log_db::DB<PyRecord, String>,
+}
+
+#[pymethods]
+impl DB {
+ #[staticmethod]
+ pub fn configure() -> Config {
+ Config {
+ data_dir: None,
+ segment_size: None,
+ write_durability: None,
+ read_consistency: None,
+ schema: None,
+ primary_key: None,
+ secondary_keys: None,
+ }
+ }
+
+ pub fn upsert(&mut self, record: PyRecord) -> PyResult<()> {
+ self.db
+ .upsert(record)
+ .map_err(|e| PyException::new_err(e.to_string()))?;
+ Ok(())
+ }
+
+ pub fn get(&mut self, key: Value) -> PyResult<Option<PyRecord>> {
+ self.db
+ .get(&key.record_value)
+ .map_err(|e| PyException::new_err(e.to_string()))
+ }
+
+ // TODO: refactor out &String
+ pub fn find_by(&mut self, field: PyField, key: &Value) -> PyResult<Vec<PyRecord>> {
+ self.db
+ .find_by(&field, &key.record_value)
+ .map_err(|e| PyException::new_err(e.to_string()))
+ }
+
+ // batch_find_by
+ pub fn batch_find_by(
+ &mut self,
+ field: PyField,
+ keys: Vec<Value>,
+ ) -> PyResult<Vec<(usize, PyRecord)>> {
+ let keys: Vec<log_db::Value> = keys.into_iter().map(|key| key.record_value).collect();
+ self.db
+ .batch_find_by(&field, &keys)
+ .map_err(|e| PyException::new_err(e.to_string()))
+ }
+
+ pub fn range_by(
+ &mut self,
+ field: PyField,
+ start: &PyRangeBound,
+ end: &PyRangeBound,
+ ) -> PyResult<Vec<PyRecord>> {
+ let range = OwnedBounds::new(
+ match start {
+ PyRangeBound::Unbounded() => StdBound::Unbounded,
+ PyRangeBound::Included(value) => StdBound::Included(value.record_value.clone()),
+ PyRangeBound::Excluded(value) => StdBound::Excluded(value.record_value.clone()),
+ },
+ match end {
+ PyRangeBound::Unbounded() => StdBound::Unbounded,
+ PyRangeBound::Included(value) => StdBound::Included(value.record_value.clone()),
+ PyRangeBound::Excluded(value) => StdBound::Excluded(value.record_value.clone()),
+ },
+ );
+
+ self.db
+ .range_by(&field, range)
+ .map_err(|e| PyException::new_err(e.to_string()))
+ }
+
+ pub fn delete(&mut self, key: &Value) -> PyResult<Option<PyRecord>> {
+ self.db
+ .delete(&key.record_value)
+ .map_err(|e| PyException::new_err(e.to_string()))
+ }
+
+ pub fn delete_by(&mut self, field: PyField, key: &Value) -> PyResult<Vec<PyRecord>> {
+ self.db
+ .delete_by(&field, &key.record_value)
+ .map_err(|e| PyException::new_err(e.to_string()))
+ }
+
+ pub fn tx_begin(&mut self) -> PyResult<()> {
+ self.db
+ .tx_begin()
+ .map_err(|e| PyException::new_err(e.to_string()))?;
+ Ok(())
+ }
+
+ pub fn tx_commit(&mut self) -> PyResult<()> {
+ self.db
+ .tx_commit()
+ .map_err(|e| PyException::new_err(e.to_string()))?;
+ Ok(())
+ }
+
+ pub fn tx_rollback(&mut self) -> PyResult<()> {
+ self.db
+ .tx_rollback()
+ .map_err(|e| PyException::new_err(e.to_string()))?;
+ Ok(())
+ }
+}
+
+#[pyclass(name = "Bound", eq)]
+#[derive(Clone, PartialEq, Eq)]
+pub enum PyRangeBound {
+ Unbounded(),
+ Included(Value),
+ Excluded(Value),
+}
+
+#[pymethods]
+impl PyRangeBound {
+ #[staticmethod]
+ pub fn unbounded() -> Self {
+ PyRangeBound::Unbounded()
+ }
+
+ #[staticmethod]
+ pub fn included(value: Value) -> Self {
+ PyRangeBound::Included(value)
+ }
+
+ #[staticmethod]
+ pub fn excluded(value: Value) -> Self {
+ PyRangeBound::Excluded(value)
+ }
+}
+
+#[pymodule(name = "log_db")]
+fn log_db_py(m: &Bound<'_, PyModule>) -> PyResult<()> {
+ m.add_class::<DB>()?;
+ m.add_class::<Type>()?;
+ m.add_class::<Value>()?;
+ m.add_class::<PyRangeBound>()?;
+
+ m.add("WRITE_DURABILITY_FLUSH", WRITE_DURABILITY_FLUSH)?;
+ m.add("WRITE_DURABILITY_FLUSH_SYNC", WRITE_DURABILITY_FLUSH_SYNC)?;
+
+ m.add("READ_CONSISTENCY_EVENTUAL", READ_CONSISTENCY_EVENTUAL)?;
+ m.add("READ_CONSISTENCY_STRONG", READ_CONSISTENCY_STRONG)?;
+
+ m.add("VALUE_INT", VALUE_INT)?;
+ m.add("VALUE_DECIMAL", VALUE_DECIMAL)?;
+ m.add("VALUE_STRING", VALUE_STRING)?;
+ m.add("VALUE_BYTES", VALUE_BYTES)?;
+ m.add("VALUE_NULL", VALUE_NULL)?;
+
+ Ok(())
+}

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