perf(cubestore): Reduce memory allocations with pinned slices (#9901)

The pinnable slice optimization (get_pinned() instead of get()) eliminates unnecessary data copying when reading from RocksDB, providing substantial memory savings, especially for read-heavy operations.

## Memory stats

  queue_get (queues: 5, size: 128 KB, per_queue: 10000/128)

  | Metric            | Before               | After                | Change                | Change % |
  |-------------------|----------------------|----------------------|-----------------------|----------|
  | Total allocated   | 84,383,988,026 bytes | 65,841,827,472 bytes | −18,542,160,554 bytes | −21.97%  |
  | Peak allocated    | 105,403,976 bytes    | 105,143,200 bytes    | −260,776 bytes        | −0.25%   |
  | Allocations       | 924,523,001          | 924,192,417          | −330,584              | −0.036%  |
  | Reallocations     | 3,307,477            | 3,303,015            | −4,462                | −0.135%  |

  queue_add (queues: 1, size: 512 KB/512)

  | Metric            | Before               | After                | Change             | Change % |
  |-------------------|----------------------|----------------------|--------------------|----------|
  | Total allocated   | 99,058,945,725 bytes | 98,877,885,492 bytes | −181,060,233 bytes | −0.183%  |
  | Peak allocated    | 105,926,929 bytes    | 105,713,460 bytes    | −213,469 bytes     | −0.202%  |
  | Allocations       | 1,370,479,237        | 1,370,357,307        | −121,930           | −0.0089% |
  | Reallocations     | 3,102,373            | 3,099,589            | −2,784             | −0.090%  |

  Summary

  | Operation | Total Memory Saved | Improvement     |
  |-----------|--------------------|-----------------|
  | queue_get | 18.5 GB            | 22% reduction   |
  | queue_add | 181 MB             | 0.18% reduction |

## CPU stats

  | Benchmark                                    | Parameters | Before (Min) | After (Min) | % Change |
  |----------------------------------------------|------------|--------------|-------------|----------|
  | get_tables_with_path_include_non_ready_true  | 100        | 17.734 ms    | 17.341 ms   | -2.22% ✅ |
  | get_tables_with_path_include_non_ready_true  | 1000       | 84.730 ms    | 81.181 ms   | -4.19% ✅ |
  | get_tables_with_path_include_non_ready_true  | 25000      | 407.80 ms    | 399.59 ms   | -2.01% ✅ |
  | get_tables_with_path_include_non_ready_false | 100        | 557.35 µs    | 562.17 µs   | +0.86% ❌ |
  | get_tables_with_path_include_non_ready_false | 1000       | 277.44 µs    | 287.70 µs   | +3.70% ❌ |
  | get_tables_with_path_include_non_ready_false | 25000      | 55.345 µs    | 56.001 µs   | +1.19% ❌ |
  | get_tables_with_path_cold_cache              | -          | 69.834 ms    | 69.266 ms   | -0.81% ✅ |
  | get_tables_with_path_warm_cache              | -          | 5.450 µs     | 5.621 µs    | +3.14% ❌ |

  CacheStore Queue Benchmarks (Min of Runs)

  | Benchmark            | Parameters                  | Before (Min) | After (Min) | % Change  |
  |----------------------|-----------------------------|--------------|-------------|-----------|
  | queue_add            | 64 kb/512                   | 512.13 ms    | 512.74 ms   | +0.12% ❌  |
  | queue_add            | 256 kb/512                  | 268.82 ms    | 269.62 ms   | +0.30% ❌  |
  | queue_add            | 512 kb/512                  | 286.93 ms    | 293.60 ms   | +2.32% ❌  |
  | queue_list (Pending) | 5 queues, 128 kb            | 894.36 µs    | 888.31 µs   | -0.68% ✅  |
  | queue_list (Active)  | 5 queues, 128 kb            | 979.40 µs    | 907.67 µs   | -7.33% ✅  |
  | queue_get            | 5 queues, 128 kb, 10000/128 | 13.246 ms    | 11.025 ms   | -16.77% ✅ |

  Combined Results Summary

  | Category               | Benchmark Type                     | Min Before | Min After | % Change  |
  |------------------------|------------------------------------|------------|-----------|-----------|
  | Read-Heavy Operations  |                                    |            |           |           |
  | Metastore              | get_tables (non_ready_true) 100    | 17.734 ms  | 17.341 ms | -2.22% ✅  |
  | Metastore              | get_tables (non_ready_true) 1000   | 84.730 ms  | 81.181 ms | -4.19% ✅  |
  | Metastore              | get_tables (non_ready_true) 25000  | 407.80 ms  | 399.59 ms | -2.01% ✅  |
  | Metastore              | cold_cache                         | 69.834 ms  | 69.266 ms | -0.81% ✅  |
  | Queue                  | queue_get                          | 13.246 ms  | 11.025 ms | -16.77% ✅ |
  | Queue                  | queue_list (Active)                | 979.40 µs  | 907.67 µs | -7.33% ✅  |
  | Queue                  | queue_list (Pending)               | 894.36 µs  | 888.31 µs | -0.68% ✅  |
  | Write/Mixed Operations |                                    |            |           |           |
  | Metastore              | get_tables (non_ready_false) 100   | 557.35 µs  | 562.17 µs | +0.86% ❌  |
  | Metastore              | get_tables (non_ready_false) 1000  | 277.44 µs  | 287.70 µs | +3.70% ❌  |
  | Metastore              | get_tables (non_ready_false) 25000 | 55.345 µs  | 56.001 µs | +1.19% ❌  |
  | Metastore              | warm_cache                         | 5.450 µs   | 5.621 µs  | +3.14% ❌  |
  | Queue                  | queue_add 64kb                     | 512.13 ms  | 512.74 ms | +0.12% ❌  |
  | Queue                  | queue_add 256kb                    | 268.82 ms  | 269.62 ms | +0.30% ❌  |
  | Queue                  | queue_add 512kb                    | 286.93 ms  | 293.60 ms | +2.32% ❌  |

Author: ovr

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

@@ -518,14 +518,14 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
         let inserted_row = self.insert_row_kv(row_id, serialized_row)?;
 
         batch_pipe.add_event(MetaStoreEvent::Insert(Self::table_id(), row_id));
-        if self.snapshot().get(&inserted_row.key)?.is_some() {
+        if self.snapshot().get_pinned(&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)));
         }
         batch_pipe.batch().put(inserted_row.key, inserted_row.val);
 
         let index_row = self.insert_index_row(&row, row_id)?;
         for to_insert in index_row {
-            if self.snapshot().get(&to_insert.key)?.is_some() {
+            if self.snapshot().get_pinned(&to_insert.key)?.is_some() {
                 return Err(CubeError::internal(format!("Primary key constraint violation in secondary index. Primary key already exists for a row id {}: {:?}", row_id, &row)));
             }
             batch_pipe.batch().put(to_insert.key, to_insert.val);
@@ -573,15 +573,15 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
     fn migration_check_table(&self) -> Result<(), CubeError> {
         let snapshot = self.snapshot();
 
-        let table_info = snapshot.get(
+        let table_info = snapshot.get_pinned(
             &RowKey::TableInfo {
                 table_id: Self::table_id(),
             }
             .to_bytes(),
         )?;
 
         if let Some(table_info) = table_info {
-            let table_info = self.deserialize_table_info(table_info.as_slice())?;
+            let table_info = self.deserialize_table_info(&table_info)?;
 
             if table_info.version != Self::T::version()
                 || table_info.value_version != Self::T::value_version()
@@ -633,14 +633,14 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
     fn migration_check_indexes(&self) -> Result<(), CubeError> {
         let snapshot = self.snapshot();
         for index in Self::indexes().into_iter() {
-            let index_info = snapshot.get(
+            let index_info = snapshot.get_pinned(
                 &RowKey::SecondaryIndexInfo {
                     index_id: Self::index_id(index.get_id()),
                 }
                 .to_bytes(),
             )?;
             if let Some(index_info) = index_info {
-                let index_info = self.deserialize_index_info(index_info.as_slice())?;
+                let index_info = self.deserialize_index_info(&index_info)?;
                 if index_info.version != index.version()
                     || index_info.value_version != index.value_version()
                 {
@@ -977,7 +977,7 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
             RowKey::SecondaryIndex(Self::index_id(index_id), secondary_key_hash, row_id);
         let secondary_index_key = secondary_index_row_key.to_bytes();
 
-        if let Some(secondary_key_bytes) = self.db().get(&secondary_index_key)? {
+        if let Some(secondary_key_bytes) = self.db().get_pinned(&secondary_index_key)? {
             let index_value_version = RocksSecondaryIndex::value_version(secondary_index);
             let new_value = match RocksSecondaryIndexValue::from_bytes(
                 &secondary_key_bytes,
@@ -1102,10 +1102,10 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
 
     fn get_row(&self, row_id: u64) -> Result<Option<IdRow<Self::T>>, CubeError> {
         let ref db = self.snapshot();
-        let res = db.get(RowKey::Table(Self::table_id(), row_id).to_bytes())?;
+        let res = db.get_pinned(RowKey::Table(Self::table_id(), row_id).to_bytes())?;
 
         if let Some(buffer) = res {
-            let row = self.deserialize_id_row(row_id, buffer.as_slice())?;
+            let row = self.deserialize_id_row(row_id, &buffer)?;
             return Ok(Some(row));
         }