respect engine name in table regional value when deregistering tables

GreptimeTeam · Apr 26, 2023 · 0a0b01b · 0a0b01b
1 parent a270cac
commit 0a0b01b
Showing 1 changed file with 22 additions and 5 deletions.
diff --git 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(),