feat: add open_table_on_shard and close_table_on_shard in meta_event_…

…service (#542) * feat: add open_table_on_shard and close_table_on_shard in meta_event_service * refactor by CR
apache · Jan 9, 2023 · 99ca8d0 · 99ca8d0
1 parent 8e53951
commit 99ca8d0
Show file tree

Hide file tree

Showing 5 changed files with 241 additions and 77 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -111,7 +111,7 @@ message_queue = { path = "components/message_queue" }
 
 [workspace.dependencies.ceresdbproto]
 git = "https://github.com/CeresDB/ceresdbproto.git"
-rev = "6e12241032e01d00e130b65118dfafc0794a3102"
+rev = "6cc1754ffc31ddfe0710972c94f8fe5acd61af98"
 
 [workspace.dependencies.datafusion]
 git = "https://github.com/CeresDB/arrow-datafusion.git"

diff --git a/cluster/src/cluster_impl.rs b/cluster/src/cluster_impl.rs
@@ -6,8 +6,12 @@ use std::{
 };
 
 use async_trait::async_trait;
-use ceresdbproto::meta_event::{
-    CloseShardRequest, CreateTableOnShardRequest, DropTableOnShardRequest, OpenShardRequest,
+use ceresdbproto::{
+    meta_event::{
+        CloseShardRequest, CloseTableOnShardRequest, CreateTableOnShardRequest,
+        DropTableOnShardRequest, OpenShardRequest, OpenTableOnShardRequest, UpdateShardInfo,
+    },
+    meta_service::TableInfo as TableInfoPb,
 };
 use common_util::runtime::{JoinHandle, Runtime};
 use log::{error, info, warn};
@@ -247,14 +251,34 @@ impl Inner {
     }
 
     fn create_table_on_shard(&self, req: &CreateTableOnShardRequest) -> Result<()> {
-        let update_shard_info = req.update_shard_info.clone().context(ShardNotFound {
-            msg: "update shard info is missing in CreateTableOnShardRequest",
+        self.insert_table_to_shard(req.update_shard_info.clone(), req.table_info.clone())
+    }
+
+    fn drop_table_on_shard(&self, req: &DropTableOnShardRequest) -> Result<()> {
+        self.remove_table_to_shard(req.update_shard_info.clone(), req.table_info.clone())
+    }
+
+    fn open_table_on_shard(&self, req: &OpenTableOnShardRequest) -> Result<()> {
+        self.insert_table_to_shard(req.update_shard_info.clone(), req.table_info.clone())
+    }
+
+    fn close_table_on_shard(&self, req: &CloseTableOnShardRequest) -> Result<()> {
+        self.remove_table_to_shard(req.update_shard_info.clone(), req.table_info.clone())
+    }
+
+    fn insert_table_to_shard(
+        &self,
+        update_shard_info: Option<UpdateShardInfo>,
+        table_info: Option<TableInfoPb>,
+    ) -> Result<()> {
+        let update_shard_info = update_shard_info.context(ShardNotFound {
+            msg: "update shard info is missing",
         })?;
         let curr_shard_info = update_shard_info.curr_shard_info.context(ShardNotFound {
-            msg: "current shard info is missing in UpdateShardInfo",
+            msg: "current shard info is missing",
         })?;
-        let table_info = req.table_info.clone().context(TableNotFound {
-            msg: "table info is missing in CreateTableOnShardRequest",
+        let table_info = table_info.context(TableNotFound {
+            msg: "table info is missing",
         })?;
 
         self.shard_tables_cache.try_insert_table_to_shard(
@@ -264,15 +288,19 @@ impl Inner {
         )
     }
 
-    fn drop_table_on_shard(&self, req: &DropTableOnShardRequest) -> Result<()> {
-        let update_shard_info = req.update_shard_info.clone().context(ShardNotFound {
-            msg: "update shard info is missing in DropTableOnShardRequest",
+    fn remove_table_to_shard(
+        &self,
+        update_shard_info: Option<UpdateShardInfo>,
+        table_info: Option<TableInfoPb>,
+    ) -> Result<()> {
+        let update_shard_info = update_shard_info.context(ShardNotFound {
+            msg: "update shard info is missing",
         })?;
         let curr_shard_info = update_shard_info.curr_shard_info.context(ShardNotFound {
-            msg: "current shard info is missing in UpdateShardInfo",
+            msg: "current shard info is missing",
         })?;
-        let table_info = req.table_info.clone().context(TableNotFound {
-            msg: "table info is missing in CreateTableOnShardRequest",
+        let table_info = table_info.context(TableNotFound {
+            msg: "table info is missing",
         })?;
 
         self.shard_tables_cache.try_remove_table_from_shard(
@@ -332,6 +360,14 @@ impl Cluster for ClusterImpl {
         self.inner.drop_table_on_shard(req)
     }
 
+    async fn open_table_on_shard(&self, req: &OpenTableOnShardRequest) -> Result<()> {
+        self.inner.open_table_on_shard(req)
+    }
+
+    async fn close_table_on_shard(&self, req: &CloseTableOnShardRequest) -> Result<()> {
+        self.inner.close_table_on_shard(req)
+    }
+
     async fn route_tables(&self, req: &RouteTablesRequest) -> Result<RouteTablesResponse> {
         self.inner.route_tables(req).await
     }

diff --git a/cluster/src/lib.rs b/cluster/src/lib.rs
@@ -12,7 +12,8 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use ceresdbproto::meta_event::{
-    CloseShardRequest, CreateTableOnShardRequest, DropTableOnShardRequest, OpenShardRequest,
+    CloseShardRequest, CloseTableOnShardRequest, CreateTableOnShardRequest,
+    DropTableOnShardRequest, OpenShardRequest, OpenTableOnShardRequest,
 };
 use common_types::schema::SchemaName;
 use common_util::define_result;
@@ -118,6 +119,8 @@ pub trait Cluster {
     async fn close_shard(&self, req: &CloseShardRequest) -> Result<TablesOfShard>;
     async fn create_table_on_shard(&self, req: &CreateTableOnShardRequest) -> Result<()>;
     async fn drop_table_on_shard(&self, req: &DropTableOnShardRequest) -> Result<()>;
+    async fn open_table_on_shard(&self, req: &OpenTableOnShardRequest) -> Result<()>;
+    async fn close_table_on_shard(&self, req: &CloseTableOnShardRequest) -> Result<()>;
     async fn route_tables(&self, req: &RouteTablesRequest) -> Result<RouteTablesResponse>;
     async fn fetch_nodes(&self) -> Result<ClusterNodesResp>;
 }