diff --git a/Cargo.lock b/Cargo.lock
index a58712d54a..073e0fc405 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -11114,7 +11114,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "22505a5c94da8e3b7c2996394d1c933236c4d743e81a410bcca4e6989fc066a4"
 dependencies = [
  "bytes",
- "heck 0.4.1",
+ "heck 0.5.0",
  "itertools 0.12.1",
  "log",
  "multimap",
@@ -11134,8 +11134,8 @@ version = "0.14.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ac6c3320f9abac597dcbc668774ef006702672474aad53c6d596b62e487b40b1"
 dependencies = [
- "heck 0.4.1",
- "itertools 0.14.0",
+ "heck 0.5.0",
+ "itertools 0.10.5",
  "log",
  "multimap",
  "once_cell",
@@ -11183,7 +11183,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
 dependencies = [
  "anyhow",
- "itertools 0.14.0",
+ "itertools 0.10.5",
  "proc-macro2",
  "quote",
  "syn 2.0.117",
@@ -11196,7 +11196,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9120690fafc389a67ba3803df527d0ec9cbbc9cc45e4cc20b332996dfb672425"
 dependencies = [
  "anyhow",
- "itertools 0.14.0",
+ "itertools 0.10.5",
  "proc-macro2",
  "quote",
  "syn 2.0.117",
@@ -13483,7 +13483,7 @@ version = "0.8.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1961e2ef424c1424204d3a5d6975f934f56b6d50ff5732382d84ebf460e147f7"
 dependencies = [
- "heck 0.4.1",
+ "heck 0.5.0",
  "proc-macro2",
  "quote",
  "syn 2.0.117",
diff --git a/src/api/src/helper.rs b/src/api/src/helper.rs
index ed8d64b57a..e650ddfcf2 100644
--- a/src/api/src/helper.rs
+++ b/src/api/src/helper.rs
@@ -713,6 +713,8 @@ fn ddl_request_type(request: &DdlRequest) -> &'static str {
         Some(Expr::DropView(_)) => "ddl.drop_view",
         Some(Expr::AlterDatabase(_)) => "ddl.alter_database",
         Some(Expr::CommentOn(_)) => "ddl.comment_on",
+        Some(Expr::UndropTable(_)) => "ddl.undrop_table",
+        Some(Expr::PurgeDroppedTable(_)) => "ddl.purge_dropped_table",
         None => "ddl.empty",
     }
 }
diff --git a/src/common/meta/src/rpc/ddl.rs b/src/common/meta/src/rpc/ddl.rs
index d78c72b31a..00f91baff2 100644
--- a/src/common/meta/src/rpc/ddl.rs
+++ b/src/common/meta/src/rpc/ddl.rs
@@ -38,7 +38,7 @@ use api::v1::{
     AlterDatabaseExpr, AlterTableExpr, CommentObjectType as PbCommentObjectType, CommentOnExpr,
     CreateDatabaseExpr, CreateFlowExpr, CreateTableExpr, CreateViewExpr, DropDatabaseExpr,
     DropFlowExpr, DropTableExpr, DropViewExpr, EvalInterval, ExpireAfter, Option as PbOption,
-    QueryContext as PbQueryContext, TruncateTableExpr,
+    PurgeDroppedTableExpr, QueryContext as PbQueryContext, TruncateTableExpr, UndropTableExpr,
 };
 use base64::Engine as _;
 use base64::engine::general_purpose;
@@ -262,7 +262,7 @@ impl TryFrom<Task> for DdlTask {
                 Ok(DdlTask::UndropTable(undrop_table.try_into()?))
             }
             Task::PurgeDroppedTableTask(purge_dropped_table) => {
-                Ok(DdlTask::PurgeDroppedTable(purge_dropped_table.into()))
+                Ok(DdlTask::PurgeDroppedTable(purge_dropped_table.try_into()?))
             }
             Task::AlterTableTask(alter_table) => Ok(DdlTask::AlterTable(alter_table.try_into()?)),
             Task::TruncateTableTask(truncate_table) => {
@@ -693,11 +693,15 @@ impl TryFrom<PbUndropTableTask> for UndropTableTask {
     type Error = error::Error;
 
     fn try_from(pb: PbUndropTableTask) -> Result<Self> {
+        let undrop_table = pb.undrop_table.context(error::InvalidProtoMsgSnafu {
+            err_msg: "expected undrop table",
+        })?;
+
         Ok(Self {
-            catalog: pb.catalog_name,
-            schema: pb.schema_name,
-            table: pb.table_name,
-            table_id: pb
+            catalog: undrop_table.catalog_name,
+            schema: undrop_table.schema_name,
+            table: undrop_table.table_name,
+            table_id: undrop_table
                 .table_id
                 .context(error::InvalidProtoMsgSnafu {
                     err_msg: "expected table_id",
@@ -710,32 +714,44 @@ impl TryFrom<PbUndropTableTask> for UndropTableTask {
 impl From<UndropTableTask> for PbUndropTableTask {
     fn from(task: UndropTableTask) -> Self {
         Self {
-            catalog_name: task.catalog,
-            schema_name: task.schema,
-            table_name: task.table,
-            table_id: Some(api::v1::TableId { id: task.table_id }),
+            undrop_table: Some(UndropTableExpr {
+                catalog_name: task.catalog,
+                schema_name: task.schema,
+                table_name: task.table,
+                table_id: Some(api::v1::TableId { id: task.table_id }),
+            }),
         }
     }
 }
 
-impl From<PbPurgeDroppedTableTask> for PurgeDroppedTableTask {
-    fn from(pb: PbPurgeDroppedTableTask) -> Self {
-        Self {
-            catalog: pb.catalog_name,
-            schema: pb.schema_name,
-            table: pb.table_name,
-            table_id: pb.table_id.map(|table_id| table_id.id),
-        }
+impl TryFrom<PbPurgeDroppedTableTask> for PurgeDroppedTableTask {
+    type Error = error::Error;
+
+    fn try_from(pb: PbPurgeDroppedTableTask) -> Result<Self> {
+        let purge_dropped_table = pb
+            .purge_dropped_table
+            .context(error::InvalidProtoMsgSnafu {
+                err_msg: "expected purge dropped table",
+            })?;
+
+        Ok(Self {
+            catalog: purge_dropped_table.catalog_name,
+            schema: purge_dropped_table.schema_name,
+            table: purge_dropped_table.table_name,
+            table_id: purge_dropped_table.table_id.map(|table_id| table_id.id),
+        })
     }
 }
 
 impl From<PurgeDroppedTableTask> for PbPurgeDroppedTableTask {
     fn from(task: PurgeDroppedTableTask) -> Self {
         Self {
-            catalog_name: task.catalog,
-            schema_name: task.schema,
-            table_name: task.table,
-            table_id: task.table_id.map(|id| api::v1::TableId { id }),
+            purge_dropped_table: Some(PurgeDroppedTableExpr {
+                catalog_name: task.catalog,
+                schema_name: task.schema,
+                table_name: task.table,
+                table_id: task.table_id.map(|id| api::v1::TableId { id }),
+            }),
         }
     }
 }
@@ -1948,6 +1964,15 @@ mod tests {
         assert!(matches!(de, DdlTask::PurgeDroppedTable(task) if task == expected));
     }
 
+    #[test]
+    fn test_purge_dropped_table_task_requires_expr() {
+        let result = DdlTask::try_from(Task::PurgeDroppedTableTask(
+            PbPurgeDroppedTableTask::default(),
+        ));
+
+        assert!(result.is_err());
+    }
+
     #[test]
     fn test_sort_columns() {
         // construct RawSchema