feat(cubestore): Optimization of inmemory chunk operations for streaming (#6354)

Author: waralexrom

rust/cubestore/cubestore/src/cluster/message.rs

@@ -43,6 +43,9 @@ pub enum NetworkMessage {
     },
     FreeMemoryChunkResult(Result<(), CubeError>),
 
+    FreeDeletedMemoryChunks,
+    FreeDeletedMemoryChunksResult(Result<(), CubeError>),
+
     MetaStoreCall(MetaStoreRpcMethodCall),
     MetaStoreCallResult(MetaStoreRpcMethodResult),
 

rust/cubestore/cubestore/src/cluster/mod.rs

@@ -128,6 +128,7 @@ pub trait Cluster: DIService + Send + Sync {
     ) -> Result<(), CubeError>;
 
     async fn free_memory_chunk(&self, node_name: &str, chunk_id: u64) -> Result<(), CubeError>;
+    async fn free_deleted_memory_chunks(&self, node_name: &str) -> Result<(), CubeError>;
 
     fn job_result_listener(&self) -> JobResultListener;
 
@@ -435,6 +436,16 @@ impl Cluster for ClusterImpl {
         }
     }
 
+    async fn free_deleted_memory_chunks(&self, node_name: &str) -> Result<(), CubeError> {
+        let response = self
+            .send_or_process_locally(node_name, NetworkMessage::FreeDeletedMemoryChunks)
+            .await?;
+        match response {
+            NetworkMessage::FreeDeletedMemoryChunksResult(r) => r,
+            x => panic!("Unexpected result for add chunk: {:?}", x),
+        }
+    }
+
     fn job_result_listener(&self) -> JobResultListener {
         JobResultListener {
             receiver: self.meta_store_sender.subscribe(),
@@ -573,8 +584,21 @@ impl Cluster for ClusterImpl {
                 let res = chunk_store.free_memory_chunk(chunk_id).await;
                 NetworkMessage::FreeMemoryChunkResult(res)
             }
+            NetworkMessage::FreeDeletedMemoryChunks => {
+                let chunk_store = self
+                    .injector
+                    .upgrade()
+                    .unwrap()
+                    .get_service_typed::<dyn ChunkDataStore>()
+                    .await;
+                let res = chunk_store.free_deleted_memory_chunks().await;
+                NetworkMessage::FreeDeletedMemoryChunksResult(res)
+            }
             NetworkMessage::FreeMemoryChunkResult(_) => {
-                panic!("AddChunkResult sent to worker");
+                panic!("FreeMemoryChunkResult sent to worker");
+            }
+            NetworkMessage::FreeDeletedMemoryChunksResult(_) => {
+                panic!("FreeDeletedMemoryChunksResult sent to worker");
             }
             NetworkMessage::MetaStoreCall(_) | NetworkMessage::MetaStoreCallResult(_) => {
                 panic!("MetaStoreCall sent to worker");
@@ -890,6 +914,9 @@ impl JobRunner {
                 if let RowKey::Table(TableId::Partitions, partition_id) = job.row_reference() {
                     let compaction_service = self.compaction_service.clone();
                     let partition_id = *partition_id;
+                    log::warn!(
+                        "JobType::InMemoryChunksCompaction is deprecated and should not be used"
+                    );
                     Ok(cube_ext::spawn(async move {
                         compaction_service
                             .compact_in_memory_chunks(partition_id)
@@ -899,6 +926,19 @@ impl JobRunner {
                     Self::fail_job_row_key(job)
                 }
             }
+            JobType::NodeInMemoryChunksCompaction(_) => {
+                if let RowKey::Table(TableId::Tables, _) = job.row_reference() {
+                    let compaction_service = self.compaction_service.clone();
+                    let node_name = self.server_name.clone();
+                    Ok(cube_ext::spawn(async move {
+                        compaction_service
+                            .compact_node_in_memory_chunks(node_name)
+                            .await
+                    }))
+                } else {
+                    Self::fail_job_row_key(job)
+                }
+            }
             JobType::MultiPartitionSplit => {
                 if let RowKey::Table(TableId::MultiPartitions, id) = job.row_reference() {
                     let compaction_service = self.compaction_service.clone();

rust/cubestore/cubestore/src/config/mod.rs

@@ -361,6 +361,8 @@ pub trait ConfigObj: DIService {
 
     fn compaction_in_memory_chunks_ratio_check_threshold(&self) -> u64;
 
+    fn compaction_in_memory_chunks_schedule_period_secs(&self) -> u64;
+
     fn wal_split_threshold(&self) -> u64;
 
     fn select_worker_pool_size(&self) -> usize;
@@ -482,6 +484,7 @@ pub struct ConfigObjImpl {
     pub compaction_in_memory_chunks_count_threshold: usize,
     pub compaction_in_memory_chunks_ratio_threshold: u64,
     pub compaction_in_memory_chunks_ratio_check_threshold: u64,
+    pub compaction_in_memory_chunks_schedule_period_secs: u64,
     pub wal_split_threshold: u64,
     pub data_dir: PathBuf,
     pub dump_dir: Option<PathBuf>,
@@ -592,6 +595,10 @@ impl ConfigObj for ConfigObjImpl {
         self.compaction_in_memory_chunks_ratio_check_threshold
     }
 
+    fn compaction_in_memory_chunks_schedule_period_secs(&self) -> u64 {
+        self.compaction_in_memory_chunks_schedule_period_secs
+    }
+
     fn wal_split_threshold(&self) -> u64 {
         self.wal_split_threshold
     }
@@ -1002,6 +1009,10 @@ impl Config {
                     "CUBESTORE_IN_MEMORY_CHUNKS_RATIO_CHECK_THRESHOLD",
                     1000,
                 ),
+                compaction_in_memory_chunks_schedule_period_secs: env_parse(
+                    "CUBESTORE_IN_MEMORY_CHUNKS_SCHEDULE_PERIOD_SECS",
+                    5,
+                ),
                 store_provider: {
                     if let Ok(bucket_name) = env::var("CUBESTORE_S3_BUCKET") {
                         FileStoreProvider::S3 {
@@ -1195,6 +1206,7 @@ impl Config {
                 compaction_in_memory_chunks_count_threshold: 10,
                 compaction_in_memory_chunks_ratio_threshold: 3,
                 compaction_in_memory_chunks_ratio_check_threshold: 1000,
+                compaction_in_memory_chunks_schedule_period_secs: 5,
                 store_provider: FileStoreProvider::Filesystem {
                     remote_dir: Some(
                         env::current_dir()

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

@@ -20,6 +20,7 @@ pub enum JobType {
     FinishMultiSplit,
     RepartitionChunk,
     InMemoryChunksCompaction,
+    NodeInMemoryChunksCompaction(/*node*/ String),
 }
 
 fn get_job_type_index(j: &JobType) -> u32 {
@@ -33,6 +34,7 @@ fn get_job_type_index(j: &JobType) -> u32 {
         JobType::FinishMultiSplit => 7,
         JobType::RepartitionChunk => 8,
         JobType::InMemoryChunksCompaction => 9,
+        JobType::NodeInMemoryChunksCompaction(_) => 10,
     }
 }
 
@@ -154,7 +156,7 @@ impl RocksSecondaryIndex<Job, JobIndexKey> for JobRocksIndex {
                 buf.write_u32::<BigEndian>(get_job_type_index(job_type))
                     .unwrap();
                 match job_type {
-                    JobType::TableImportCSV(l) => {
+                    JobType::TableImportCSV(l) | JobType::NodeInMemoryChunksCompaction(l) => {
                         buf.write_u64::<BigEndian>(l.len() as u64).unwrap();
                         buf.write(l.as_bytes()).unwrap();
                     }