Allow keys without zero byte terminator

Cargo.toml

@@ -32,7 +32,7 @@ futures = "0.3.30"
 bytes = "1.5.0"
 tokio = { version = "1.36", features = ["rt", "sync"] }
 quick_cache = "0.6.0"
-vart = "0.7.0"
+vart = "0.8.0"
 revision = "0.10.0"
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 getrandom = { version = "0.2.15", features = ["js"] }
@@ -56,3 +56,7 @@ walkdir = "2.5.0"
 [[bench]]
 name = "store_bench"
 harness = false
+
+[[bench]]
+name = "load_bench"
+harness = false

src/storage/kv/compaction.rs

@@ -112,8 +112,6 @@ impl StoreInner {
                                ts: u64,
                                metadata: Option<Metadata>|
          -> Result<()> {
-            let mut key = key;
-            key.truncate(key.len() - 1);
             let mut entry = Entry::new(&key, &value);
             entry.set_ts(ts);
 

src/storage/kv/indexer.rs

@@ -31,7 +31,6 @@ impl Indexer {
         ts: u64,
         check_version: bool,
     ) -> Result<()> {
-        *key = key.terminate();
         if check_version {
             self.index.insert(key, value, version, ts)?;
         } else {
@@ -48,7 +47,6 @@ impl Indexer {
         ts: u64,
         check_version: bool,
     ) -> Result<()> {
-        *key = key.terminate();
         if check_version {
             self.index.insert_or_replace(key, value, version, ts)?;
         } else {
@@ -59,7 +57,6 @@ impl Indexer {
     }
 
     pub fn delete(&mut self, key: &mut VariableSizeKey) {
-        *key = key.terminate();
         self.index.remove(key);
     }
 

src/storage/kv/oracle.rs

@@ -270,7 +270,7 @@ fn key_in_range(
     range_start: &Bound<VariableSizeKey>,
     range_end: &Bound<VariableSizeKey>,
 ) -> bool {
-    let key = VariableSizeKey::from_slice_with_termination(key);
+    let key = VariableSizeKey::from_slice(key);
 
     let start_inclusive = match &range_start {
         Bound::Included(start) => key >= *start,

src/storage/kv/snapshot.rs

@@ -43,21 +43,13 @@ impl Snapshot {
 
     /// Set a key-value pair into the snapshot.
     pub(crate) fn set(&mut self, key: &VariableSizeKey, value: IndexValue) {
-        // TODO: need to fix this to avoid cloning the key
-        // This happens because the VariableSizeKey transfrom from
-        // a &[u8] does not terminate the key with a null byte.
-        let key = &key.terminate();
         self.snap
             .insert(key, value, self.version, now())
             .expect("incorrect snapshot version");
     }
 
     #[allow(unused)]
     pub(crate) fn delete(&mut self, key: &VariableSizeKey) -> bool {
-        // TODO: need to fix this to avoid cloning the key
-        // This happens because the VariableSizeKey transfrom from
-        // a &[u8] does not terminate the key with a null byte.
-        let key = &key.terminate();
         self.snap.remove(key)
     }
 
@@ -71,18 +63,13 @@ impl Snapshot {
 
     /// Retrieves the latest value associated with the given key from the snapshot.
     pub(crate) fn get(&self, key: &VariableSizeKey) -> Result<(IndexValue, u64)> {
-        // TODO: need to fix this to avoid cloning the key
-        // This happens because the VariableSizeKey transfrom from
-        // a &[u8] does not terminate the key with a null byte.
-        let key = &key.terminate();
         let (snap_val, version, _) = self.snap.get(key, self.version).ok_or(Error::KeyNotFound)?;
         let val = self.apply_filters(snap_val, &FILTERS)?;
         Ok((val, version))
     }
 
     /// Retrieves the value associated with the given key at the given timestamp from the snapshot.
     pub(crate) fn get_at_ts(&self, key: &VariableSizeKey, ts: u64) -> Result<IndexValue> {
-        let key = &key.terminate();
         let (val, _, _) = self.snap.get_at_ts(key, ts).ok_or(Error::KeyNotFound)?;
         self.apply_filters(val, &FILTERS)
     }
@@ -92,8 +79,6 @@ impl Snapshot {
         &self,
         key: &VariableSizeKey,
     ) -> Result<Vec<(IndexValue, u64)>> {
-        let key = &key.terminate();
-
         let mut results = Vec::new();
 
         let items = self

src/storage/kv/transaction.rs

@@ -437,8 +437,7 @@ impl Transaction {
             // Only add the key to the read set if the version is less than or equal to the
             // read timestamp. This is to prevent adding keys that are added during the transaction.
             if *version <= self.read_ts {
-                // the keys in the vart leaf are terminated with a null byte
-                let key = Bytes::copy_from_slice(&key[..&key.len() - 1]);
+                let key = Bytes::copy_from_slice(&key);
                 let entry = ReadSetEntry::new(key, *version, self.savepoints);
                 self.read_set.push(entry);
             }
@@ -447,8 +446,6 @@ impl Transaction {
             let v = value.resolve(&self.core)?;
 
             // Add the value, version, and timestamp to the results vector.
-            let mut key = key;
-            key.truncate(key.len() - 1);
             results.push((key, v, *version, *ts));
         }
 
@@ -466,15 +463,7 @@ impl Transaction {
         // Convert the range to a tuple of bounds of variable keys.
         let range = convert_range_bounds(range);
         let keys = self.snapshot.as_ref().unwrap().range(range);
-
-        // Remove the trailing `\0`.
-        let result = keys
-            .into_iter()
-            .map(|(mut key, _, _, _)| {
-                key.truncate(key.len() - 1);
-                key
-            })
-            .collect();
+        let result = keys.into_iter().map(|(key, _, _, _)| key).collect();
 
         Ok(result)
     }
@@ -761,7 +750,7 @@ impl Transaction {
         let items = self.snapshot.as_ref().unwrap().scan_at_ts(range, ts);
 
         let mut results = Vec::new();
-        'outer: for (mut key, value) in items {
+        'outer: for (key, value) in items {
             // If a limit is set and we've already got enough results, break the loop.
             if let Some(limit) = limit {
                 if results.len() >= limit {
@@ -779,8 +768,6 @@ impl Transaction {
             // Resolve the value reference to get the actual value.
             let v = value.resolve(&self.core)?;
 
-            // Remove the trailing `\0`.
-            key.truncate(key.len() - 1);
             results.push((key, v));
         }
 
@@ -798,16 +785,7 @@ impl Transaction {
         let range = convert_range_bounds(range);
         let keys = self.snapshot.as_ref().unwrap().keys_at_ts(range, ts);
 
-        // Remove the trailing `\0`.
-        let result = keys
-            .into_iter()
-            .map(|mut key| {
-                key.truncate(key.len() - 1);
-                key
-            })
-            .collect();
-
-        Ok(result)
+        Ok(keys)
     }
 
     /// Returns the value associated with the key at the given timestamp.
@@ -866,10 +844,7 @@ impl Transaction {
         let mut current_key_versions = Vec::new();
 
         // Iterate over the keys in the range.
-        for (mut key, value, _, ts) in ranger {
-            // Remove the trailing `\0`.
-            key.truncate(key.len() - 1);
-
+        for (key, value, _, ts) in ranger {
             // If the key changes, process the previous key's versions.
             if current_key.as_ref().map_or(false, |k| k != &key) {
                 // Add the previous key's versions to the results.

src/storage/kv/util.rs

@@ -74,17 +74,13 @@ where
 {
     // Step 2: Apply the conversion logic for both start and end bounds
     let start_bound = match range.start_bound() {
-        Bound::Included(start) => {
-            Bound::Included(VariableSizeKey::from_slice_with_termination(start))
-        }
-        Bound::Excluded(start) => {
-            Bound::Excluded(VariableSizeKey::from_slice_with_termination(start))
-        }
+        Bound::Included(start) => Bound::Included(VariableSizeKey::from_slice(start)),
+        Bound::Excluded(start) => Bound::Excluded(VariableSizeKey::from_slice(start)),
         Bound::Unbounded => Bound::Unbounded,
     };
     let end_bound = match range.end_bound() {
-        Bound::Included(end) => Bound::Included(VariableSizeKey::from_slice_with_termination(end)),
-        Bound::Excluded(end) => Bound::Excluded(VariableSizeKey::from_slice_with_termination(end)),
+        Bound::Included(end) => Bound::Included(VariableSizeKey::from_slice(end)),
+        Bound::Excluded(end) => Bound::Excluded(VariableSizeKey::from_slice(end)),
         Bound::Unbounded => Bound::Unbounded,
     };
     (start_bound, end_bound)