changeset 350:e158f02819cb

memtable: Distinguish between deleted and non-existing entries.
author Lewin Bormann <lbo@spheniscida.de>
date Thu, 05 Oct 2017 19:44:34 +0200
parents 534d28d9fe36
children b46303bd0520
files src/db_impl.rs src/memtable.rs
diffstat 2 files changed, 41 insertions(+), 23 deletions(-) [+]
line wrap: on
line diff
--- a/src/db_impl.rs	Thu Oct 05 19:10:48 2017 +0200
+++ b/src/db_impl.rs	Thu Oct 05 19:44:34 2017 +0200
@@ -384,16 +384,28 @@
         let current = self.current();
         let mut current = current.borrow_mut();
 
+        // Using this lookup key will skip all entries with higher sequence numbers, because they
+        // will compare "Lesser" using the InternalKeyCmp
         let lkey = LookupKey::new(key, seq);
 
-        if let Some(v) = self.mem.get(&lkey) {
-            return Ok(Some(v));
+        match self.mem.get(&lkey) {
+            (Some(v), _) => return Ok(Some(v)),
+            // deleted entry
+            (None, true) => return Ok(None),
+            // not found entry
+            (None, false) => {}
         }
+
         if let Some(imm) = self.imm.as_ref() {
-            if let Some(v) = imm.get(&lkey) {
-                return Ok(Some(v));
+            match imm.get(&lkey) {
+                (Some(v), _) => return Ok(Some(v)),
+                // deleted entry
+                (None, true) => return Ok(None),
+                // not found entry
+                (None, false) => {}
             }
         }
+
         if let Ok(Some((v, st))) = current.get(lkey.internal_key()) {
             if current.update_stats(st) {
                 if let Err(e) = self.maybe_do_compaction() {
--- a/src/memtable.rs	Thu Oct 05 19:10:48 2017 +0200
+++ b/src/memtable.rs	Thu Oct 05 19:44:34 2017 +0200
@@ -40,8 +40,10 @@
         self.map.insert(build_memtable_key(key, value, t, seq), Vec::new())
     }
 
+    /// get returns the value for the given entry and whether the entry is marked as deleted. This
+    /// is to distinguish between not-found and found-deleted.
     #[allow(unused_variables)]
-    pub fn get(&self, key: &LookupKey) -> Option<Vec<u8>> {
+    pub fn get(&self, key: &LookupKey) -> (Option<Vec<u8>>, bool) {
         let mut iter = self.map.iter();
         iter.seek(key.memtable_key());
 
@@ -52,13 +54,13 @@
             // We only care about user key equality here
             if key.user_key() == &foundkey[fkeyoff..fkeyoff + fkeylen] {
                 if tag & 0xff == ValueType::TypeValue as u64 {
-                    return Some(foundkey[valoff..valoff + vallen].to_vec());
+                    return (Some(foundkey[valoff..valoff + vallen].to_vec()), false);
                 } else {
-                    return None;
+                    return (None, true);
                 }
             }
         }
-        None
+        (None, false)
     }
 
     pub fn iter(&self) -> MemtableIterator {
@@ -69,6 +71,8 @@
 /// MemtableIterator is an iterator over a MemTable. It is mostly concerned with converting to and
 /// from the MemtableKey format used in the inner map; all key-taking or -returning methods deal
 /// with InternalKeys.
+///
+/// This iterator skips deleted entries.
 pub struct MemtableIterator {
     skipmapiter: SkipMapIter,
 }
@@ -183,14 +187,14 @@
 
     fn get_memtable() -> MemTable {
         let mut mt = MemTable::new(options::for_test().cmp);
-        let entries = vec![(115, "abc", "122"),
-                           (120, "abc", "123"),
-                           (121, "abd", "124"),
-                           (122, "abe", "125"),
-                           (123, "abf", "126")];
+        let entries = vec![(ValueType::TypeValue, 115, "abc", "122"),
+                           (ValueType::TypeValue, 120, "abc", "123"),
+                           (ValueType::TypeValue, 121, "abd", "124"),
+                           (ValueType::TypeDeletion, 122, "abe", "125"),
+                           (ValueType::TypeValue, 123, "abf", "126")];
 
         for e in entries.iter() {
-            mt.add(e.0, ValueType::TypeValue, e.1.as_bytes(), e.2.as_bytes());
+            mt.add(e.1, e.0, e.2.as_bytes(), e.3.as_bytes());
         }
         mt
     }
@@ -220,37 +224,38 @@
         let mt = get_memtable();
 
         // Smaller sequence number doesn't find entry
-        if let Some(v) = mt.get(&LookupKey::new("abc".as_bytes(), 110)) {
+        if let Some(v) = mt.get(&LookupKey::new("abc".as_bytes(), 110)).0 {
             println!("{:?}", v);
             panic!("found");
         }
 
-        if let Some(v) = mt.get(&LookupKey::new("abf".as_bytes(), 110)) {
+        if let Some(v) = mt.get(&LookupKey::new("abf".as_bytes(), 110)).0 {
             println!("{:?}", v);
             panic!("found");
         }
 
         // Bigger sequence number falls back to next smaller
-        if let Some(v) = mt.get(&LookupKey::new("abc".as_bytes(), 116)) {
+        if let Some(v) = mt.get(&LookupKey::new("abc".as_bytes(), 116)).0 {
             assert_eq!(v, "122".as_bytes());
         } else {
             panic!("not found");
         }
 
         // Exact match works
-        if let Some(v) = mt.get(&LookupKey::new("abc".as_bytes(), 120)) {
+        if let (Some(v), deleted) = mt.get(&LookupKey::new("abc".as_bytes(), 120)) {
             assert_eq!(v, "123".as_bytes());
+            assert!(!deleted);
         } else {
             panic!("not found");
         }
 
-        if let Some(v) = mt.get(&LookupKey::new("abe".as_bytes(), 122)) {
-            assert_eq!(v, "125".as_bytes());
+        if let (None, deleted) = mt.get(&LookupKey::new("abe".as_bytes(), 122)) {
+            assert!(deleted);
         } else {
-            panic!("not found");
+            panic!("found deleted");
         }
 
-        if let Some(v) = mt.get(&LookupKey::new("abf".as_bytes(), 129)) {
+        if let Some(v) = mt.get(&LookupKey::new("abf".as_bytes(), 129)).0 {
             assert_eq!(v, "126".as_bytes());
         } else {
             panic!("not found");
@@ -303,7 +308,8 @@
                                                * higher sequence number comes first */
                             "122".as_bytes(),
                             "124".as_bytes(),
-                            "125".as_bytes(),
+                            // deleted entry:
+                            // "125".as_bytes(),
                             "126".as_bytes()];
         let mut i = 0;