compaction

cozorocks/bridge/db.h

@@ -108,6 +108,7 @@ struct RocksDbBridge {
     [[nodiscard]] virtual unique_ptr<TxBridge> transact() const = 0;
 
     virtual void del_range(RustBytes start, RustBytes end, RocksDbStatus &status) const = 0;
+    virtual void compact_range(RustBytes start, RustBytes end, RocksDbStatus &status) const = 0;
 
     [[nodiscard]] inline const string &get_db_path() const {
         return db_path;
@@ -124,6 +125,13 @@ struct OptimisticRocksDb : public RocksDbBridge {
     }
 
     void del_range(RustBytes, RustBytes, RocksDbStatus &status) const override;
+    void compact_range(RustBytes start, RustBytes end, RocksDbStatus &status) const override {
+        CompactRangeOptions options;
+        auto start_s = convert_slice(start);
+        auto end_s = convert_slice(end);
+        auto s = db->CompactRange(options, &start_s, &end_s);
+        write_status(s, status);
+    }
 
     virtual ~OptimisticRocksDb();
 };
@@ -152,6 +160,14 @@ struct PessimisticRocksDb : public RocksDbBridge {
         write_status(s2, status);
     }
 
+    void compact_range(RustBytes start, RustBytes end, RocksDbStatus &status) const override {
+        CompactRangeOptions options;
+        auto start_s = convert_slice(start);
+        auto end_s = convert_slice(end);
+        auto s = db->CompactRange(options, &start_s, &end_s);
+        write_status(s, status);
+    }
+
     virtual ~PessimisticRocksDb();
 };
 

cozorocks/src/bridge/db.rs

@@ -266,6 +266,16 @@ impl RocksDb {
             Err(status)
         }
     }
+    #[inline]
+    pub fn range_compact(&self, lower: &[u8], upper: &[u8]) -> Result<(), RocksDbStatus> {
+        let mut status = RocksDbStatus::default();
+        self.inner.compact_range(lower, upper, &mut status);
+        if status.is_ok() {
+            Ok(())
+        } else {
+            Err(status)
+        }
+    }
 }
 
 unsafe impl Send for RocksDb {}

cozorocks/src/bridge/mod.rs

@@ -151,6 +151,12 @@ pub(crate) mod ffi {
             upper: &[u8],
             status: &mut RocksDbStatus,
         );
+        fn compact_range(
+            self: &RocksDbBridge,
+            lower: &[u8],
+            upper: &[u8],
+            status: &mut RocksDbStatus,
+        );
 
         type TxBridge;
         // fn get_r_opts(self: Pin<&mut TxBridge>) -> Pin<&mut ReadOptions>;

src/data/encode.rs

@@ -10,7 +10,7 @@ use crate::data::attr::Attribute;
 use crate::data::id::{AttrId, EntityId, TxId, Validity};
 use crate::data::symb::Symbol;
 use crate::data::triple::StoreOp;
-use crate::data::value::DataValue;
+use crate::data::value::{DataValue, LARGEST_UTF_CHAR};
 use crate::runtime::transact::TxLog;
 
 #[repr(u8)]
@@ -250,6 +250,10 @@ pub(crate) fn encode_eav_key(
     ret.into()
 }
 
+pub(crate) fn smallest_key() -> EncodedVec<LARGE_VEC_SIZE> {
+    encode_eav_key(EntityId::ZERO, AttrId(0), &DataValue::Null, Validity::MIN)
+}
+
 #[inline]
 pub(crate) fn decode_ea_key(src: &[u8]) -> Result<(EntityId, AttrId, Validity)> {
     let eid = EntityId::from_bytes(&src[0..VEC_SIZE_8]);
@@ -438,6 +442,12 @@ pub(crate) fn encode_sentinel_attr_by_name(name: &Symbol) -> EncodedVec<LARGE_VE
     ret.into()
 }
 
+pub(crate) fn largest_key() -> EncodedVec<LARGE_VEC_SIZE> {
+    let name = String::from(LARGEST_UTF_CHAR);
+    let symb = Symbol::from(&name as &str);
+    encode_sentinel_attr_by_name(&symb)
+}
+
 #[inline]
 pub(crate) fn decode_sentinel_attr_by_name(src: &[u8]) -> Result<Symbol> {
     Symbol::try_from(&src[1..])

src/runtime/db.rs

@@ -13,7 +13,8 @@ use cozorocks::{DbBuilder, DbIter, RocksDb};
 
 use crate::data::compare::{rusty_cmp, DB_KEY_PREFIX_LEN};
 use crate::data::encode::{
-    decode_ea_key, decode_value_from_key, decode_value_from_val, encode_eav_key, StorageTag,
+    decode_ea_key, decode_value_from_key, decode_value_from_val, encode_eav_key, largest_key,
+    smallest_key, StorageTag,
 };
 use crate::data::id::{AttrId, EntityId, TxId, Validity};
 use crate::data::json::JsonValue;
@@ -83,6 +84,20 @@ impl Db {
         Ok(ret)
     }
 
+    pub fn compact_main(&self) -> Result<()> {
+        let l = smallest_key();
+        let u = largest_key();
+        self.db.range_compact(&l, &u)?;
+        Ok(())
+    }
+
+    pub fn compact_view(&self) -> Result<()> {
+        let l = Tuple::default().encode_as_key(ViewRelId(0));
+        let u = Tuple(vec![DataValue::Bottom]).encode_as_key(ViewRelId(u64::MAX));
+        self.db.range_compact(&l, &u)?;
+        Ok(())
+    }
+
     pub fn new_session(&self) -> Result<Self> {
         let old_count = self.n_sessions.fetch_add(1, Ordering::AcqRel);
 

tests/air_routes.rs

@@ -79,6 +79,14 @@ fn air_routes() -> Result<()> {
 
     println!("views: {}", db.list_views()?);
 
+    let compact_main_time = Instant::now();
+    db.compact_main()?;
+    dbg!(compact_main_time.elapsed());
+
+    let compact_view_time = Instant::now();
+    db.compact_view()?;
+    dbg!(compact_view_time.elapsed());
+
     let deg_centrality_time = Instant::now();
     let res = db.run_script(
         r#"