aboutsummaryrefslogtreecommitdiffstats
path: root/log_db/src/engine.rs
diff options
context:
space:
mode:
authorJan Tuomi <jan@jantuomi.fi>2025-02-05 17:50:02 +0200
committerJan Tuomi <jan@jantuomi.fi>2025-02-09 23:52:26 +0200
commitd422ffe3d48d2061b4d3ddeb0c08796e4d71a5fa (patch)
tree0b6820b060bf0cbeb2686a53a23c9ebee5654334 /log_db/src/engine.rs
parentd4c953dc4bbd190dae4b5be69941f5c42c59af34 (diff)
Remove Recordable trait, implement Python bindings, fixes
Diffstat (limited to 'log_db/src/engine.rs')
-rw-r--r--log_db/src/engine.rs44
1 files changed, 21 insertions, 23 deletions
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 {