feat: add range tombstone (delete_range) support

src/abstract.rs

@@ -98,9 +98,15 @@ pub trait AbstractTree {
         );
         let stream = CompactionStream::new(merger, seqno_threshold);
 
+        // Collect range tombstones from sealed memtables
+        let mut range_tombstones = Vec::new();
+        for mt in latest.sealed_memtables.iter() {
+            range_tombstones.extend(mt.range_tombstones_by_start());
+        }
+
         drop(version_history);
 
-        if let Some((tables, blob_files)) = self.flush_to_tables(stream)? {
+        if let Some((tables, blob_files)) = self.flush_to_tables(stream, range_tombstones)? {
             self.register_tables(
                 &tables,
                 blob_files.as_deref(),
@@ -216,6 +222,7 @@ pub trait AbstractTree {
     fn flush_to_tables(
         &self,
         stream: impl Iterator<Item = crate::Result<InternalValue>>,
+        range_tombstones: Vec<crate::range_tombstone::RangeTombstone>,
     ) -> crate::Result<Option<FlushToTablesResult>>;
 
     /// Atomically registers flushed tables into the tree, removing their associated sealed memtables.
@@ -598,4 +605,9 @@ pub trait AbstractTree {
     /// Will return `Err` if an IO error occurs.
     #[doc(hidden)]
     fn remove_weak<K: Into<UserKey>>(&self, key: K, seqno: SeqNo) -> (u64, u64);
+
+    /// Inserts a range tombstone that suppresses all keys in `[start, end)` with
+    /// sequence numbers strictly less than `seqno`.
+    #[doc(hidden)]
+    fn remove_range<K: AsRef<[u8]>>(&self, start: K, end: K, seqno: SeqNo);
 }