3 files changed, 139 insertions, 39 deletions

diff --git a/log_db/src/common.rs b/log_db/src/common.rs
index 4be2b39..a4f253c 100644
--- a/log_db/src/common.rs
+++ b/log_db/src/common.rs
@@ -26,6 +26,16 @@ pub const INIT_LOCK_FILENAME: &str = "init_lock";
 pub const DEFAULT_READ_BUF_SIZE: usize = 1024 * 1024; // 1 MB
 pub const TEST_RESOURCES_DIR: &str = "tests/resources";
 
+// Serialized value tags
+pub const B_NULL: u8 = 0x0;
+pub const B_INT: u8 = 0x1;
+pub const B_FLOAT: u8 = 0x2;
+pub const B_STRING: u8 = 0x3;
+pub const B_BYTES: u8 = 0x4;
+// Tombstone marker tags
+pub const B_LIVE: u8 = 0x0;
+pub const B_TOMBSTONE: u8 = 0xFF;
+
 pub fn metadata_filename(num: u16) -> String {
     format!("metadata.{}", num)
 }
@@ -42,6 +52,14 @@ pub enum DBError {
     IOError(#[from] io::Error),
 }
 
+#[derive(Debug, Error)]
+pub enum LogKeySetError {
+    #[error("log key not found in set")]
+    NotFoundError,
+    #[error("attempted to remove last element of non-empty set")]
+    RemovingLastElementError,
+}
+
 /// LogKey is a packed struct that contains:
 /// - a log segment number (16 bits)
 /// - a log index within the segment (48 bits)
@@ -102,6 +120,10 @@ impl LogKeySet {
         self.set.iter()
     }
 
+    pub fn contains(&self, key: &LogKey) -> bool {
+        self.set.contains(key)
+    }
+
     /// The number of LogKeys in the set.
     pub fn len(&self) -> usize {
         self.set.len()
@@ -113,22 +135,16 @@ impl LogKeySet {
     }
 
     /// Remove a LogKey from the set. Return Ok(()) if the key was found and removed.
-    /// Return io::Error::InvalidInput if trying to remove the last element.
-    /// Return io::Error::NotFound if the key was not found.
-    pub fn remove(&mut self, key: &LogKey) -> Result<(), io::Error> {
+    /// Return `LogKeySetError::RemovingLastElementError` if trying to remove the last element.
+    /// Return `LogKeySetError::NotFoundError` if the key was not found.
+    pub fn remove(&mut self, key: &LogKey) -> Result<(), LogKeySetError> {
         if self.set.len() == 1 {
-            return Err(io::Error::new(
-                io::ErrorKind::InvalidInput,
-                "Cannot remove the last element from LogKeySet",
-            ));
+            return Err(LogKeySetError::RemovingLastElementError);
         }
         let removed = self.set.remove(key);
 
         if !removed {
-            return Err(io::Error::new(
-                io::ErrorKind::NotFound,
-                "LogKey not found in LogKeySet",
-            ));
+            return Err(LogKeySetError::NotFoundError);
         }
 
         assert!(
@@ -309,27 +325,27 @@ impl Value {
     pub fn serialize(&self) -> Vec<u8> {
         match self {
             Value::Null => {
-                vec![0] // Tag for Null
+                vec![B_NULL]
             }
             Value::Int(i) => {
-                let mut bytes = vec![1]; // Tag for Int
+                let mut bytes = vec![B_INT];
                 bytes.extend(&i.to_be_bytes());
                 bytes
             }
             Value::Float(f) => {
-                let mut bytes = vec![2]; // Tag for Float
+                let mut bytes = vec![B_FLOAT];
                 bytes.extend(&f.to_be_bytes());
                 bytes
             }
             Value::String(s) => {
-                let mut bytes = vec![3]; // Tag for String
+                let mut bytes = vec![B_STRING];
                 let length = s.len() as u64;
                 bytes.extend(&length.to_be_bytes());
                 bytes.extend(s.as_bytes());
                 bytes
             }
             Value::Bytes(b) => {
-                let mut bytes = vec![4]; // Tag for Bytes
+                let mut bytes = vec![B_BYTES];
                 let length = b.len() as u64;
                 bytes.extend(&length.to_be_bytes());
                 bytes.extend(b);
@@ -342,18 +358,18 @@ impl Value {
     /// Returns the deserialized Value and the number of bytes consumed.
     pub fn deserialize(bytes: &[u8]) -> (Value, usize) {
         match bytes[0] {
-            0 => (Value::Null, 1),
-            1 => {
+            B_NULL => (Value::Null, 1),
+            B_INT => {
                 let mut int_bytes = [0; 8];
                 int_bytes.copy_from_slice(&bytes[1..1 + 8]);
                 (Value::Int(i64::from_be_bytes(int_bytes)), 1 + 8)
             }
-            2 => {
+            B_FLOAT => {
                 let mut float_bytes = [0; 8];
                 float_bytes.copy_from_slice(&bytes[1..1 + 8]);
                 (Value::Float(f64::from_be_bytes(float_bytes)), 1 + 8)
             }
-            3 => {
+            B_STRING => {
                 let length_bytes = &bytes[1..1 + 8];
                 let length = u64::from_be_bytes(length_bytes.try_into().unwrap()) as usize;
                 (
@@ -363,7 +379,7 @@ impl Value {
                     1 + 8 + length,
                 )
             }
-            4 => {
+            B_BYTES => {
                 let length_bytes = &bytes[1..1 + 8];
                 let length = u64::from_be_bytes(length_bytes.try_into().unwrap()) as usize;
                 (
@@ -395,9 +411,9 @@ impl Record {
         let mut bytes = Vec::new();
 
         if self.tombstone {
-            bytes.extend(&[0xFF]);
+            bytes.extend(&[B_TOMBSTONE]);
         } else {
-            bytes.extend(&[0]);
+            bytes.extend(&[B_LIVE]);
         }
 
         for value in &self.values {
@@ -409,7 +425,7 @@ impl Record {
     pub fn deserialize(bytes: &[u8]) -> Record {
         let mut values = Vec::new();
 
-        let tombstone = bytes[0] == 0xFF;
+        let tombstone = bytes[0] == B_TOMBSTONE;
 
         let mut start = 1;
         while start < bytes.len() {
diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs
index 194b6d6..3c3f7bb 100644
--- a/log_db/src/lib.rs
+++ b/log_db/src/lib.rs
@@ -355,19 +355,20 @@ 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();
+        if let Some(plk) = self.primary_memtable.remove(&pk) {
+            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
+                    .iter()
+                    .position(|(f, _)| sk_field == f)
+                    .unwrap();
+                let sk = record.at(sk_field_index).as_indexable().unwrap();
 
-            secondary_memtable.remove(&sk);
+                secondary_memtable.remove(&sk, &plk);
+            }
         }
     }
 
@@ -651,9 +652,58 @@ 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!()
+    /// Delete record by primary key.
+    pub fn delete(&mut self, pk: &Value) -> Result<Option<Record>, DBError> {
+        let record = match self.get(pk)? {
+            Some(record) => record,
+            None => return Ok(None),
+        };
+
+        // The Record interface does not allow manually setting the tombstone flag,
+        // so we have to serialize the record and manually set the first byte to B_TOMBSTONE.
+        let mut record_serialized = vec![B_TOMBSTONE];
+        record_serialized.extend(&record.serialize()[1..]);
+
+        request_exclusive_lock(&self.data_dir, &mut self.active_metadata_file)?;
+        self.active_data_file.lock_exclusive()?;
+
+        let offset = self.active_data_file.seek(SeekFrom::End(0))?;
+        let length = record_serialized.len() as u64;
+
+        self.active_data_file.write_all(&record_serialized)?;
+
+        // Flush and sync data to disk
+        if self.config.write_durability == WriteDurability::Flush {
+            self.active_data_file.flush()?;
+        }
+        if self.config.write_durability == WriteDurability::FlushSync {
+            self.active_data_file.flush()?;
+            self.active_data_file.sync_all()?;
+        }
+
+        let mut metadata_entry = vec![];
+        metadata_entry.extend(offset.to_be_bytes().iter());
+        metadata_entry.extend(length.to_be_bytes().iter());
+
+        self.active_metadata_file.write_all(&metadata_entry)?;
+
+        // Flush and sync metadata to disk
+        if self.config.write_durability == WriteDurability::Flush {
+            self.active_metadata_file.flush()?;
+        }
+        if self.config.write_durability == WriteDurability::FlushSync {
+            self.active_metadata_file.flush()?;
+            self.active_metadata_file.sync_all()?;
+        }
+
+        self.remove_record_from_memtables(&record);
+
+        self.active_metadata_file.unlock()?;
+        self.active_data_file.unlock()?;
+
+        debug!("Record deleted, returning from delete");
+
+        Ok(Some(record))
     }
 
     /// Check if there are any pending tasks and do them. Tasks include:
diff --git a/log_db/src/memtable_secondary.rs b/log_db/src/memtable_secondary.rs
index 4479516..a7a9a8a 100644
--- a/log_db/src/memtable_secondary.rs
+++ b/log_db/src/memtable_secondary.rs
@@ -38,7 +38,41 @@ impl SecondaryMemtable {
         }
     }
 
-    pub fn remove(&mut self, key: &IndexableValue) -> Option<LogKeySet> {
+    // Remove all log keys associated with the given key
+    pub fn remove_all(&mut self, key: &IndexableValue) -> Option<LogKeySet> {
         self.records.remove(key)
     }
+
+    // Remove a single log key associated with the given key. Returns `true`
+    // if the log key existed and was removed, `false` otherwise.
+    pub fn remove(&mut self, key: &IndexableValue, log_key: &LogKey) -> bool {
+        let set = match self.records.get_mut(key) {
+            Some(set) => set,
+            None => return false,
+        };
+        if set.len() == 1 && set.contains(log_key) {
+            self.records.remove(key);
+            true
+        } else {
+            return match set.remove(log_key) {
+                Ok(_) => true,
+                Err(LogKeySetError::NotFoundError) => false,
+                Err(e) => panic!("{:?}", e),
+            };
+        }
+    }
+
+    // Remove all log keys associated with the given log key
+    // Note: This is a linear time operation, prefer using the `remove` method
+    // if you know the secondary key associated with the log key.
+    pub fn scan_remove(&mut self, log_key: &LogKey) -> u64 {
+        let mut removed = 0;
+        self.records.iter_mut().for_each(|(_, set)| {
+            if let Ok(_) = set.remove(&log_key) {
+                removed += 1;
+            }
+        });
+
+        removed
+    }
 }