feat: validate table name match in undrop procedure

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>
This commit is contained in:
Lei, HUANG
2026-07-02 14:44:00 +08:00
parent d37a4b56b1
commit 70520b0649
2 changed files with 37 additions and 0 deletions
@@ -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);
+6
View File
@@ -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;