Remove <T> polymorphism from DB, replace with From<T> based approach

10 files changed, 302 insertions, 307 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);
diff --git a/py_bindings/src/lib.rs b/py_bindings/src/lib.rs
index 6bf3505..a4abfde 100644
--- a/py_bindings/src/lib.rs
+++ b/py_bindings/src/lib.rs
@@ -1,7 +1,7 @@
 use std::str::FromStr;
 use std::usize;
 
-use log_db::{self, OwnedBounds, QueryParams, DEFAULT_QUERY_PARAMS};
+use log_db::{self, OwnedBounds, QueryParams, Record, DEFAULT_QUERY_PARAMS};
 use pyo3::exceptions::PyException;
 use pyo3::prelude::*;
 use rust_decimal::Decimal;
@@ -137,25 +137,28 @@ impl Config {
         }
 
         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 })
     }
 }
 
-fn py_from_record(record: Vec<log_db::Value>) -> Vec<Value> {
+fn py_from_record(record: Record) -> Vec<Value> {
     record
         .into_iter()
         .map(|value| Value {
             record_value: value,
         })
-        .collect()
+        .collect::<Vec<Value>>()
+        .into()
 }
 
-fn py_into_record(record: Vec<Value>) -> Vec<log_db::Value> {
-    record.into_iter().map(|value| value.record_value).collect()
+fn py_into_record(record: Vec<Value>) -> Record {
+    record
+        .into_iter()
+        .map(|value| value.record_value)
+        .collect::<Vec<log_db::Value>>()
+        .into()
 }
 
 const VALUE_INT: u8 = 0;
@@ -269,7 +272,7 @@ impl Value {
 
 #[pyclass]
 struct DB {
-    db: log_db::DB<PyRecord>,
+    db: log_db::DB,
 }
 
 #[pymethods]
@@ -288,6 +291,7 @@ impl DB {
     }
 
     pub fn upsert(&mut self, record: PyRecord) -> PyResult<()> {
+        let record: Record = py_into_record(record);
         self.db
             .upsert(record)
             .map_err(|e| PyException::new_err(e.to_string()))?;
@@ -295,9 +299,12 @@ impl DB {
     }
 
     pub fn get(&mut self, key: Value) -> PyResult<Option<PyRecord>> {
-        self.db
+        let recs = self
+            .db
             .get(&key.record_value)
-            .map_err(|e| PyException::new_err(e.to_string()))
+            .map_err(|e| PyException::new_err(e.to_string()))?;
+
+        Ok(recs.map(|rec| py_from_record(rec)))
     }
 
     #[pyo3(signature = (field, key, offset = DEFAULT_QUERY_PARAMS.offset, limit = DEFAULT_QUERY_PARAMS.limit))]
@@ -309,9 +316,12 @@ impl DB {
         limit: usize,
     ) -> PyResult<Vec<PyRecord>> {
         let params = QueryParams { offset, limit };
-        self.db
+        let recs = self
+            .db
             .find_by_with_params(&field, &key.record_value, &params)
-            .map_err(|e| PyException::new_err(e.to_string()))
+            .map_err(|e| PyException::new_err(e.to_string()))?;
+
+        Ok(recs.into_iter().map(|rec| py_from_record(rec)).collect())
     }
 
     #[pyo3(signature = (field, keys, offset = DEFAULT_QUERY_PARAMS.offset, limit = DEFAULT_QUERY_PARAMS.limit))]
@@ -324,9 +334,15 @@ impl DB {
     ) -> PyResult<Vec<(usize, PyRecord)>> {
         let params = QueryParams { offset, limit };
         let keys: Vec<log_db::Value> = keys.into_iter().map(|key| key.record_value).collect();
-        self.db
+        let recs = self
+            .db
             .batch_find_by_with_params(&field, &keys, &params)
-            .map_err(|e| PyException::new_err(e.to_string()))
+            .map_err(|e| PyException::new_err(e.to_string()))?;
+
+        Ok(recs
+            .into_iter()
+            .map(|(idx, rec)| (idx, py_from_record(rec)))
+            .collect())
     }
 
     #[pyo3(signature = (field, start, end, offset = DEFAULT_QUERY_PARAMS.offset, limit = DEFAULT_QUERY_PARAMS.limit))]
@@ -352,21 +368,30 @@ impl DB {
             },
         );
 
-        self.db
+        let recs = self
+            .db
             .range_by_with_params(&field, range, &params)
-            .map_err(|e| PyException::new_err(e.to_string()))
+            .map_err(|e| PyException::new_err(e.to_string()))?;
+
+        Ok(recs.into_iter().map(|rec| py_from_record(rec)).collect())
     }
 
     pub fn delete(&mut self, key: &Value) -> PyResult<Option<PyRecord>> {
-        self.db
+        let recs = self
+            .db
             .delete(&key.record_value)
-            .map_err(|e| PyException::new_err(e.to_string()))
+            .map_err(|e| PyException::new_err(e.to_string()))?;
+
+        Ok(recs.map(|rec| py_from_record(rec)))
     }
 
     pub fn delete_by(&mut self, field: PyField, key: &Value) -> PyResult<Vec<PyRecord>> {
-        self.db
+        let recs = self
+            .db
             .delete_by(&field, &key.record_value)
-            .map_err(|e| PyException::new_err(e.to_string()))
+            .map_err(|e| PyException::new_err(e.to_string()))?;
+
+        Ok(recs.into_iter().map(|rec| py_from_record(rec)).collect())
     }
 
     pub fn tx_begin(&mut self) -> PyResult<()> {