From 0a0b01bf77177eb00e3534e8f34d8de5d9a6e679 Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <mrsatangel@gmail.com>
Date: Wed, 26 Apr 2023 11:11:01 +0800
Subject: [PATCH] respect engine name in table regional value when
 deregistering tables

---
 src/catalog/src/remote/manager.rs | 27 ++++++++++++++++++++++-----
 1 file changed, 22 insertions(+), 5 deletions(-)

diff --git a/src/catalog/src/remote/manager.rs b/src/catalog/src/remote/manager.rs
index 3b73de37114c..a23d332be420 100644
--- a/src/catalog/src/remote/manager.rs
+++ b/src/catalog/src/remote/manager.rs
@@ -20,7 +20,7 @@ use std::sync::Arc;
 use async_stream::stream;
 use async_trait::async_trait;
 use common_catalog::consts::{MIN_USER_TABLE_ID, MITO_ENGINE};
-use common_telemetry::{debug, error, info};
+use common_telemetry::{debug, error, info, warn};
 use dashmap::DashMap;
 use futures::Stream;
 use futures_util::{StreamExt, TryStreamExt};
@@ -828,6 +828,25 @@ impl SchemaProvider for RemoteSchemaProvider {
 
     async fn deregister_table(&self, name: &str) -> Result<Option<TableRef>> {
         let table_key = self.build_regional_table_key(name).to_string();
+
+        let engine_opt = self
+            .backend
+            .get(table_key.as_bytes())
+            .await?
+            .map(|Kv(_, v)| {
+                let TableRegionalValue { engine_name, .. } =
+                    TableRegionalValue::parse(String::from_utf8_lossy(&v))
+                        .context(InvalidCatalogValueSnafu)?;
+                Ok(engine_name)
+            })
+            .transpose()?
+            .flatten();
+
+        let engine_name = engine_opt.as_deref().unwrap_or_else(|| {
+            warn!("Cannot find table engine name for {table_key}");
+            MITO_ENGINE
+        });
+
         self.backend.delete(table_key.as_bytes()).await?;
         debug!(
             "Successfully deleted catalog table entry, key: {}",
@@ -842,10 +861,8 @@ impl SchemaProvider for RemoteSchemaProvider {
         // deregistering table does not necessarily mean dropping the table
         let table = self
             .engine_manager
-            .engine(MITO_ENGINE)
-            .context(TableEngineNotFoundSnafu {
-                engine_name: MITO_ENGINE,
-            })?
+            .engine(engine_name)
+            .context(TableEngineNotFoundSnafu { engine_name })?
             .get_table(&EngineContext {}, &reference)
             .with_context(|_| OpenTableSnafu {
                 table_info: reference.to_string(),