feat: support meta event service

catalog_impls/src/volatile.rs

@@ -22,7 +22,7 @@ use catalog::{
 };
 use common_types::schema::SchemaName;
 use log::{debug, info};
-use meta_client::MetaClientRef;
+use meta_client::{types::AllocSchemaIdRequest, MetaClientRef};
 use snafu::{ensure, ResultExt};
 use table_engine::table::{SchemaId, TableRef};
 use tokio::sync::Mutex;
@@ -127,7 +127,7 @@ impl Catalog for CatalogImpl {
 
         let schema_id = self
             .meta_client
-            .alloc_schema_id(cluster::AllocSchemaIdRequest {
+            .alloc_schema_id(AllocSchemaIdRequest {
                 name: name.to_string(),
             })
             .await
@@ -324,7 +324,7 @@ impl Schema for SchemaImpl {
 
         // Request CeresMeta to drop this table.
         self.meta_client
-            .drop_table(cluster::DropTableRequest {
+            .drop_table(meta_client::types::DropTableRequest {
                 schema_name: schema_name.to_string(),
                 name: table_name.to_string(),
                 id: table.id().as_u64(),

cluster/Cargo.toml

@@ -11,6 +11,7 @@ workspace = true
 analytic_engine = { workspace = true }
 async-trait = { workspace = true }
 catalog = { workspace = true }
+ceresdbproto = { workspace = true }
 common_types = { workspace = true }
 common_util = { workspace = true }
 log = { workspace = true }

cluster/src/cluster_impl.rs

@@ -6,6 +6,9 @@ use std::{
 };
 
 use async_trait::async_trait;
+use ceresdbproto::meta_event::{
+    CloseShardsRequest, CreateTableOnShardRequest, DropTableOnShardRequest, OpenShardsRequest,
+};
 use common_util::runtime::{JoinHandle, Runtime};
 use log::{error, info, warn};
 use meta_client::{
@@ -19,8 +22,8 @@ use tokio::{
 };
 
 use crate::{
-    config::ClusterConfig, table_manager::TableManager, topology::ClusterTopology, Cluster,
-    ClusterNodesNotFound, ClusterNodesResp, MetaClientFailure, Result,
+    config::ClusterConfig, table_manager::TableManager, topology::ClusterTopology, CloseShardsOpts,
+    Cluster, ClusterNodesNotFound, ClusterNodesResp, MetaClientFailure, OpenShardsOpts, Result,
 };
 
 /// ClusterImpl is an implementation of [`Cluster`] based [`MetaClient`].
@@ -206,6 +209,22 @@ impl Cluster for ClusterImpl {
         Ok(())
     }
 
+    async fn open_shards(&self, _req: &OpenShardsRequest, _opts: OpenShardsOpts) -> Result<()> {
+        todo!();
+    }
+
+    async fn close_shards(&self, _req: &CloseShardsRequest, _opts: CloseShardsOpts) -> Result<()> {
+        todo!();
+    }
+
+    async fn create_table_on_shard(&self, _req: &CreateTableOnShardRequest) -> Result<()> {
+        todo!();
+    }
+
+    async fn drop_table_on_shard(&self, _req: &DropTableOnShardRequest) -> Result<()> {
+        todo!();
+    }
+
     async fn route_tables(&self, req: &RouteTablesRequest) -> Result<RouteTablesResponse> {
         self.inner.route_tables(req).await
     }

cluster/src/lib.rs

@@ -11,12 +11,11 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
+use ceresdbproto::meta_event::{
+    CloseShardsRequest, CreateTableOnShardRequest, DropTableOnShardRequest, OpenShardsRequest,
+};
 use common_types::schema::SchemaName;
 use common_util::define_result;
-pub use meta_client::types::{
-    AllocSchemaIdRequest, AllocSchemaIdResponse, CreateTableRequest, CreateTableResponse,
-    DropTableRequest, GetShardTablesRequest,
-};
 use meta_client::types::{ClusterNodesRef, RouteTablesRequest, RouteTablesResponse, ShardId};
 use snafu::{Backtrace, Snafu};
 
@@ -78,11 +77,21 @@ pub struct ClusterNodesResp {
     pub cluster_nodes: ClusterNodesRef,
 }
 
+#[derive(Debug, Default)]
+pub struct OpenShardsOpts {}
+
+#[derive(Debug, Default)]
+pub struct CloseShardsOpts {}
+
 /// Cluster manages tables and shard infos in cluster mode.
 #[async_trait]
 pub trait Cluster {
     async fn start(&self) -> Result<()>;
     async fn stop(&self) -> Result<()>;
+    async fn open_shards(&self, req: &OpenShardsRequest, opts: OpenShardsOpts) -> Result<()>;
+    async fn close_shards(&self, req: &CloseShardsRequest, opts: CloseShardsOpts) -> Result<()>;
+    async fn create_table_on_shard(&self, req: &CreateTableOnShardRequest) -> Result<()>;
+    async fn drop_table_on_shard(&self, req: &DropTableOnShardRequest) -> Result<()>;
     async fn route_tables(&self, req: &RouteTablesRequest) -> Result<RouteTablesResponse>;
     async fn fetch_nodes(&self) -> Result<ClusterNodesResp>;
 }