feat: update topic-region map when create and drop table

src/common/meta/src/ddl/drop_database/executor.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::any::Any;
+use std::collections::HashMap;
 
 use common_procedure::Status;
 use common_telemetry::info;
@@ -107,8 +108,29 @@ impl State for DropDatabaseExecutor {
             self.physical_table_id,
             self.physical_region_routes.clone(),
         );
+
+        // Deletes topic-region mapping if dropping physical table
+        let region_wal_options =
+            if let TableRouteValue::Physical(table_route_value) = &table_route_value {
+                ddl_ctx
+                    .table_metadata_manager
+                    .datanode_table_manager()
+                    .regions(self.physical_table_id, table_route_value)
+                    .await?
+                    .into_iter()
+                    .flat_map(|datanode_table_value| {
+                        datanode_table_value
+                            .region_info
+                            .region_wal_options
+                            .into_iter()
+                    })
+                    .collect::<HashMap<_, _>>()
+            } else {
+                HashMap::new()
+            };
+
         executor
-            .on_destroy_metadata(ddl_ctx, &table_route_value)
+            .on_destroy_metadata(ddl_ctx, &table_route_value, &region_wal_options)
             .await?;
         executor.invalidate_table_cache(ddl_ctx).await?;
         executor

src/common/meta/src/ddl/drop_table.rs

@@ -15,6 +15,8 @@
 pub(crate) mod executor;
 mod metadata;
 
+use std::collections::HashMap;
+
 use async_trait::async_trait;
 use common_error::ext::BoxedError;
 use common_procedure::error::{ExternalSnafu, FromJsonSnafu, ToJsonSnafu};
@@ -163,7 +165,11 @@ impl DropTableProcedure {
             self.data.physical_region_routes.clone(),
         );
         self.executor
-            .on_delete_metadata_tombstone(&self.context, table_route_value)
+            .on_delete_metadata_tombstone(
+                &self.context,
+                table_route_value,
+                &self.data.region_wal_options,
+            )
             .await?;
 
         self.dropping_regions.clear();
@@ -257,6 +263,8 @@ pub struct DropTableData {
     pub physical_region_routes: Vec<RegionRoute>,
     pub physical_table_id: Option<TableId>,
     #[serde(default)]
+    pub region_wal_options: HashMap<u32, String>,
+    #[serde(default)]
     pub allow_rollback: bool,
 }
 
@@ -268,6 +276,7 @@ impl DropTableData {
             task,
             physical_region_routes: vec![],
             physical_table_id: None,
+            region_wal_options: HashMap::new(),
             allow_rollback: false,
         }
     }

src/common/meta/src/ddl/drop_table/executor.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
+
 use api::v1::region::{
     region_request, DropRequest as PbDropRegionRequest, RegionRequest, RegionRequestHeader,
 };
@@ -124,9 +126,15 @@ impl DropTableExecutor {
         &self,
         ctx: &DdlContext,
         table_route_value: &TableRouteValue,
+        region_wal_options: &HashMap<u32, String>,
     ) -> Result<()> {
         ctx.table_metadata_manager
-            .delete_table_metadata_tombstone(self.table_id, &self.table, table_route_value)
+            .delete_table_metadata_tombstone(
+                self.table_id,
+                &self.table,
+                table_route_value,
+                region_wal_options,
+            )
             .await
     }
 
@@ -135,9 +143,15 @@ impl DropTableExecutor {
         &self,
         ctx: &DdlContext,
         table_route_value: &TableRouteValue,
+        region_wal_options: &HashMap<u32, String>,
     ) -> Result<()> {
         ctx.table_metadata_manager
-            .destroy_table_metadata(self.table_id, &self.table, table_route_value)
+            .destroy_table_metadata(
+                self.table_id,
+                &self.table,
+                table_route_value,
+                region_wal_options,
+            )
             .await?;
 
         let detecting_regions = if table_route_value.is_physical() {

src/common/meta/src/ddl/drop_table/metadata.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
+
 use common_catalog::format_full_table_name;
 use snafu::OptionExt;
 use store_api::metric_engine_consts::METRIC_ENGINE_NAME;
@@ -30,9 +32,6 @@ impl DropTableProcedure {
             .get_physical_table_route(task.table_id)
             .await?;
 
-        self.data.physical_region_routes = physical_table_route_value.region_routes;
-        self.data.physical_table_id = Some(physical_table_id);
-
         if physical_table_id == self.data.table_id() {
             let table_info_value = self
                 .context
@@ -47,9 +46,29 @@ impl DropTableProcedure {
 
             let engine = table_info_value.table_info.meta.engine;
             // rollback only if dropping the metric physical table fails
-            self.data.allow_rollback = engine.as_str() == METRIC_ENGINE_NAME
+            self.data.allow_rollback = engine.as_str() == METRIC_ENGINE_NAME;
+
+            // Deletes topic-region mapping if dropping physical table
+            let region_wal_options = self
+                .context
+                .table_metadata_manager
+                .datanode_table_manager()
+                .regions(physical_table_id, &physical_table_route_value)
+                .await?
+                .into_iter()
+                .flat_map(|datanode_table_value| {
+                    datanode_table_value
+                        .region_info
+                        .region_wal_options
+                        .into_iter()
+                })
+                .collect::<HashMap<_, _>>();
+            self.data.region_wal_options = region_wal_options;
         }
 
+        self.data.physical_region_routes = physical_table_route_value.region_routes;
+        self.data.physical_table_id = Some(physical_table_id);
+
         Ok(())
     }
 }