fix(cubestore): Reduce memory usage while truncating (#7031)

Author: ovr

rust/cubestore/cubestore/src/cachestore/cache_rocksstore.rs

@@ -581,10 +581,7 @@ impl CacheStore for RocksCacheStore {
         self.store
             .write_operation(move |db_ref, batch_pipe| {
                 let cache_schema = CacheItemRocksTable::new(db_ref);
-                let rows = cache_schema.all_rows()?;
-                for row in rows.iter() {
-                    cache_schema.delete(row.get_id(), batch_pipe)?;
-                }
+                cache_schema.truncate(batch_pipe)?;
 
                 Ok(())
             })
@@ -728,16 +725,10 @@ impl CacheStore for RocksCacheStore {
         self.store
             .write_operation(move |db_ref, batch_pipe| {
                 let queue_item_schema = QueueItemRocksTable::new(db_ref.clone());
-                let rows = queue_item_schema.all_rows()?;
-                for row in rows.iter() {
-                    queue_item_schema.delete(row.get_id(), batch_pipe)?;
-                }
+                queue_item_schema.truncate(batch_pipe)?;
 
                 let queue_result_schema = QueueResultRocksTable::new(db_ref);
-                let rows = queue_result_schema.all_rows()?;
-                for row in rows.iter() {
-                    queue_result_schema.delete(row.get_id(), batch_pipe)?;
-                }
+                queue_result_schema.truncate(batch_pipe)?;
 
                 Ok(())
             })

rust/cubestore/cubestore/src/metastore/rocks_table.rs

@@ -284,6 +284,7 @@ where
                 if table_id != self.table_id {
                     return None;
                 }
+
                 Some(self.table.deserialize_id_row(row_id, &value))
             } else {
                 None
@@ -879,6 +880,18 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
         Ok(IdRow::new(row_id, new_row))
     }
 
+    fn truncate(&self, batch_pipe: &mut BatchPipe) -> Result<(), CubeError> {
+        let iter = self.table_scan(self.snapshot())?;
+
+        for item in iter {
+            let item = item?;
+
+            self.delete_impl(item, batch_pipe)?;
+        }
+
+        Ok(())
+    }
+
     fn delete(&self, row_id: u64, batch_pipe: &mut BatchPipe) -> Result<IdRow<Self::T>, CubeError> {
         let row = self.get_row_or_not_found(row_id)?;
         self.delete_impl(row, batch_pipe)