[VQueues] Increase successive-merges limit and remove wasteful partial merges

crates/partition-store/src/keys.rs

@@ -232,25 +232,15 @@ impl KeyKind {
     }
 
     // Rocksdb merge operator function (partial merge)
+    #[inline]
     pub fn partial_merge(
-        key: &[u8],
+        _key: &[u8],
         _unused: Option<&[u8]>,
-        operands: &MergeOperands,
+        _operands: &MergeOperands,
     ) -> Option<Vec<u8>> {
-        let mut kind_buf = key;
-        let kind = match KeyKind::deserialize(&mut kind_buf) {
-            Ok(kind) => kind,
-            Err(e) => {
-                error!("Cannot apply merge operator; {e}");
-                return None;
-            }
-        };
-        trace!(?kind, "partial merge");
-
-        match kind {
-            KeyKind::VQueueMeta => vqueue_meta_merge::partial_merge(key, operands),
-            _ => None,
-        }
+        // Currently, we have no partial merge operator for any key. Change this
+        // if/when this is needed.
+        None
     }
 }
 

crates/partition-store/src/partition_db.rs

@@ -555,7 +555,7 @@ impl CfConfigurator for RocksConfigurator<AllDataCf> {
             KeyKind::full_merge,
             KeyKind::partial_merge,
         );
-        cf_options.set_max_successive_merges(100);
+        cf_options.set_max_successive_merges(5000);
 
         cf_options.set_disable_auto_compactions(config.rocksdb.rocksdb_disable_auto_compactions());
         if let Some(compaction_period) = config.rocksdb.rocksdb_periodic_compaction_seconds() {

crates/partition-store/src/vqueue_table/metadata.rs

@@ -57,7 +57,6 @@ impl MetaKey {
     }
 }
 
-// todo: check if this is still needed
 impl From<&VQueueId> for MetaKey {
     #[inline]
     fn from(qid: &VQueueId) -> Self {
@@ -85,7 +84,8 @@ pub(crate) mod vqueue_meta_merge {
     use rocksdb::MergeOperands;
     use tracing::error;
 
-    use restate_storage_api::vqueue_table::metadata::{VQueueMeta, VQueueMetaUpdates};
+    use restate_storage_api::vqueue_table;
+    use restate_storage_api::vqueue_table::metadata::VQueueMeta;
 
     use crate::keys::DecodeTableKey;
 
@@ -117,43 +117,21 @@ pub(crate) mod vqueue_meta_merge {
             }
         };
 
+        let mut update = <vqueue_table::metadata::Update as bilrost::encoding::RawMessage>::empty();
         for op in operands {
-            let batch = match VQueueMetaUpdates::decode(op) {
-                Err(err) => {
-                    let key = MetaKey::deserialize_from(&mut key);
-                    error!(
-                        ?err,
-                        ?key,
-                        "[full merge] Failed to decode vqueue meta batched updates ({} bytes)",
-                        op.len(),
-                    );
-                    return None;
-                }
-                Ok(batch) => batch,
-            };
-            for update in batch.updates.iter() {
-                vqueue_meta.apply_update(update);
+            if let Err(err) = update.replace_from_slice(op) {
+                let key = MetaKey::deserialize_from(&mut key);
+                error!(
+                    ?err,
+                    ?key,
+                    "[full merge] Failed to decode vqueue meta update ({} bytes)",
+                    op.len(),
+                );
+                return None;
             }
+            vqueue_meta.apply_update(&update);
         }
-        Some(vqueue_meta.encode_to_vec())
-    }
 
-    pub fn partial_merge(mut _key: &[u8], operands: &MergeOperands) -> Option<Vec<u8>> {
-        let mut updates =
-            VQueueMetaUpdates::with_capacity(operands.len() * VQueueMetaUpdates::INLINED_UPDATES);
-        for op in operands {
-            let partial_updates = match VQueueMetaUpdates::decode(op) {
-                Err(err) => {
-                    error!(
-                        ?err,
-                        "[partial merge] Failed to decode vqueue meta batched updates"
-                    );
-                    return None;
-                }
-                Ok(u) => u,
-            };
-            updates.extend(partial_updates);
-        }
-        Some(updates.encode_to_vec())
+        Some(vqueue_meta.encode_contiguous().into_vec())
     }
 }

crates/partition-store/src/vqueue_table/mod.rs

@@ -34,7 +34,7 @@ use tracing::error;
 
 use restate_rocksdb::Priority;
 use restate_storage_api::StorageError;
-use restate_storage_api::vqueue_table::metadata::{VQueueMeta, VQueueMetaRef, VQueueMetaUpdates};
+use restate_storage_api::vqueue_table::metadata::{VQueueMeta, VQueueMetaRef};
 use restate_storage_api::vqueue_table::{
     EntryKey, EntryMetadata, EntryStatusHeader, EntryValue, LazyEntryStatus, ReadVQueueTable,
     ScanVQueueTable, Stage, Status, WriteVQueueTable, stats::EntryStatistics,
@@ -180,15 +180,11 @@ impl WriteVQueueTable for PartitionStoreTransaction<'_> {
         update: &restate_storage_api::vqueue_table::metadata::Update,
     ) {
         let key_buffer = MetaKey::from(qid).to_bytes();
-        let updates = VQueueMetaUpdates::new(update.clone());
-        let value_buf = {
-            let value_buf = self.cleared_value_buffer_mut(updates.encoded_len());
-            // unwrap is safe because we know the buffer is big enough.
-            updates.encode(value_buf).unwrap();
-            value_buf.split()
-        };
-
-        self.raw_merge_cf(KeyKind::VQueueMeta, key_buffer, value_buf);
+        self.raw_merge_cf(
+            KeyKind::VQueueMeta,
+            key_buffer,
+            update.encode_contiguous().into_vec(),
+        );
     }
 
     fn put_vqueue_inbox(

crates/storage-api/Cargo.toml

@@ -23,15 +23,14 @@ restate-util-bytecount = { workspace = true, features = ["bilrost"] }
 
 ahash = { workspace = true }
 anyhow = { workspace = true }
-bilrost = { workspace = true, features = ["smallvec"] }
+bilrost = { workspace = true }
 bytes = { workspace = true }
 bytestring = { workspace = true }
 derive_more = { workspace = true, features = ["from", "into"] }
 futures = { workspace = true }
 opentelemetry = { workspace = true }
 prost = { workspace = true }
 serde = { workspace = true }
-smallvec = { workspace = true, features = ["const_new"] }
 static_assertions = { workspace = true }
 strum = { workspace = true }
 thiserror = { workspace = true }

crates/storage-api/src/vqueue_table/metadata.rs

@@ -8,8 +8,6 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
-use smallvec::SmallVec;
-
 use restate_clock::time::MillisSinceEpoch;
 use restate_limiter::LimitKey;
 use restate_types::clock::UniqueTimestamp;
@@ -519,16 +517,6 @@ impl VQueueMeta {
     }
 }
 
-/// A collection of differential updates to the vqueue meta data structure.
-///
-/// Those updates can be applied to the storage layer via a merge operator and at the same
-/// time they can be accepted by the vqueue's cache to keep them in sync.
-#[derive(Clone, Default, Debug, bilrost::Message)]
-pub struct VQueueMetaUpdates {
-    #[bilrost(1)]
-    pub updates: SmallVec<[Update; VQueueMetaUpdates::INLINED_UPDATES]>,
-}
-
 #[derive(Debug, Clone, bilrost::Message)]
 pub struct MoveMetrics {
     /// Timestamp of the entry's previous stage transition.
@@ -552,42 +540,6 @@ pub struct MoveMetrics {
     pub blocked_on_invoker_throttling_ms: u32,
 }
 
-impl VQueueMetaUpdates {
-    pub const INLINED_UPDATES: usize = 1;
-
-    pub fn new(update: Update) -> Self {
-        let updates = smallvec::smallvec_inline![update];
-        Self { updates }
-    }
-
-    pub fn with_capacity(capacity: usize) -> Self {
-        Self {
-            updates: SmallVec::with_capacity(capacity),
-        }
-    }
-
-    #[inline(always)]
-    pub fn push(&mut self, ts: UniqueTimestamp, action: Action) {
-        self.updates.push(Update { ts, action });
-    }
-
-    pub fn len(&self) -> usize {
-        self.updates.len()
-    }
-
-    pub fn iter(&self) -> impl Iterator<Item = &Update> {
-        self.updates.iter()
-    }
-
-    pub fn is_empty(&self) -> bool {
-        self.updates.is_empty()
-    }
-
-    pub fn extend(&mut self, other: Self) {
-        self.updates.extend(other.updates);
-    }
-}
-
 #[derive(Debug, Clone, Default, bilrost::Oneof, bilrost::Message)]
 pub enum Action {
     #[default]