feat: split operations of Cluster and Shard, and serialize operations of Shard

catalog_impls/src/volatile.rs

@@ -20,7 +20,7 @@ use catalog::{
     },
     Catalog, CatalogRef, CreateSchemaWithCause,
 };
-use cluster::shard_tables_cache::ShardTablesCache;
+use cluster::shard_set::ShardSet;
 use common_types::schema::SchemaName;
 use common_util::error::BoxError;
 use log::{debug, info};
@@ -32,15 +32,15 @@ use tokio::sync::Mutex;
 /// ManagerImpl manages multiple volatile catalogs.
 pub struct ManagerImpl {
     catalogs: HashMap<String, Arc<CatalogImpl>>,
-    shard_tables_cache: ShardTablesCache,
+    shard_set: ShardSet,
     meta_client: MetaClientRef,
 }
 
 impl ManagerImpl {
-    pub fn new(shard_tables_cache: ShardTablesCache, meta_client: MetaClientRef) -> Self {
+    pub fn new(shard_set: ShardSet, meta_client: MetaClientRef) -> Self {
         let mut manager = ManagerImpl {
             catalogs: HashMap::new(),
-            shard_tables_cache,
+            shard_set,
             meta_client,
         };
 
@@ -87,7 +87,7 @@ impl ManagerImpl {
         let catalog = Arc::new(CatalogImpl {
             name: catalog_name.clone(),
             schemas: RwLock::new(HashMap::new()),
-            shard_tables_cache: self.shard_tables_cache.clone(),
+            shard_set: self.shard_set.clone(),
             meta_client: self.meta_client.clone(),
         });
 
@@ -107,7 +107,7 @@ struct CatalogImpl {
     name: String,
     /// All the schemas belonging to the catalog.
     schemas: RwLock<HashMap<SchemaName, SchemaRef>>,
-    shard_tables_cache: ShardTablesCache,
+    shard_set: ShardSet,
     meta_client: MetaClientRef,
 }
 
@@ -156,7 +156,7 @@ impl Catalog for CatalogImpl {
             self.name.to_string(),
             name.to_string(),
             SchemaId::from_u32(schema_id),
-            self.shard_tables_cache.clone(),
+            self.shard_set.clone(),
         ));
 
         schemas.insert(name.to_string(), schema);
@@ -188,7 +188,7 @@ struct SchemaImpl {
     /// Schema name
     schema_name: String,
     schema_id: SchemaId,
-    shard_tables_cache: ShardTablesCache,
+    shard_set: ShardSet,
     /// Tables of schema
     tables: RwLock<HashMap<String, TableRef>>,
     /// Guard for creating/dropping table
@@ -200,13 +200,13 @@ impl SchemaImpl {
         catalog_name: String,
         schema_name: String,
         schema_id: SchemaId,
-        shard_tables_cache: ShardTablesCache,
+        shard_set: ShardSet,
     ) -> Self {
         Self {
             catalog_name,
             schema_name,
             schema_id,
-            shard_tables_cache,
+            shard_set,
             tables: Default::default(),
             create_table_mutex: Mutex::new(()),
         }
@@ -303,17 +303,20 @@ impl Schema for SchemaImpl {
         // Do real create table.
         // Partition table is not stored in ShardTableManager.
         if request.partition_info.is_none() {
-            let _ = self
-                .shard_tables_cache
-                .find_table_by_name(
-                    &request.catalog_name,
-                    &request.schema_name,
-                    &request.table_name,
-                )
+            let shard =
+                self.shard_set
+                    .get(request.shard_id)
+                    .with_context(|| schema::CreateTable {
+                        request: request.clone(),
+                        msg: "shard not found".to_string(),
+                    })?;
+
+            // TODO: seems unnecessary?
+            let _ = shard
+                .find_table(&request.schema_name, &request.table_name)
                 .with_context(|| schema::CreateTable {
                     request: request.clone(),
-                    msg: format!("table with shards is not found in the ShardTableManager, catalog_name:{}, schema_name:{}, table_name:{}",
-                                 request.catalog_name, request.schema_name, request.table_name),
+                    msg: "table not found in shard".to_string(),
                 })?;
         }
         let request = request.into_engine_create_request(None, self.schema_id);

cluster/Cargo.toml

@@ -12,6 +12,7 @@ workspace = true
 
 [dependencies]
 async-trait = { workspace = true }
+catalog = { workspace = true }
 ceresdbproto = { workspace = true }
 common_types = { workspace = true }
 common_util = { workspace = true }
@@ -24,3 +25,4 @@ serde_json = { workspace = true }
 snafu = { workspace = true }
 table_engine = { workspace = true }
 tokio = { workspace = true }
+wal = { workspace = true }