feat(cubestore): Upgrade RocksDB to 7.9.2 (0.20) (#6221)

Author: ovr

rust/cubestore/Cargo.lock

@@ -53,7 +53,7 @@ tar = "0.4.38"
 
 [dev-dependencies]
 criterion = { version = "0.4.0", features = ["async_tokio", "html_reports"] }
-rocksdb = { version = "0.16.0", default-features = false, features = ["bzip2"] }
+rocksdb = { version = "0.20.1", default-features = false, features = ["bzip2"] }
 
 [[bench]]
 name = "in_process"

rust/cubestore/cubestore-sql-tests/Cargo.toml

@@ -41,7 +41,7 @@ simple_logger = "1.7.0"
 async-trait = "0.1.36"
 actix-rt = "2.7.0"
 regex = "1.3.9"
-rocksdb = { version = "0.16.0", default-features = false, features = ["bzip2"] }
+rocksdb = { version = "0.20.1", default-features = false, features = ["bzip2"] }
 uuid = { version = "0.8", features = ["serde", "v4"] }
 num = "0.3.0"
 enum_primitive = "0.1.1"

rust/cubestore/cubestore/Cargo.toml

@@ -24,7 +24,7 @@ use crate::CubeError;
 use async_trait::async_trait;
 
 use futures_timer::Delay;
-use rocksdb::{Options, DB};
+use rocksdb::{BlockBasedOptions, Options, DB};
 
 use crate::cachestore::compaction::CompactionPreloadedState;
 use crate::cachestore::listener::RocksCacheStoreListener;
@@ -81,6 +81,15 @@ impl RocksStoreDetails for RocksCacheStoreDetails {
         // Disable automatic compaction before migration, will be enabled later in after_migration
         opts.set_disable_auto_compactions(true);
 
+        let mut block_opts = BlockBasedOptions::default();
+        // https://github.com/facebook/rocksdb/blob/v7.9.2/include/rocksdb/table.h#L524
+        // RocksDB 7.x uses a new format = 5, but our previous version if RocksDB has some issues with it
+        // Let force the usage of old format for 1 month to save an ability to revert releases
+        // todo(ovr): Migrate to 5
+        block_opts.set_format_version(2);
+
+        opts.set_block_based_table_factory(&block_opts);
+
         DB::open(&opts, path)
             .map_err(|err| CubeError::internal(format!("DB::open error for cachestore: {}", err)))
     }

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

@@ -21,7 +21,7 @@ pub use rocks_table::*;
 use crate::cluster::node_name_by_partition;
 use async_trait::async_trait;
 use log::info;
-use rocksdb::{MergeOperands, Options, DB};
+use rocksdb::{BlockBasedOptions, Env, MergeOperands, Options, DB};
 use serde::{Deserialize, Serialize};
 use std::hash::Hash;
 use std::{env, io::Cursor, sync::Arc};
@@ -67,7 +67,7 @@ use parquet::basic::{ConvertedType, Repetition};
 use parquet::{basic::Type, schema::types};
 use partition::{PartitionRocksIndex, PartitionRocksTable};
 use regex::Regex;
-use rocksdb::backup::BackupEngineOptions;
+use rocksdb::backup::{BackupEngine, BackupEngineOptions};
 
 use schema::{SchemaRocksIndex, SchemaRocksTable};
 use smallvec::alloc::fmt::Formatter;
@@ -1127,15 +1127,17 @@ pub enum MetaStoreEvent {
 fn meta_store_merge(
     _new_key: &[u8],
     existing_val: Option<&[u8]>,
-    operands: &mut MergeOperands,
+    operands: &MergeOperands,
 ) -> Option<Vec<u8>> {
     let mut result: Vec<u8> = Vec::with_capacity(8);
     let mut counter = existing_val
         .map(|v| Cursor::new(v).read_u64::<BigEndian>().unwrap())
         .unwrap_or(0);
+
     for op in operands {
         counter += Cursor::new(op).read_u64::<BigEndian>().unwrap()
     }
+
     result.write_u64::<BigEndian>(counter).unwrap();
     Some(result)
 }
@@ -1149,6 +1151,15 @@ impl RocksStoreDetails for RocksMetaStoreDetails {
         opts.set_prefix_extractor(rocksdb::SliceTransform::create_fixed_prefix(13));
         opts.set_merge_operator_associative("meta_store merge", meta_store_merge);
 
+        let mut block_opts = BlockBasedOptions::default();
+        // https://github.com/facebook/rocksdb/blob/v7.9.2/include/rocksdb/table.h#L524
+        // RocksDB 7.x uses a new format = 5, but our previous version if RocksDB has some issues with it
+        // Let force the usage of old format for 1 month to save an ability to revert releases
+        // todo(ovr): Migrate to 5
+        block_opts.set_format_version(2);
+
+        opts.set_block_based_table_factory(&block_opts);
+
         DB::open(&opts, path)
             .map_err(|err| CubeError::internal(format!("DB::open error for metastore: {}", err)))
     }
@@ -3895,8 +3906,8 @@ impl MetaStore for RocksMetaStore {
 
     async fn debug_dump(&self, out_path: String) -> Result<(), CubeError> {
         self.read_operation(|db| {
-            let mut e =
-                rocksdb::backup::BackupEngine::open(&BackupEngineOptions::default(), out_path)?;
+            let opts = BackupEngineOptions::new(out_path)?;
+            let mut e = BackupEngine::open(&opts, &Env::new()?)?;
             Ok(e.create_new_backup_flush(db.db, true)?)
         })
         .await
@@ -4805,7 +4816,8 @@ mod tests {
             let iterator = meta_store.store.db.iterator(IteratorMode::Start);
 
             println!("Keys in db");
-            for (key, _) in iterator {
+            for kv_res in iterator {
+                let (key, _) = kv_res.unwrap();
                 println!("Key {:?}", RowKey::from_bytes(&key));
             }
 

rust/cubestore/cubestore/src/metastore/mod.rs

@@ -10,9 +10,9 @@ use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
 use datafusion::cube_ext;
 
 use log::{info, trace};
-use rocksdb::backup::BackupEngineOptions;
+use rocksdb::backup::{BackupEngine, BackupEngineOptions, RestoreOptions};
 use rocksdb::checkpoint::Checkpoint;
-use rocksdb::{Snapshot, WriteBatch, WriteBatchIterator, DB};
+use rocksdb::{Env, Snapshot, WriteBatch, WriteBatchIterator, DB};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::fmt::Debug;
@@ -610,13 +610,9 @@ impl RocksStore {
         details: Arc<dyn RocksStoreDetails>,
     ) -> Result<Arc<Self>, CubeError> {
         if !fs::metadata(path).await.is_ok() {
-            let mut backup =
-                rocksdb::backup::BackupEngine::open(&BackupEngineOptions::default(), dump_path)?;
-            backup.restore_from_latest_backup(
-                &path,
-                &path,
-                &rocksdb::backup::RestoreOptions::default(),
-            )?;
+            let opts = BackupEngineOptions::new(dump_path)?;
+            let mut backup = BackupEngine::open(&opts, &Env::new()?)?;
+            backup.restore_from_latest_backup(&path, &path, &RestoreOptions::default())?;
         } else {
             info!(
                 "Using existing {} in {}",
@@ -751,11 +747,12 @@ impl RocksStore {
             let mut serializer = WriteBatchContainer::new();
 
             let mut seq_numbers = Vec::new();
-
-            updates.into_iter().for_each(|(n, write_batch)| {
+            for update in updates.into_iter() {
+                let (n, write_batch) = update?;
                 seq_numbers.push(n);
                 write_batch.iterate(&mut serializer);
-            });
+            }
+
             (
                 serializer,
                 seq_numbers.iter().min().map(|v| *v),

rust/cubestore/cubestore/src/metastore/rocks_store.rs

@@ -246,7 +246,8 @@ where
 
     fn next(&mut self) -> Option<Self::Item> {
         let option = self.iter.next();
-        if let Some((key, value)) = option {
+        if let Some(res) = option {
+            let (key, value) = res.unwrap();
             if let RowKey::Table(table_id, row_id) = RowKey::from_bytes(&key) {
                 if table_id != self.table_id {
                     return None;
@@ -278,7 +279,8 @@ where
     fn next(&mut self) -> Option<Self::Item> {
         loop {
             let option = self.iter.next();
-            if let Some((key, value)) = option {
+            if let Some(res) = option {
+                let (key, value) = res.unwrap();
                 if let RowKey::SecondaryIndex(_, secondary_index_hash, row_id) =
                     RowKey::from_bytes(&key)
                 {
@@ -923,7 +925,8 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
         );
         let index = self.get_index_by_id(secondary_id);
 
-        for (key, value) in iter {
+        for kv_res in iter {
+            let (key, value) = kv_res?;
             if let RowKey::SecondaryIndex(_, secondary_index_hash, row_id) =
                 RowKey::from_bytes(&key)
             {
@@ -972,7 +975,8 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
             Direction::Forward,
         ));
 
-        for (key, _) in iter {
+        for kv_res in iter {
+            let (key, _) = kv_res?;
             let row_key = RowKey::from_bytes(&key);
             if let RowKey::Table(row_table_id, _) = row_key {
                 if row_table_id == table_id {
@@ -1000,7 +1004,8 @@ pub trait RocksTable: BaseRocksTable + Debug + Send + Sync {
             Direction::Forward,
         ));
 
-        for (key, _) in iter {
+        for kv_res in iter {
+            let (key, _) = kv_res?;
             let row_key = RowKey::from_bytes(&key);
             if let RowKey::SecondaryIndex(index_id, _, _) = row_key {
                 if index_id == Self::index_id(secondary_id) {