fix: snapshot get() complexity

Author: marvin-j97

Cargo.toml

@@ -2,7 +2,7 @@
 name = "lsm-tree"
 description = "A K.I.S.S. implementation of log-structured merge trees (LSM-trees/LSMTs)"
 license = "MIT OR Apache-2.0"
-version = "2.0.0"
+version = "2.0.2"
 edition = "2021"
 rust-version = "1.74.0"
 readme = "README.md"

src/segment/mod.rs

@@ -297,32 +297,6 @@ impl Segment {
         ));
         reader.lo_initialized = true;
 
-        let iter = reader.filter_map(move |entry| {
-            let entry = fail_iter!(entry);
-
-            // NOTE: We are past the searched key, so we can immediately return None
-            if &*entry.key.user_key > key {
-                None
-            } else {
-                // Check for seqno if needed
-                if let Some(seqno) = seqno {
-                    if entry.key.seqno < seqno {
-                        Some(Ok(InternalValue {
-                            key: entry.key.clone(),
-                            value: entry.value,
-                        }))
-                    } else {
-                        None
-                    }
-                } else {
-                    Some(Ok(InternalValue {
-                        key: entry.key.clone(),
-                        value: entry.value,
-                    }))
-                }
-            }
-        });
-
         // NOTE: For finding a specific seqno,
         // we need to use a reader
         // because nothing really prevents the version
@@ -340,7 +314,30 @@ impl Segment {
         // unfortunately is in the next block
         //
         // Also because of weak tombstones, we may have to look further than the first item we encounter
-        MvccStream::new(iter).next().transpose()
+        let reader = reader.filter(|x| {
+            match x {
+                Ok(entry) => {
+                    // Check for seqno if needed
+                    if let Some(seqno) = seqno {
+                        entry.key.seqno < seqno
+                    } else {
+                        true
+                    }
+                }
+                Err(_) => true,
+            }
+        });
+
+        let Some(entry) = MvccStream::new(reader).next().transpose()? else {
+            return Ok(None);
+        };
+
+        // NOTE: We are past the searched key, so don't return anything
+        if &*entry.key.user_key > key {
+            return Ok(None);
+        }
+
+        Ok(Some(entry))
     }
 
     // NOTE: Clippy false positive
@@ -380,6 +377,7 @@ impl Segment {
     }
 
     // TODO: move segment tests into module, then make pub(crate)
+
     /// Creates an iterator over the `Segment`.
     ///
     /// # Errors
@@ -389,14 +387,7 @@ impl Segment {
     #[allow(clippy::iter_without_into_iter)]
     #[doc(hidden)]
     pub fn iter(&self) -> Range {
-        Range::new(
-            self.offsets.index_block_ptr,
-            self.descriptor_table.clone(),
-            (self.tree_id, self.metadata.id).into(),
-            self.block_cache.clone(),
-            self.block_index.clone(),
-            (std::ops::Bound::Unbounded, std::ops::Bound::Unbounded),
-        )
+        self.range((std::ops::Bound::Unbounded, std::ops::Bound::Unbounded))
     }
 
     /// Creates a ranged iterator over the `Segment`.

tests/snapshot_point_read.rs

@@ -1,6 +1,39 @@
 use lsm_tree::{AbstractTree, Config, SequenceNumberCounter};
 use test_log::test;
 
+#[test]
+fn snapshot_404() -> lsm_tree::Result<()> {
+    let folder = tempfile::tempdir()?;
+
+    let tree = Config::new(&folder)
+        .data_block_size(1_024)
+        .index_block_size(1_024)
+        .open()?;
+
+    tree.insert("a", "a", 0);
+    tree.insert("a2", "a2", 0);
+    tree.insert("c", "c", 0);
+
+    tree.flush_active_memtable(0)?;
+
+    assert!(tree.get("a")?.is_some());
+    assert!(tree.get("a2")?.is_some());
+    assert!(tree.get("b")?.is_none());
+    assert!(tree.get("c")?.is_some());
+
+    assert!(tree.get_with_seqno("a", 0)?.is_none());
+    assert!(tree.get_with_seqno("a2", 0)?.is_none());
+    assert!(tree.get_with_seqno("b", 0)?.is_none());
+    assert!(tree.get_with_seqno("c", 0)?.is_none());
+
+    assert!(tree.get_with_seqno("a", 1)?.is_some());
+    assert!(tree.get_with_seqno("a2", 1)?.is_some());
+    assert!(tree.get_with_seqno("b", 1)?.is_none());
+    assert!(tree.get_with_seqno("c", 1)?.is_some());
+
+    Ok(())
+}
+
 #[test]
 fn snapshot_lots_of_versions() -> lsm_tree::Result<()> {
     let version_count = 600;