fix review comments

Author: zhyass

src/meta/api/src/api_impl/schema_api.rs

@@ -21,7 +21,6 @@ use chrono::DateTime;
 use chrono::Utc;
 use databend_common_meta_app::app_error::AppError;
 use databend_common_meta_app::app_error::DropTableWithDropTime;
-use databend_common_meta_app::app_error::TableEngineMismatch;
 use databend_common_meta_app::app_error::UndropTableAlreadyExists;
 use databend_common_meta_app::app_error::UndropTableHasNoHistory;
 use databend_common_meta_app::app_error::UndropTableRetentionGuard;
@@ -178,16 +177,22 @@ pub async fn construct_drop_table_txn_operations(
     tenant: &Tenant,
     catalog_name: Option<String>,
     table_id: u64,
-    expected_engine: Option<&str>,
+    preloaded_table_meta: Option<SeqV<TableMeta>>,
     db_id: u64,
     if_exists: bool,
     if_delete: bool,
     txn: &mut TxnRequest,
 ) -> Result<(u64, u64), KVAppError> {
     let tbid = TableId { table_id };
 
+    // Reuse metadata already loaded by CREATE OR REPLACE; ordinary DROP callers
+    // do not have it and load it here. A preloaded value must belong to `table_id`.
+    let (tb_meta_seq, tb_meta) = if let Some(preloaded_table_meta) = preloaded_table_meta {
+        (preloaded_table_meta.seq, Some(preloaded_table_meta.data))
+    } else {
+        kv_api.get_pb_seq_and_value(&tbid).await?
+    };
     // Check if table exists.
-    let (tb_meta_seq, tb_meta) = kv_api.get_pb_seq_and_value(&tbid).await?;
     if tb_meta_seq == 0 {
         return Err(KVAppError::AppError(AppError::UnknownTableId(
             UnknownTableId::new(table_id, "drop_table_by_id failed to find valid tb_meta"),
@@ -235,13 +240,6 @@ pub async fn construct_drop_table_txn_operations(
     );
 
     let mut tb_meta = tb_meta.unwrap();
-    if let Some(expected_engine) = expected_engine {
-        if !tb_meta.engine.eq_ignore_ascii_case(expected_engine) {
-            return Err(KVAppError::AppError(
-                TableEngineMismatch::new(table_name, &tb_meta.engine, expected_engine).into(),
-            ));
-        }
-    }
     // drop a table with drop_on time
     if tb_meta.drop_on.is_some() {
         return if if_exists {

src/meta/api/src/api_impl/table_api.rs

@@ -362,54 +362,48 @@ where
                             });
                         }
                         CreateOption::CreateOrReplace => {
-                            if req.as_dropped {
-                                // CTAS does not call construct_drop_table_txn_operations(),
-                                // so validate its existing table here.
-                                let existing_meta =
-                                    self.get_pb(&TableId::new(*id.data)).await?.ok_or_else(|| {
-                                        KVAppError::AppError(AppError::UnknownTableId(
-                                            UnknownTableId::new(
-                                                *id.data,
-                                                "create or replace failed to find existing table meta",
-                                            ),
-                                        ))
-                                    })?;
-                                if !existing_meta
-                                    .engine
-                                    .eq_ignore_ascii_case(&req.table_meta.engine)
-                                {
-                                    return Err(KVAppError::AppError(
-                                        TableEngineMismatch::new(
-                                            req.table_name(),
-                                            &existing_meta.engine,
-                                            &req.table_meta.engine,
-                                        )
-                                        .into(),
-                                    ));
-                                }
-
-                                // Guard the engine check against a concurrent metadata update.
-                                txn.condition.push(txn_cond_seq(
-                                    &TableId::new(*id.data),
-                                    Eq,
-                                    existing_meta.seq,
+                            // CREATE OR REPLACE must not replace an existing table with a
+                            // different engine.
+                            let existing_table_meta = self
+                                .get_pb(&TableId::new(*id.data))
+                                .await?
+                                .ok_or_else(|| {
+                                KVAppError::AppError(AppError::UnknownTableId(UnknownTableId::new(
+                                    *id.data,
+                                    "create or replace failed to find existing table meta",
+                                )))
+                            })?;
+                            if !existing_table_meta
+                                .engine
+                                .eq_ignore_ascii_case(&req.table_meta.engine)
+                            {
+                                return Err(KVAppError::AppError(
+                                    TableEngineMismatch::new(
+                                        req.table_name(),
+                                        &existing_table_meta.engine,
+                                        &req.table_meta.engine,
+                                    )
+                                    .into(),
                                 ));
+                            }
+
+                            if req.as_dropped {
                                 // If the table is being created as a dropped table, we do not
                                 // need to combine with drop_table_txn operations, just return
                                 // the sequence number associated with the value part of
                                 // the key-value pair (key_dbid_tbname, table_id).
 
                                 SeqV::new(id.seq, *id.data)
                             } else {
-                                // The drop helper validates the engine against the metadata
-                                // sequence used by the replacement transaction.
+                                // Reuse the metadata loaded for the engine check when marking the
+                                // existing table as dropped.
                                 let (seq, id) = construct_drop_table_txn_operations(
                                     self,
                                     req.name_ident.table_name.clone(),
                                     &req.name_ident.tenant,
                                     req.catalog_name.clone(),
                                     *id.data,
-                                    Some(&req.table_meta.engine),
+                                    Some(existing_table_meta),
                                     *seq_db_id.data,
                                     true,
                                     false,