dft

Author: ktmlm

src/compaction/leveled.rs

@@ -1023,7 +1023,15 @@ impl LeveledCompaction {
                     CompactionFilterDecision::Keep => {}
                     CompactionFilterDecision::Remove => continue,
                     CompactionFilterDecision::ChangeValue(_) => {
-                        // force_merge_level doesn't rewrite values; treat as Keep
+                        // force_merge_level doesn't rewrite values — the output
+                        // SST reuses the existing encoded block. Silently treat
+                        // as Keep. Regular level-to-level compaction handles
+                        // ChangeValue correctly.
+                        debug_assert!(
+                            false,
+                            "CompactionFilter returned ChangeValue in force_merge_level, \
+                             which cannot rewrite values — treating as Keep"
+                        );
                     }
                 }
             }

src/db.rs

@@ -24,7 +24,9 @@ use crate::manifest::version_edit::{FileMetaData, VersionEdit};
 use crate::manifest::version_set::VersionSet;
 use crate::memtable::MemTable;
 use crate::memtable::skiplist::MemTableCursorIter;
-use crate::options::{CompactionFilter, CompactionFilterDecision, DbOptions, ReadOptions, WriteOptions};
+use crate::options::{
+    CompactionFilter, CompactionFilterDecision, DbOptions, ReadOptions, WriteOptions,
+};
 use crate::rate_limiter::RateLimiter;
 use crate::sst::table_builder::{TableBuildOptions, TableBuilder};
 use crate::sst::table_reader::TableIterator;
@@ -1582,25 +1584,35 @@ impl DB {
     /// **Note:** The key remains readable via `get()` until compaction
     /// physically removes it. Use regular `delete()` if immediate
     /// invisibility is required.
+    ///
+    /// **Warning:** Dead keys are held in memory only and are **not**
+    /// persisted to WAL or SST. If the process crashes or restarts
+    /// before compaction removes them, the registrations are lost and
+    /// the keys will remain in the database permanently unless
+    /// re-registered.
     pub fn lazy_delete(&self, key: &[u8]) {
         let mut set = self.dead_keys.write().unwrap();
         set.insert(key.to_vec());
     }
 
     /// Batch version of [`lazy_delete`](Self::lazy_delete).
     ///
-    /// If the number of accumulated dead keys exceeds
+    /// Dead keys are **not** persisted — see [`lazy_delete`](Self::lazy_delete)
+    /// for durability caveats.
+    ///
+    /// If the number of accumulated dead keys newly crosses
     /// `lazy_delete_compaction_threshold`, a background compaction is
     /// automatically signalled.
-    pub fn lazy_delete_batch(&self, keys: &[Vec<u8>]) {
+    pub fn lazy_delete_batch(&self, keys: impl IntoIterator<Item = impl AsRef<[u8]>>) {
         let threshold = self.options.lazy_delete_compaction_threshold;
         let mut set = self.dead_keys.write().unwrap();
+        let prev_len = set.len();
         for k in keys {
-            set.insert(k.clone());
+            set.insert(k.as_ref().to_vec());
         }
         let len = set.len();
         drop(set);
-        if threshold > 0 && len >= threshold {
+        if threshold > 0 && len >= threshold && prev_len < threshold {
             self.signal_compaction();
         }
     }
@@ -1610,8 +1622,13 @@ impl DB {
         self.dead_keys.read().unwrap().len()
     }
 
-    /// Clear all dead keys. Call after compaction has processed all levels
-    /// to reclaim memory used by the dead-keys set.
+    /// Clear all dead keys to reclaim memory used by the dead-keys set.
+    ///
+    /// **Caution:** Only call this after a full compaction
+    /// ([`compact_range(None, None)`](Self::compact_range)) has
+    /// finished. Any dead keys that have not yet been visited by
+    /// compaction will be silently abandoned — those keys will remain
+    /// in the database and will not be removed unless re-registered.
     pub fn clear_dead_keys(&self) {
         self.dead_keys.write().unwrap().clear();
     }

tests/lazy_delete.rs

@@ -97,3 +97,72 @@ fn clear_dead_keys_resets_count() {
     db.clear_dead_keys();
     assert_eq!(db.dead_key_count(), 0);
 }
+
+/// Verify that `lazy_delete_batch` triggers background compaction when the
+/// dead-key count newly crosses `lazy_delete_compaction_threshold`.
+#[test]
+fn lazy_delete_batch_auto_triggers_compaction() {
+    let dir = tempfile::tempdir().unwrap();
+    let db = DB::open(
+        DbOptions {
+            create_if_missing: true,
+            write_buffer_size: 1024,
+            lazy_delete_compaction_threshold: 5,
+            ..Default::default()
+        },
+        dir.path(),
+    )
+    .unwrap();
+
+    // Write enough data across flushes so that compaction has work to do.
+    for i in 0u32..20 {
+        db.put(&i.to_be_bytes(), &[i as u8; 64]).unwrap();
+    }
+
+    // Register 6 keys — crosses the threshold of 5, should signal compaction.
+    let dead: Vec<Vec<u8>> = (0u32..6).map(|i| i.to_be_bytes().to_vec()).collect();
+    db.lazy_delete_batch(&dead);
+
+    // Give background compaction a moment to run.
+    std::thread::sleep(std::time::Duration::from_secs(2));
+
+    // The dead keys should have been removed by background compaction.
+    for i in 0u32..6 {
+        assert!(
+            db.get(&i.to_be_bytes()).unwrap().is_none(),
+            "key {} should have been removed by auto-triggered compaction",
+            i
+        );
+    }
+
+    // Surviving keys should still be readable.
+    for i in 6u32..20 {
+        assert!(
+            db.get(&i.to_be_bytes()).unwrap().is_some(),
+            "key {} should survive",
+            i
+        );
+    }
+}
+
+/// Verify that `lazy_delete_batch` accepts `&[&[u8]]` slices (not just
+/// `Vec<u8>`), confirming the generic `impl AsRef<[u8]>` signature works.
+#[test]
+fn lazy_delete_batch_accepts_slices() {
+    let dir = tempfile::tempdir().unwrap();
+    let db = make_db(dir.path());
+
+    db.put(b"a", b"1").unwrap();
+    db.put(b"pad", &[0u8; 1024]).unwrap();
+    db.put(b"b", b"2").unwrap();
+
+    // Pass &[u8] slices — must compile with the generic signature.
+    let keys: Vec<&[u8]> = vec![b"a", b"b"];
+    db.lazy_delete_batch(keys);
+    assert_eq!(db.dead_key_count(), 2);
+
+    db.compact_range(None::<&[u8]>, None::<&[u8]>).unwrap();
+
+    assert!(db.get(b"a").unwrap().is_none());
+    assert!(db.get(b"b").unwrap().is_none());
+}