From 30e91f8c24912a2958acdc0a19c043469ed42c89 Mon Sep 17 00:00:00 2001
From: Rong Ma <160457111+rongma7@users.noreply.github.com>
Date: Tue, 7 Jan 2025 18:34:00 +0800
Subject: [PATCH] Support deletions in `LvmtStore` by using `Option<Box<[u8]>>`
 for `changes` and `LvmtValue.value`

---
 src/lvmt/storage.rs          |  2 +-
 src/lvmt/types/lvmt_value.rs | 28 ++++++++++++++++++++++------
 src/lvmt/types/mod.rs        |  7 +++++--
 3 files changed, 28 insertions(+), 9 deletions(-)

diff --git a/src/lvmt/storage.rs b/src/lvmt/storage.rs
index ebada00..d0d59dd 100644
--- a/src/lvmt/storage.rs
+++ b/src/lvmt/storage.rs
@@ -51,7 +51,7 @@ impl<'cache, 'db> LvmtStore<'cache, 'db> {
         &mut self,
         old_commit: Option<CommitID>,
         new_commit: CommitID,
-        changes: impl Iterator<Item = (Box<[u8]>, Box<[u8]>)>,
+        changes: impl Iterator<Item = (Box<[u8]>, Option<Box<[u8]>>)>,
         write_schema: &impl WriteSchemaTrait,
         pp: &AmtParams<PE>,
     ) -> Result<()> {
diff --git a/src/lvmt/types/lvmt_value.rs b/src/lvmt/types/lvmt_value.rs
index 9547f85..fb989f8 100644
--- a/src/lvmt/types/lvmt_value.rs
+++ b/src/lvmt/types/lvmt_value.rs
@@ -8,32 +8,48 @@ use super::allocation::AllocatePosition;
 pub struct LvmtValue {
     pub(in crate::lvmt) allocation: AllocatePosition,
     pub(in crate::lvmt) version: u64,
-    pub(in crate::lvmt) value: Box<[u8]>,
+    pub(in crate::lvmt) value: Option<Box<[u8]>>,
 }
 
 impl Encode for LvmtValue {
     fn encode(&self) -> std::borrow::Cow<[u8]> {
         let mut encoded: Vec<u8> = self.allocation.encode().into_owned();
         encoded.extend(&self.version.to_le_bytes()[0..5]);
-        encoded.extend(&*self.value);
+
+        // Add a flag to indicate whether the value is present
+        let value_present = self.value.is_some() as u8;
+        encoded.push(value_present);
+        if let Some(value) = &self.value {
+            encoded.extend(&**value);
+        }
+
         Cow::Owned(encoded)
     }
 }
 
 impl Decode for LvmtValue {
     fn decode(input: &[u8]) -> DecResult<Cow<Self>> {
-        if input.len() < 6 {
+        if input.len() < 7 {
             return Err(DecodeError::TooShortHeader);
         }
-        let (header, body) = input.split_at(6);
-        let (allocation_raw, version_raw) = header.split_at(1);
+
+        let (header, body) = input.split_at(7);
+        let (allocation_raw, version_raw_with_flag) = header.split_at(1);
+        let (version_raw, flag) = version_raw_with_flag.split_at(5);
+
         let allocation = AllocatePosition::decode(allocation_raw)?.into_owned();
 
         let mut version_bytes = [0u8; 8];
         version_bytes[0..5].copy_from_slice(version_raw);
 
         let version = u64::from_le_bytes(version_bytes);
-        let value = body.to_vec().into_boxed_slice();
+
+        let value_present = flag[0] != 0;
+        let value = if value_present {
+            Some(body.to_vec().into_boxed_slice())
+        } else {
+            None
+        };
 
         Ok(Cow::Owned(Self {
             allocation,
diff --git a/src/lvmt/types/mod.rs b/src/lvmt/types/mod.rs
index ca7616c..3a3194d 100644
--- a/src/lvmt/types/mod.rs
+++ b/src/lvmt/types/mod.rs
@@ -31,8 +31,11 @@ pub mod test_utils {
         prop_oneof![0u64..10000, 0u64..(1 << 40)]
     }
 
-    pub fn value_strategy() -> impl Strategy<Value = Box<[u8]>> {
-        vec(0u8..=255, 0..128).prop_map(|x| x.into_boxed_slice())
+    pub fn value_strategy() -> impl Strategy<Value = Option<Box<[u8]>>> {
+        prop_oneof![
+            vec(0u8..=255, 0..128).prop_map(|x| Some(x.into_boxed_slice())),
+            Just(None),
+        ]
     }
 
     pub fn bytes32_strategy() -> impl Strategy<Value = H256> {