feat(cubestore): Use separate column family for queue

rust/cubestore/cubestore/src/cachestore/cache_item.rs

@@ -110,12 +110,18 @@ impl<'a> BaseRocksTable for CacheItemRocksTable<'a> {
     }
 }
 
-rocks_table_new!(CacheItem, CacheItemRocksTable, TableId::CacheItems, {
-    vec![
-        Box::new(CacheItemRocksIndex::ByPath),
-        Box::new(CacheItemRocksIndex::ByPrefix),
-    ]
-});
+rocks_table_new!(
+    CacheItem,
+    CacheItemRocksTable,
+    TableId::CacheItems,
+    {
+        vec![
+            Box::new(CacheItemRocksIndex::ByPath),
+            Box::new(CacheItemRocksIndex::ByPrefix),
+        ]
+    },
+    rocksdb::DEFAULT_COLUMN_FAMILY_NAME
+);
 
 #[derive(Hash, Clone, Debug)]
 pub enum CacheItemIndexKey {

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

@@ -24,7 +24,7 @@ use crate::CubeError;
 use async_trait::async_trait;
 
 use futures_timer::Delay;
-use rocksdb::{BlockBasedOptions, Options, DB};
+use rocksdb::{BlockBasedOptions, ColumnFamilyDescriptor, Options, DB};
 
 use crate::cachestore::compaction::CompactionPreloadedState;
 use crate::cachestore::listener::RocksCacheStoreListener;
@@ -38,6 +38,8 @@ use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tokio::sync::broadcast::Sender;
 
+pub const CACHESTORE_QUEUE_COLUMN_FAMILY_NAME: &str = "queue";
+
 pub(crate) struct RocksCacheStoreDetails {}
 
 impl RocksCacheStoreDetails {
@@ -74,10 +76,8 @@ impl RocksStoreDetails for RocksCacheStoreDetails {
 
         let mut opts = Options::default();
         opts.create_if_missing(true);
+        opts.create_missing_column_families(true);
         opts.set_prefix_extractor(rocksdb::SliceTransform::create_fixed_prefix(13));
-        opts.set_compaction_filter_factory(compaction::MetaStoreCacheCompactionFactory::new(
-            compaction_state,
-        ));
         // Disable automatic compaction before migration, will be enabled later in after_migration
         opts.set_disable_auto_compactions(true);
 
@@ -90,7 +90,27 @@ impl RocksStoreDetails for RocksCacheStoreDetails {
 
         opts.set_block_based_table_factory(&block_opts);
 
-        DB::open(&opts, path)
+        let default_cf = {
+            let mut opts = Options::default();
+            opts.set_prefix_extractor(rocksdb::SliceTransform::create_fixed_prefix(13));
+            opts.set_compaction_filter_factory(compaction::MetaStoreCacheCompactionFactory::new(
+                compaction_state.clone(),
+            ));
+
+            ColumnFamilyDescriptor::new(rocksdb::DEFAULT_COLUMN_FAMILY_NAME, opts)
+        };
+
+        let queue_cf = {
+            let mut opts = Options::default();
+            opts.set_prefix_extractor(rocksdb::SliceTransform::create_fixed_prefix(13));
+            opts.set_compaction_filter_factory(compaction::MetaStoreCacheCompactionFactory::new(
+                compaction_state,
+            ));
+
+            ColumnFamilyDescriptor::new(CACHESTORE_QUEUE_COLUMN_FAMILY_NAME, opts)
+        };
+
+        DB::open_cf_descriptors(&opts, path, vec![default_cf, queue_cf])
             .map_err(|err| CubeError::internal(format!("DB::open error for cachestore: {}", err)))
     }
 

rust/cubestore/cubestore/src/cachestore/queue_item.rs

@@ -1,3 +1,4 @@
+use super::cache_rocksstore::CACHESTORE_QUEUE_COLUMN_FAMILY_NAME;
 use crate::metastore::{
     BaseRocksTable, IndexId, RocksEntity, RocksSecondaryIndex, RocksTable, TableId, TableInfo,
 };
@@ -364,13 +365,19 @@ impl<'a> BaseRocksTable for QueueItemRocksTable<'a> {
     }
 }
 
-rocks_table_new!(QueueItem, QueueItemRocksTable, TableId::QueueItems, {
-    vec![
-        Box::new(QueueItemRocksIndex::ByPath),
-        Box::new(QueueItemRocksIndex::ByPrefixAndStatus),
-        Box::new(QueueItemRocksIndex::ByPrefix),
-    ]
-});
+rocks_table_new!(
+    QueueItem,
+    QueueItemRocksTable,
+    TableId::QueueItems,
+    {
+        vec![
+            Box::new(QueueItemRocksIndex::ByPath),
+            Box::new(QueueItemRocksIndex::ByPrefixAndStatus),
+            Box::new(QueueItemRocksIndex::ByPrefix),
+        ]
+    },
+    CACHESTORE_QUEUE_COLUMN_FAMILY_NAME
+);
 
 #[derive(Hash, Clone, Debug)]
 pub enum QueueItemIndexKey {

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

@@ -1,3 +1,4 @@
+use super::cache_rocksstore::CACHESTORE_QUEUE_COLUMN_FAMILY_NAME;
 use crate::metastore::{
     BaseRocksTable, IndexId, RocksEntity, RocksSecondaryIndex, RocksTable, TableId, TableInfo,
 };
@@ -67,9 +68,13 @@ impl<'a> BaseRocksTable for QueueResultRocksTable<'a> {
     }
 }
 
-rocks_table_new!(QueueResult, QueueResultRocksTable, TableId::QueueResults, {
-    vec![Box::new(QueueResultRocksIndex::ByPath)]
-});
+rocks_table_new!(
+    QueueResult,
+    QueueResultRocksTable,
+    TableId::QueueResults,
+    { vec![Box::new(QueueResultRocksIndex::ByPath)] },
+    CACHESTORE_QUEUE_COLUMN_FAMILY_NAME
+);
 
 #[derive(Hash, Clone, Debug)]
 pub enum QueueResultIndexKey {

rust/cubestore/cubestore/src/metastore/chunks.rs

@@ -159,12 +159,18 @@ pub(crate) enum ChunkRocksIndex {
     ReplayHandleId = 2,
 }
 
-rocks_table_impl!(Chunk, ChunkRocksTable, TableId::Chunks, {
-    vec![
-        Box::new(ChunkRocksIndex::PartitionId),
-        Box::new(ChunkRocksIndex::ReplayHandleId),
-    ]
-});
+rocks_table_impl!(
+    Chunk,
+    ChunkRocksTable,
+    TableId::Chunks,
+    {
+        vec![
+            Box::new(ChunkRocksIndex::PartitionId),
+            Box::new(ChunkRocksIndex::ReplayHandleId),
+        ]
+    },
+    rocksdb::DEFAULT_COLUMN_FAMILY_NAME
+);
 
 base_rocks_secondary_index!(Chunk, ChunkRocksIndex);
 

rust/cubestore/cubestore/src/metastore/index.rs

@@ -80,13 +80,19 @@ pub(crate) enum IndexRocksIndex {
 
 crate::base_rocks_secondary_index!(Index, IndexRocksIndex);
 
-rocks_table_impl!(Index, IndexRocksTable, TableId::Indexes, {
-    vec![
-        Box::new(IndexRocksIndex::TableID),
-        Box::new(IndexRocksIndex::Name),
-        Box::new(IndexRocksIndex::MultiIndexId),
-    ]
-});
+rocks_table_impl!(
+    Index,
+    IndexRocksTable,
+    TableId::Indexes,
+    {
+        vec![
+            Box::new(IndexRocksIndex::TableID),
+            Box::new(IndexRocksIndex::Name),
+            Box::new(IndexRocksIndex::MultiIndexId),
+        ]
+    },
+    rocksdb::DEFAULT_COLUMN_FAMILY_NAME
+);
 
 #[derive(Hash, Clone, Debug)]
 pub enum IndexIndexKey {

rust/cubestore/cubestore/src/metastore/job.rs

@@ -118,12 +118,18 @@ pub enum JobRocksIndex {
 
 base_rocks_secondary_index!(Job, JobRocksIndex);
 
-rocks_table_impl!(Job, JobRocksTable, TableId::Jobs, {
-    vec![
-        Box::new(JobRocksIndex::RowReference),
-        Box::new(JobRocksIndex::ByShard),
-    ]
-});
+rocks_table_impl!(
+    Job,
+    JobRocksTable,
+    TableId::Jobs,
+    {
+        vec![
+            Box::new(JobRocksIndex::RowReference),
+            Box::new(JobRocksIndex::ByShard),
+        ]
+    },
+    rocksdb::DEFAULT_COLUMN_FAMILY_NAME
+);
 
 #[derive(Hash, Clone, Debug)]
 pub enum JobIndexKey {

rust/cubestore/cubestore/src/metastore/mod.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::{BlockBasedOptions, Env, MergeOperands, Options, DB};
+use rocksdb::{BlockBasedOptions, ColumnFamilyDescriptor, Env, MergeOperands, Options, DB};
 use serde::{Deserialize, Serialize};
 use std::hash::Hash;
 use std::{env, io::Cursor, sync::Arc};
@@ -1160,7 +1160,14 @@ impl RocksStoreDetails for RocksMetaStoreDetails {
 
         opts.set_block_based_table_factory(&block_opts);
 
-        DB::open(&opts, path)
+        let default_cf = {
+            let mut opts = Options::default();
+            opts.set_prefix_extractor(rocksdb::SliceTransform::create_fixed_prefix(13));
+
+            ColumnFamilyDescriptor::new(rocksdb::DEFAULT_COLUMN_FAMILY_NAME, opts)
+        };
+
+        DB::open_cf_descriptors(&opts, path, vec![default_cf])
             .map_err(|err| CubeError::internal(format!("DB::open error for metastore: {}", err)))
     }
 

rust/cubestore/cubestore/src/metastore/multi_index.rs

@@ -55,9 +55,13 @@ pub(crate) enum MultiIndexRocksIndex {
 
 crate::base_rocks_secondary_index!(MultiIndex, MultiIndexRocksIndex);
 
-rocks_table_impl!(MultiIndex, MultiIndexRocksTable, TableId::MultiIndexes, {
-    vec![Box::new(MultiIndexRocksIndex::ByName)]
-});
+rocks_table_impl!(
+    MultiIndex,
+    MultiIndexRocksTable,
+    TableId::MultiIndexes,
+    { vec![Box::new(MultiIndexRocksIndex::ByName)] },
+    rocksdb::DEFAULT_COLUMN_FAMILY_NAME
+);
 
 #[derive(Hash, Clone, Debug)]
 pub enum MultiIndexIndexKey {
@@ -210,7 +214,8 @@ rocks_table_impl!(
             Box::new(MultiPartitionRocksIndex::ByMultiIndexId),
             Box::new(MultiPartitionRocksIndex::ByParentId),
         ]
-    }
+    },
+    rocksdb::DEFAULT_COLUMN_FAMILY_NAME
 );
 
 #[derive(Hash, Clone, Debug)]

rust/cubestore/cubestore/src/metastore/partition.rs

@@ -199,15 +199,21 @@ pub(crate) enum PartitionRocksIndex {
     ParentPartitionId = 5,
 }
 
-rocks_table_impl!(Partition, PartitionRocksTable, TableId::Partitions, {
-    vec![
-        Box::new(PartitionRocksIndex::IndexId),
-        Box::new(PartitionRocksIndex::MultiPartitionId),
-        Box::new(PartitionRocksIndex::Active),
-        Box::new(PartitionRocksIndex::JustCreated),
-        Box::new(PartitionRocksIndex::ParentPartitionId),
-    ]
-});
+rocks_table_impl!(
+    Partition,
+    PartitionRocksTable,
+    TableId::Partitions,
+    {
+        vec![
+            Box::new(PartitionRocksIndex::IndexId),
+            Box::new(PartitionRocksIndex::MultiPartitionId),
+            Box::new(PartitionRocksIndex::Active),
+            Box::new(PartitionRocksIndex::JustCreated),
+            Box::new(PartitionRocksIndex::ParentPartitionId),
+        ]
+    },
+    rocksdb::DEFAULT_COLUMN_FAMILY_NAME
+);
 
 #[derive(Hash, Clone, Debug)]
 pub enum PartitionIndexKey {

rust/cubestore/cubestore/src/metastore/replay_handle.rs

@@ -324,7 +324,8 @@ rocks_table_impl!(
     ReplayHandle,
     ReplayHandleRocksTable,
     TableId::ReplayHandles,
-    { vec![Box::new(ReplayHandleRocksIndex::ByTableId),] }
+    { vec![Box::new(ReplayHandleRocksIndex::ByTableId),] },
+    rocksdb::DEFAULT_COLUMN_FAMILY_NAME
 );
 
 #[derive(Hash, Clone, Debug)]