Add tombstone flag to record to support delete

4 files changed, 85 insertions, 29 deletions

diff --git a/log_db/src/common.rs b/log_db/src/common.rs
index bf949c2..f20e501 100644
--- a/log_db/src/common.rs
+++ b/log_db/src/common.rs
@@ -385,12 +385,22 @@ impl Value {
 }
 
 #[derive(Debug, Clone)]
-pub struct Record(Vec<Value>);
+pub struct Record {
+    values: Vec<Value>,
+    tombstone: bool,
+}
 
 impl Record {
     pub fn serialize(&self) -> Vec<u8> {
         let mut bytes = Vec::new();
-        for value in &self.0 {
+
+        if self.tombstone {
+            bytes.extend(&[0xFF]);
+        } else {
+            bytes.extend(&[0]);
+        }
+
+        for value in &self.values {
             bytes.extend(value.serialize());
         }
         bytes
@@ -398,43 +408,50 @@ impl Record {
 
     pub fn deserialize(bytes: &[u8]) -> Record {
         let mut values = Vec::new();
-        let mut start = 0;
+
+        let tombstone = bytes[0] == 0xFF;
+
+        let mut start = 1;
         while start < bytes.len() {
             let (rv, consumed) = Value::deserialize(&bytes[start..]);
             values.push(rv);
             start += consumed;
         }
-        Record(values)
+        Record { values, tombstone }
     }
 
     pub fn from(values: &[Value]) -> Record {
-        Record(values.to_vec())
+        Record {
+            values: values.to_vec(),
+            tombstone: false,
+        }
     }
 
     pub fn values(&self) -> &[Value] {
-        &self.0
+        &self.values
     }
 
     pub fn at(&self, index: usize) -> &Value {
-        &self.0[index]
+        &self.values[index]
+    }
+
+    pub fn is_tombstone(&self) -> bool {
+        self.tombstone
     }
 
-    pub fn validate<Field: Eq>(&self, schema: &Vec<(Field, ValueType)>) -> Result<(), io::Error> {
+    pub fn validate<Field: Eq>(&self, schema: &Vec<(Field, ValueType)>) -> Result<(), DBError> {
         // Validate the record length
-        if self.0.len() != schema.len() {
-            return Err(io::Error::new(
-                io::ErrorKind::InvalidInput,
-                format!(
-                    "Record has an incorrect number of fields: {}, expected {}",
-                    self.0.len(),
-                    schema.len()
-                ),
-            ));
+        if self.values.len() != schema.len() {
+            return Err(DBError::ValidationError(format!(
+                "Record has an incorrect number of fields: {}, expected {}",
+                self.values.len(),
+                schema.len()
+            )));
         }
 
         // Validate that record fields match schema types
         for (i, (_, field)) in schema.iter().enumerate() {
-            match (&self.0[i], field) {
+            match (&self.values[i], field) {
                 (
                     Value::Null,
                     ValueType {
@@ -464,13 +481,10 @@ impl Record {
                     },
                 ) => {}
                 _ => {
-                    return Err(io::Error::new(
-                        io::ErrorKind::InvalidInput,
-                        format!(
-                            "Record field {} has incorrect type: {:?}, expected {:?}",
-                            &i, &self.0[i], &field.prim_value_type
-                        ),
-                    ))
+                    return Err(DBError::ValidationError(format!(
+                        "Record field {} has incorrect type: {:?}, expected {:?}",
+                        &i, &self.values[i], &field.prim_value_type
+                    )));
                 }
             }
         }
diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs
index 400d580..9e5e7c2 100644
--- a/log_db/src/lib.rs
+++ b/log_db/src/lib.rs
@@ -19,7 +19,6 @@ use std::fmt::Debug;
 use std::fs::{self};
 use std::io::{self, Read, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
-use thiserror::Error;
 
 pub struct ConfigBuilder<Field: Eq + Clone + Debug> {
     data_dir: Option<String>,
@@ -312,7 +311,12 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
                 ForwardLogReader::new_with_index(metadata_file, data_file, from_index)
             {
                 let log_key = LogKey::new(segnum, index);
-                self.insert_record_to_memtables(&log_key, &record);
+
+                if record.is_tombstone() {
+                    self.remove_record_from_memtables(&record);
+                } else {
+                    self.insert_record_to_memtables(&log_key, &record);
+                }
 
                 // Update from_index in case this is the last iteration: we need to know the next
                 // index that should be read on later invocations of refresh_indexes.
@@ -349,6 +353,24 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
         }
     }
 
+    fn remove_record_from_memtables(&mut self, record: &Record) {
+        let pk = record.at(self.primary_key_index).as_indexable().unwrap();
+        self.primary_memtable.remove(&pk);
+
+        for (sk_index, sk_field) in self.config.secondary_keys.iter().enumerate() {
+            let secondary_memtable = &mut self.secondary_memtables[sk_index];
+            let sk_field_index = self
+                .config
+                .fields
+                .iter()
+                .position(|(f, _)| sk_field == f)
+                .unwrap();
+            let sk = record.at(sk_field_index).as_indexable().unwrap();
+
+            secondary_memtable.remove(&sk);
+        }
+    }
+
     /// 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, record: &Record) -> Result<(), DBError> {
@@ -627,6 +649,11 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
         }
     }
 
+    /// Delete records by a field value.
+    pub fn delete(&mut self, field: &Field, value: &Value) -> Result<u64, DBError> {
+        todo!()
+    }
+
     /// Check if there are any pending tasks and do them. Tasks include:
     /// - Rotating the active log file if it has reached capacity and compacting it.
     ///
@@ -669,12 +696,19 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
 
         let mut read_n = 0;
         for (original_index, item) in forward_log_reader.enumerate() {
-            let primary_key = item
+            let pk = item
                 .record
                 .at(self.primary_key_index)
                 .as_indexable()
                 .expect("Primary key was not indexable");
-            map.insert(primary_key, (original_index, item.record));
+
+            // If the record is a tombstone, remove the PK from the map
+            if item.record.is_tombstone() {
+                map.remove(&pk);
+            } else {
+                map.insert(pk, (original_index, item.record));
+            }
+
             read_n += 1;
         }
 
diff --git a/log_db/src/memtable_primary.rs b/log_db/src/memtable_primary.rs
index 6a28f30..85648d4 100644
--- a/log_db/src/memtable_primary.rs
+++ b/log_db/src/memtable_primary.rs
@@ -26,4 +26,8 @@ impl PrimaryMemtable {
     pub fn get(&self, key: &IndexableValue) -> Option<&LogKey> {
         self.records.get(key)
     }
+
+    pub fn remove(&mut self, key: &IndexableValue) -> Option<LogKey> {
+        self.records.remove(key)
+    }
 }
diff --git a/log_db/src/memtable_secondary.rs b/log_db/src/memtable_secondary.rs
index 9086a97..4479516 100644
--- a/log_db/src/memtable_secondary.rs
+++ b/log_db/src/memtable_secondary.rs
@@ -37,4 +37,8 @@ impl SecondaryMemtable {
             None => &EMPTY_SET,
         }
     }
+
+    pub fn remove(&mut self, key: &IndexableValue) -> Option<LogKeySet> {
+        self.records.remove(key)
+    }
 }