chore(cubestore): QueueResult - use same pk as QueueItem (#6546)

Author: ovr

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

@@ -825,7 +825,12 @@ impl CacheStore for RocksCacheStore {
 
                     if let Some(result) = result {
                         let queue_result = QueueResult::new(path.clone(), result.clone());
-                        let result_row = result_schema.insert(queue_result, batch_pipe)?;
+                        // QueueResult is a result of QueueItem, it's why we can use row_id of QueueItem
+                        let result_row = result_schema.insert_with_pk(
+                            item_row.get_id(),
+                            queue_result,
+                            batch_pipe,
+                        )?;
 
                         batch_pipe.add_event(MetaStoreEvent::AckQueueItem(QueueResultAckEvent {
                             path,

rust/cubestore/cubestore/src/cachestore/queue_result.rs

@@ -15,7 +15,11 @@ pub struct QueueResult {
     pub(crate) expire: DateTime<Utc>,
 }
 
-impl RocksEntity for QueueResult {}
+impl RocksEntity for QueueResult {
+    fn version() -> u32 {
+        2
+    }
+}
 
 impl QueueResult {
     pub fn new(path: String, value: String) -> Self {

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

@@ -376,8 +376,10 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
         Ok(())
     }
 
-    fn insert(
+    /// @internal Do not use this method directly, please use insert or insert_with_pk
+    fn do_insert(
         &self,
+        row_id: Option<u64>,
         row: Self::T,
         batch_pipe: &mut BatchPipe,
     ) -> Result<IdRow<Self::T>, CubeError> {
@@ -401,7 +403,13 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
             }
         }
 
-        let (row_id, inserted_row) = self.insert_row(serialized_row)?;
+        let row_id = if let Some(row_id) = row_id {
+            row_id
+        } else {
+            self.next_table_seq()?
+        };
+        let inserted_row = self.insert_row(row_id, serialized_row)?;
+
         batch_pipe.add_event(MetaStoreEvent::Insert(Self::table_id(), row_id));
         if self.snapshot().get(&inserted_row.key)?.is_some() {
             return Err(CubeError::internal(format!("Primary key constraint violation. Primary key already exists for a row id {}: {:?}", row_id, &row)));
@@ -419,6 +427,23 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
         Ok(IdRow::new(row_id, row))
     }
 
+    fn insert_with_pk(
+        &self,
+        row_id: u64,
+        row: Self::T,
+        batch_pipe: &mut BatchPipe,
+    ) -> Result<IdRow<Self::T>, CubeError> {
+        self.do_insert(Some(row_id), row, batch_pipe)
+    }
+
+    fn insert(
+        &self,
+        row: Self::T,
+        batch_pipe: &mut BatchPipe,
+    ) -> Result<IdRow<Self::T>, CubeError> {
+        self.do_insert(None, row, batch_pipe)
+    }
+
     fn migrate(&self) -> Result<(), CubeError> {
         self.migration_check_table()?;
         self.migration_check_indexes()?;
@@ -444,6 +469,15 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
             {
                 let mut batch = WriteBatch::default();
 
+                log::trace!(
+                    "Migrating table {:?} from [{}, {}] to [{}, {}]",
+                    Self::table_id(),
+                    table_info.version,
+                    table_info.value_version,
+                    Self::T::version(),
+                    Self::T::value_version(),
+                );
+
                 self.migrate_table(&mut batch, table_info)?;
 
                 batch.put(
@@ -798,14 +832,14 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
         Ok(next_seq)
     }
 
-    fn insert_row(&self, row: Vec<u8>) -> Result<(u64, KeyVal), CubeError> {
-        let next_seq = self.next_table_seq()?;
-        let t = RowKey::Table(Self::table_id(), next_seq);
+    fn insert_row(&self, row_id: u64, row: Vec<u8>) -> Result<KeyVal, CubeError> {
+        let t = RowKey::Table(Self::table_id(), row_id);
         let res = KeyVal {
             key: t.to_bytes(),
             val: row,
         };
-        Ok((next_seq, res))
+
+        Ok(res)
     }
 
     fn update_row(&self, row_id: u64, row: Vec<u8>) -> Result<KeyVal, CubeError> {