Delete sst files in key range in rocksdb (#19211)

## Description 

This PR adds a rocksdb endpoint to delete .sst files in key range in
rocksdb which is useful to prune data (in certain scenarios) without
compaction.

crates/typed-store/src/rocks/mod.rs

@@ -337,6 +337,15 @@ impl RocksDB {
         delegate_call!(self.drop_cf(name))
     }
 
+    pub fn delete_file_in_range<K: AsRef<[u8]>>(
+        &self,
+        cf: &impl AsColumnFamilyRef,
+        from: K,
+        to: K,
+    ) -> Result<(), rocksdb::Error> {
+        delegate_call!(self.delete_file_in_range_cf(cf, from, to))
+    }
+
     pub fn delete_cf<K: AsRef<[u8]>>(
         &self,
         cf: &impl AsColumnFamilyRef,
@@ -1989,6 +1998,23 @@ where
         Ok(())
     }
 
+    /// Deletes a range of keys between `from` (inclusive) and `to` (non-inclusive)
+    /// by immediately deleting any sst files whose key range overlaps with the range.
+    /// Files whose range only partially overlaps with the range are not deleted.
+    /// This can be useful for quickly removing a large amount of data without having
+    /// to delete individual keys. Only files at level 1 or higher are considered (
+    /// Level 0 files are skipped). It doesn't guarantee that all keys in the range are
+    /// deleted, as there might be keys in files that weren't entirely within the range.
+    #[instrument(level = "trace", skip_all, err)]
+    fn delete_file_in_range(&self, from: &K, to: &K) -> Result<(), TypedStoreError> {
+        let from_buf = be_fix_int_ser(from.borrow())?;
+        let to_buf = be_fix_int_ser(to.borrow())?;
+        self.rocksdb
+            .delete_file_in_range(&self.cf(), from_buf, to_buf)
+            .map_err(typed_store_err_from_rocks_err)?;
+        Ok(())
+    }
+
     /// This method first drops the existing column family and then creates a new one
     /// with the same name. The two operations are not atomic and hence it is possible
     /// to get into a race condition where the column family has been dropped but new

crates/typed-store/src/test_db.rs

@@ -265,6 +265,13 @@ where
         Ok(())
     }
 
+    fn delete_file_in_range(&self, from: &K, to: &K) -> Result<(), TypedStoreError> {
+        let mut locked = self.rows.write().unwrap();
+        locked
+            .retain(|k, _| k < &be_fix_int_ser(from).unwrap() || k >= &be_fix_int_ser(to).unwrap());
+        Ok(())
+    }
+
     fn schedule_delete_all(&self) -> Result<(), TypedStoreError> {
         let mut locked = self.rows.write().unwrap();
         locked.clear();

crates/typed-store/src/traits.rs

@@ -48,6 +48,9 @@ where
     /// Removes every key-value pair from the map.
     fn unsafe_clear(&self) -> Result<(), Self::Error>;
 
+    /// Removes every key-value pair from the map by deleting the underlying file.
+    fn delete_file_in_range(&self, from: &K, to: &K) -> Result<(), TypedStoreError>;
+
     /// Uses delete range on the entire key range
     fn schedule_delete_all(&self) -> Result<(), TypedStoreError>;