feat: update physical table schema on alter logical tables (#3585)

* feat: update physical table schema on alter

* feat: alter logical table in sql path

* feat: invalidate cache step1

* feat: invalidate cache step2

* feat: invalidate cache step3

* feat: invalidate cache step4

* fix: failed ut

* fix: standalone cache invalidator

* feat: log the count of already finished

* feat: re-invalidate cache

* chore: by comment

* chore: Update src/common/meta/src/ddl/create_logical_tables.rs

---------

Co-authored-by: Yingwen <realevenyag@gmail.com>

src/cmd/src/frontend.rs

@@ -16,7 +16,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use async_trait::async_trait;
-use catalog::kvbackend::CachedMetaKvBackendBuilder;
+use catalog::kvbackend::{CachedMetaKvBackendBuilder, KvBackendCatalogManager};
 use clap::Parser;
 use client::client_manager::DatanodeClients;
 use common_meta::heartbeat::handler::parse_mailbox_message::ParseMailboxMessageHandler;
@@ -248,11 +248,12 @@ impl StartCommand {
             .build();
         let cached_meta_backend = Arc::new(cached_meta_backend);
 
+        let catalog_manager =
+            KvBackendCatalogManager::new(cached_meta_backend.clone(), cached_meta_backend.clone());
+
         let executor = HandlerGroupExecutor::new(vec![
             Arc::new(ParseMailboxMessageHandler),
-            Arc::new(InvalidateTableCacheHandler::new(
-                cached_meta_backend.clone(),
-            )),
+            Arc::new(InvalidateTableCacheHandler::new(catalog_manager.clone())),
         ]);
 
         let heartbeat_task = HeartbeatTask::new(
@@ -263,10 +264,11 @@ impl StartCommand {
 
         let mut instance = FrontendBuilder::new(
             cached_meta_backend.clone(),
+            catalog_manager.clone(),
             Arc::new(DatanodeClients::default()),
             meta_client,
         )
-        .with_cache_invalidator(cached_meta_backend)
+        .with_cache_invalidator(catalog_manager.clone())
         .with_plugin(plugins.clone())
         .with_heartbeat_task(heartbeat_task)
         .try_build()

src/cmd/src/standalone.rs

@@ -16,10 +16,11 @@ use std::sync::Arc;
 use std::{fs, path};
 
 use async_trait::async_trait;
+use catalog::kvbackend::KvBackendCatalogManager;
 use clap::Parser;
 use common_catalog::consts::MIN_USER_TABLE_ID;
 use common_config::{metadata_store_dir, KvBackendConfig};
-use common_meta::cache_invalidator::DummyCacheInvalidator;
+use common_meta::cache_invalidator::{CacheInvalidatorRef, DummyCacheInvalidator};
 use common_meta::datanode_manager::DatanodeManagerRef;
 use common_meta::ddl::table_meta::{TableMetadataAllocator, TableMetadataAllocatorRef};
 use common_meta::ddl::ProcedureExecutorRef;
@@ -399,6 +400,9 @@ impl StartCommand {
         .await
         .context(StartFrontendSnafu)?;
 
+        let catalog_manager =
+            KvBackendCatalogManager::new(kv_backend.clone(), Arc::new(DummyCacheInvalidator));
+
         let builder =
             DatanodeBuilder::new(dn_opts, fe_plugins.clone()).with_kv_backend(kv_backend.clone());
         let datanode = builder.build().await.context(StartDatanodeSnafu)?;
@@ -429,15 +433,22 @@ impl StartCommand {
             table_metadata_manager,
             procedure_manager.clone(),
             datanode_manager.clone(),
+            catalog_manager.clone(),
             table_meta_allocator,
         )
         .await?;
 
-        let mut frontend = FrontendBuilder::new(kv_backend, datanode_manager, ddl_task_executor)
-            .with_plugin(fe_plugins.clone())
-            .try_build()
-            .await
-            .context(StartFrontendSnafu)?;
+        let mut frontend = FrontendBuilder::new(
+            kv_backend,
+            catalog_manager.clone(),
+            datanode_manager,
+            ddl_task_executor,
+        )
+        .with_plugin(fe_plugins.clone())
+        .with_cache_invalidator(catalog_manager)
+        .try_build()
+        .await
+        .context(StartFrontendSnafu)?;
 
         let servers = Services::new(fe_opts.clone(), Arc::new(frontend.clone()), fe_plugins)
             .build()
@@ -459,13 +470,14 @@ impl StartCommand {
         table_metadata_manager: TableMetadataManagerRef,
         procedure_manager: ProcedureManagerRef,
         datanode_manager: DatanodeManagerRef,
+        cache_invalidator: CacheInvalidatorRef,
         table_meta_allocator: TableMetadataAllocatorRef,
     ) -> Result<ProcedureExecutorRef> {
         let procedure_executor: ProcedureExecutorRef = Arc::new(
             DdlManager::try_new(
                 procedure_manager,
                 datanode_manager,
-                Arc::new(DummyCacheInvalidator),
+                cache_invalidator,
                 table_metadata_manager,
                 table_meta_allocator,
                 Arc::new(MemoryRegionKeeper::default()),