From 70520b0649d01069963b4720cc5e0d8774b76b2a Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <ratuthomm@gmail.com>
Date: Wed, 1 Jul 2026 19:53:18 +0800
Subject: [PATCH] feat: validate table name match in undrop procedure
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Add a guard in `UndropTableProcedure::on_prepare()` to check that the
dropped table name matches the undrop task name, returning `TableNotFound`
on mismatch.  This prevents undropping a table by a different name when
only the table ID is known.

- `src/common/meta/src/ddl/undrop_table.rs` — add table-name validation
- `src/common/meta/src/ddl/tests/drop_table.rs` — add test for name mismatch

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
---
 src/common/meta/src/ddl/tests/drop_table.rs | 31 +++++++++++++++++++++
 src/common/meta/src/ddl/undrop_table.rs     |  6 ++++
 2 files changed, 37 insertions(+)

diff --git a/src/common/meta/src/ddl/tests/drop_table.rs b/src/common/meta/src/ddl/tests/drop_table.rs
index c79c0d8c5eb..4aa220b6d52 100644
--- a/src/common/meta/src/ddl/tests/drop_table.rs
+++ b/src/common/meta/src/ddl/tests/drop_table.rs
@@ -757,6 +757,37 @@ async fn test_undrop_table_fails_when_live_name_exists() {
     assert_matches!(err, Error::TableAlreadyExists { .. });
 }
 
+#[tokio::test]
+async fn test_undrop_table_fails_when_task_name_mismatches_table_id() {
+    let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
+    let mut ddl_context = new_ddl_context(node_manager);
+    ddl_context.soft_drop_enabled = true;
+    let table_id = 1024;
+    let task = test_create_table_task("foo", table_id);
+    ddl_context
+        .table_metadata_manager
+        .create_table_metadata(
+            task.table_info.clone(),
+            TableRouteValue::physical(vec![]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+    let mut drop_procedure = DropTableProcedure::new(
+        new_drop_table_task("foo", table_id, false),
+        ddl_context.clone(),
+    );
+    execute_procedure_until_done(&mut drop_procedure).await;
+
+    let mut procedure = UndropTableProcedure::new(
+        new_undrop_table_task("bar", table_id),
+        ddl_context,
+    );
+    let err = procedure.on_prepare().await.unwrap_err();
+
+    assert_eq!(err.status_code(), StatusCode::TableNotFound);
+}
+
 #[tokio::test]
 async fn test_purge_dropped_table_drops_regions_and_deletes_tombstone() {
     let (tx, mut rx) = mpsc::channel(8);
diff --git a/src/common/meta/src/ddl/undrop_table.rs b/src/common/meta/src/ddl/undrop_table.rs
index 356b36248b4..a7eafa7ffbd 100644
--- a/src/common/meta/src/ddl/undrop_table.rs
+++ b/src/common/meta/src/ddl/undrop_table.rs
@@ -94,6 +94,12 @@ impl UndropTableProcedure {
             .with_context(|| error::TableNotFoundSnafu {
                 table_name: table_ref.to_string(),
             })?;
+        ensure!(
+            dropped_table.table_name == self.data.task.table_name(),
+            error::TableNotFoundSnafu {
+                table_name: table_ref.to_string()
+            }
+        );
         self.data.table_name = Some(dropped_table.table_name.clone());
         self.data.table_route_value = Some(dropped_table.table_route_value.clone());
         self.data.region_wal_options = dropped_table.region_wal_options;