fix snapshot isolation in point lookups

Author: zaidoon1

src/tree/mod.rs

@@ -84,26 +84,37 @@ impl AbstractTree for Tree {
         let version_history_lock = self.version_history.read().expect("lock is poisoned");
         let super_version = version_history_lock.get_version_for_snapshot(seqno);
 
-        // Active memtable has the newest writes; if present, it's definitive.
-        if let Some(entry) = super_version.active_memtable.get(key, seqno) {
-            return Ok(ignore_tombstone_value(entry));
+        // Track best candidate by seqno across all sources
+        let mut best: Option<InternalValue> = super_version.active_memtable.get(key, seqno);
+        // If the memtable hit matches the snapshot seqno, it's the maximal visible version
+        if let Some(ref entry) = best {
+            if entry.key.seqno == seqno {
+                return Ok(ignore_tombstone_value(entry.clone()));
+            }
         }
 
-        // Sealed memtables are newer than any tables; return on first hit.
-        if let Some(entry) =
-            Self::get_internal_entry_from_sealed_memtables(&super_version, key, seqno)
-        {
-            return Ok(ignore_tombstone_value(entry));
+        // Sealed memtables are older than active; only consult them on miss
+        if best.is_none() {
+            if let Some(entry) =
+                Self::get_internal_entry_from_sealed_memtables(&super_version, key, seqno)
+            {
+                best = Some(entry);
+            }
         }
 
-        // Otherwise consult tables (levels top-down, runs newest-first) and return first hit.
         if let Some(entry) =
             self.get_internal_entry_from_tables(&super_version.version, key, seqno)?
         {
+            if let Some(b) = best {
+                if b.key.seqno >= entry.key.seqno {
+                    return Ok(ignore_tombstone_value(b));
+                }
+            }
             return Ok(ignore_tombstone_value(entry));
         }
 
-        Ok(None)
+        // Apply tombstone semantics after selecting the newest version
+        Ok(best.and_then(ignore_tombstone_value))
     }
 
     fn current_version(&self) -> Version {
@@ -650,7 +661,6 @@ impl Tree {
         log::debug!("Finalized table write at {}", table_file_path.display());
 
         let pin_filter = self.config.filter_block_pinning_policy.get(0);
-        // Index blocks use the index pinning policy
         let pin_index = self.config.index_block_pinning_policy.get(0);
 
         let created_table = Table::recover(

tests/ingestion_api.rs

@@ -9,7 +9,6 @@ fn tree_ingestion_tombstones_delete_existing_keys() -> lsm_tree::Result<()> {
         let key = format!("k{:03}", i);
         tree.insert(key.as_bytes(), b"v", 0);
     }
-
     let mut ingest = tree.ingestion()?.with_seqno(10);
     for i in 0..10u32 {
         let key = format!("k{:03}", i);
@@ -26,6 +25,150 @@ fn tree_ingestion_tombstones_delete_existing_keys() -> lsm_tree::Result<()> {
     Ok(())
 }
 
+#[test]
+fn sealed_memtable_value_overrides_table_value() -> lsm_tree::Result<()> {
+    use lsm_tree::AbstractTree;
+    let folder = tempfile::tempdir()?;
+    let tree = lsm_tree::Config::new(folder, Default::default()).open()?;
+
+    // Older table value via ingestion (seqno 1)
+    {
+        let mut ingest = tree.ingestion()?.with_seqno(1);
+        ingest.write(b"k".as_slice().into(), b"old".as_slice().into())?;
+        ingest.finish()?;
+    }
+
+    // Newer value in memtable (seqno 2), then seal it
+    tree.insert(b"k", b"new", 2);
+    let _ = tree.rotate_memtable(); // move active -> sealed
+
+    // Read should return the sealed memtable value
+    assert_eq!(
+        tree.get(b"k", lsm_tree::SeqNo::MAX)?,
+        Some(b"new".as_slice().into())
+    );
+    Ok(())
+}
+
+#[test]
+fn sealed_memtable_tombstone_overrides_table_value() -> lsm_tree::Result<()> {
+    use lsm_tree::AbstractTree;
+    let folder = tempfile::tempdir()?;
+    let tree = lsm_tree::Config::new(folder, Default::default()).open()?;
+
+    // Older table value via ingestion (seqno 1)
+    {
+        let mut ingest = tree.ingestion()?.with_seqno(1);
+        ingest.write(b"k".as_slice().into(), b"old".as_slice().into())?;
+        ingest.finish()?;
+    }
+
+    // Newer tombstone in memtable (seqno 2), then seal it
+    tree.remove(b"k", 2);
+    let _ = tree.rotate_memtable();
+
+    // Read should see the delete from sealed memtable
+    assert!(tree.get(b"k", lsm_tree::SeqNo::MAX)?.is_none());
+    Ok(())
+}
+
+#[test]
+fn tables_newest_first_returns_highest_seqno() -> lsm_tree::Result<()> {
+    use lsm_tree::AbstractTree;
+    let folder = tempfile::tempdir()?;
+    let tree = lsm_tree::Config::new(folder, Default::default()).open()?;
+
+    // Two separate ingestions create two tables containing the same key at different seqnos
+    {
+        let mut ingest = tree.ingestion()?.with_seqno(1);
+        ingest.write(b"k".as_slice().into(), b"v1".as_slice().into())?;
+        ingest.finish()?;
+    }
+    {
+        let mut ingest = tree.ingestion()?.with_seqno(2);
+        ingest.write(b"k".as_slice().into(), b"v2".as_slice().into())?;
+        ingest.finish()?;
+    }
+
+    // With memtables empty, read should return the value from the newest table run (seqno 2)
+    assert_eq!(
+        tree.get(b"k", lsm_tree::SeqNo::MAX)?,
+        Some(b"v2".as_slice().into())
+    );
+    Ok(())
+}
+
+#[test]
+#[should_panic(expected = "next key in ingestion must be greater than last key")]
+fn ingestion_enforces_order_standard_panics() {
+    let folder = tempfile::tempdir().unwrap();
+    let tree = lsm_tree::Config::new(folder, Default::default())
+        .open()
+        .unwrap();
+
+    let mut ingest = tree.ingestion().unwrap().with_seqno(1);
+    // First write higher key, then lower to trigger ordering assertion
+    ingest
+        .write(b"k2".as_slice().into(), b"v".as_slice().into())
+        .unwrap();
+    // Panics here
+    let _ = ingest.write(b"k1".as_slice().into(), b"v".as_slice().into());
+}
+
+#[test]
+fn blob_ingestion_out_of_order_panics_without_blob_write() -> lsm_tree::Result<()> {
+    let folder = tempfile::tempdir()?;
+    let tree = lsm_tree::Config::new(folder, Default::default())
+        .with_kv_separation(Some(KvSeparationOptions::default().separation_threshold(8)))
+        .open()?;
+
+    let before = tree.blob_file_count();
+
+    // Use a small value for the first write to avoid blob I/O
+    let result = std::panic::catch_unwind(|| {
+        let mut ingest = tree.ingestion().unwrap().with_seqno(1);
+        ingest
+            .write(b"k2".as_slice().into(), b"x".as_slice().into())
+            .unwrap();
+        // Second write would require blob I/O, but ordering check should fire before any blob write
+        let _ = ingest.write(b"k1".as_slice().into(), vec![1u8; 16].into());
+    });
+    assert!(result.is_err());
+
+    let after = tree.blob_file_count();
+    assert_eq!(before, after);
+    Ok(())
+}
+
+#[test]
+fn memtable_put_overrides_table_tombstone() -> lsm_tree::Result<()> {
+    use lsm_tree::AbstractTree;
+    let folder = tempfile::tempdir()?;
+    let tree = lsm_tree::Config::new(folder, Default::default()).open()?;
+
+    // Older put written via ingestion to tables (seqno 1)
+    {
+        let mut ingest = tree.ingestion()?.with_seqno(1);
+        ingest.write(b"k".as_slice().into(), b"v1".as_slice().into())?;
+        ingest.finish()?;
+    }
+
+    // Newer tombstone written via ingestion to tables (seqno 2)
+    {
+        let mut ingest = tree.ingestion()?.with_seqno(2);
+        ingest.write_tombstone(b"k".as_slice().into())?;
+        ingest.finish()?;
+    }
+
+    // Newest put in active memtable (seqno 3) should override table tombstone
+    tree.insert(b"k", b"v3", 3);
+    assert_eq!(
+        tree.get(b"k", lsm_tree::SeqNo::MAX)?,
+        Some(b"v3".as_slice().into())
+    );
+    Ok(())
+}
+
 #[test]
 fn blob_tree_ingestion_tombstones_delete_existing_keys() -> lsm_tree::Result<()> {
     let folder = tempfile::tempdir()?;