feat: add close-first soft-drop table flow

Soft-drop now tombstones table metadata and closes datanode regions instead of
issuing physical drop requests, while preserving hard-drop cleanup semantics and
blocking conflicting drops of recreated table names.
Files:
- `src/common/meta/src/ddl.rs`
- `src/common/meta/src/ddl/drop_table.rs`
- `src/common/meta/src/ddl/drop_table/executor.rs`
- `src/common/meta/src/error.rs`
- `src/common/meta/src/ddl_manager.rs`
- `src/meta-srv/src/metasrv/builder.rs`
- `src/cmd/src/standalone.rs`
- `src/common/meta/src/test_util.rs`
- `src/meta-srv/src/procedure/utils.rs`
- `tests-integration/src/standalone.rs`
- `src/common/meta/src/ddl/tests/drop_table.rs`

Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>

src/cmd/src/standalone.rs

@@ -551,6 +551,7 @@ impl StartCommand {
             flow_metadata_manager: flow_metadata_manager.clone(),
             flow_metadata_allocator: flow_metadata_allocator.clone(),
             region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl),
+            soft_drop_enabled: false,
         };
 
         let ddl_manager = DdlManager::try_new(

src/common/meta/src/ddl.rs

@@ -117,6 +117,8 @@ pub struct DdlContext {
     pub flow_metadata_allocator: FlowMetadataAllocatorRef,
     /// controller of region failure detector.
     pub region_failure_detector_controller: RegionFailureDetectorControllerRef,
+    /// Whether table drops should stop after tombstoning metadata.
+    pub soft_drop_enabled: bool,
 }
 
 impl DdlContext {

src/common/meta/src/ddl/drop_table.rs

@@ -149,6 +149,7 @@ impl DropTableProcedure {
     /// Broadcasts invalidate table cache instruction.
     async fn on_broadcast(&mut self) -> Result<Status> {
         self.executor.invalidate_table_cache(&self.context).await?;
+
         self.data.state = DropTableState::DatanodeDropRegions;
 
         Ok(Status::executing(true))
@@ -172,6 +173,18 @@ impl DropTableProcedure {
                 .await?;
         }
 
+        if self.context.soft_drop_enabled {
+            self.executor
+                .on_close_regions(
+                    &self.context.node_manager,
+                    &self.context.leader_region_registry,
+                    &self.data.physical_region_routes,
+                )
+                .await?;
+            self.dropping_regions.clear();
+            return Ok(Status::done());
+        }
+
         self.executor
             .on_drop_regions(
                 &self.context.node_manager,
@@ -182,6 +195,7 @@ impl DropTableProcedure {
                 false,
             )
             .await?;
+
         self.data.state = DropTableState::DeleteTombstone;
         Ok(Status::executing(true))
     }

src/common/meta/src/ddl/drop_table/executor.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use api::v1::region::{
     CloseRequest as PbCloseRegionRequest, DropRequest as PbDropRegionRequest, RegionRequest,
@@ -82,6 +82,8 @@ impl DropTableExecutor {
     /// Checks whether table exists.
     /// - Early returns if table not exists and `drop_if_exists` is `true`.
     /// - Throws an error if table not exists and `drop_if_exists` is `false`.
+    /// - Rejects dropping a recreated live table while an older tombstone still owns the same
+    ///   fully qualified name.
     pub async fn on_prepare(&self, ctx: &DdlContext) -> Result<Control<()>> {
         let table_ref = self.table.table_ref();
 
@@ -106,6 +108,21 @@ impl DropTableExecutor {
             }
         );
 
+        if ctx.soft_drop_enabled
+            && let Some(dropped_table) = ctx
+                .table_metadata_manager
+                .get_dropped_table(&self.table)
+                .await?
+            && dropped_table.table_id != self.table_id
+        {
+            return error::TableNameTombstoneConflictSnafu {
+                table_name: table_ref.to_string(),
+                existing_table_id: dropped_table.table_id,
+                dropping_table_id: self.table_id,
+            }
+            .fail();
+        }
+
         Ok(Control::Continue(()))
     }
 
@@ -211,7 +228,17 @@ impl DropTableExecutor {
         Ok(())
     }
 
-    /// Drops region on datanode.
+    /// Drops regions on datanodes.
+    ///
+    /// Arguments:
+    /// - `node_manager`: resolves datanode clients from peers in `region_routes`.
+    /// - `leader_region_registry`: tracks in-flight leader region operations.
+    /// - `region_routes`: table region placement; leaders receive drop requests and followers
+    ///   receive close requests.
+    /// - `fast_path`: forwards to datanode drop requests to skip extra cleanup when safe.
+    /// - `force`: forwards to datanode drop requests to allow forced region removal.
+    /// - `partial_drop`: forwards to datanode drop requests for partial table/region drops.
+    #[allow(clippy::too_many_arguments)]
     pub async fn on_drop_regions(
         &self,
         node_manager: &NodeManagerRef,
@@ -319,6 +346,64 @@ impl DropTableExecutor {
 
         Ok(())
     }
+
+    /// Closes all table regions on datanodes without deleting region files or metadata tombstones.
+    pub async fn on_close_regions(
+        &self,
+        node_manager: &NodeManagerRef,
+        leader_region_registry: &LeaderRegionRegistryRef,
+        region_routes: &[RegionRoute],
+    ) -> Result<()> {
+        let table_id = self.table_id;
+        let mut seen_peer_ids = HashSet::new();
+        let peers = find_leaders(region_routes)
+            .into_iter()
+            .chain(find_followers(region_routes))
+            .filter(|peer| seen_peer_ids.insert(peer.id));
+        let mut close_region_tasks = Vec::new();
+
+        for datanode in peers {
+            let requester = node_manager.datanode(&datanode).await;
+            let region_ids = find_leader_regions(region_routes, &datanode)
+                .into_iter()
+                .chain(find_follower_regions(region_routes, &datanode))
+                .map(|region_number| RegionId::new(table_id, region_number));
+
+            for region_id in region_ids {
+                debug!("Closing region {region_id} on Datanode {datanode:?}");
+                let request = RegionRequest {
+                    header: Some(RegionRequestHeader {
+                        tracing_context: TracingContext::from_current_span().to_w3c(),
+                        ..Default::default()
+                    }),
+                    body: Some(region_request::Body::Close(PbCloseRegionRequest {
+                        region_id: region_id.as_u64(),
+                    })),
+                };
+
+                let datanode = datanode.clone();
+                let requester = requester.clone();
+                close_region_tasks.push(async move {
+                    if let Err(err) = requester.handle(request).await
+                        && err.status_code() != StatusCode::RegionNotFound
+                    {
+                        return Err(add_peer_context_if_needed(datanode)(err));
+                    }
+                    Ok(())
+                });
+            }
+        }
+
+        join_all(close_region_tasks)
+            .await
+            .into_iter()
+            .collect::<Result<Vec<_>>>()?;
+
+        let region_ids = operating_leader_regions(region_routes);
+        leader_region_registry.batch_delete(region_ids.into_iter().map(|(region_id, _)| region_id));
+
+        Ok(())
+    }
 }
 
 #[cfg(test)]

src/common/meta/src/ddl/tests/drop_table.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::assert_matches;
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 
@@ -30,6 +31,7 @@ use tokio::sync::mpsc;
 
 use crate::ddl::TableMetadata;
 use crate::ddl::create_logical_tables::CreateLogicalTablesProcedure;
+use crate::ddl::create_table::CreateTableProcedure;
 use crate::ddl::drop_table::{DropTableProcedure, DropTableState};
 use crate::ddl::test_util::create_table::test_create_table_task;
 use crate::ddl::test_util::datanode_handler::{DatanodeWatcher, NaiveDatanodeHandler};
@@ -37,6 +39,8 @@ use crate::ddl::test_util::{
     create_logical_table, create_physical_table, create_physical_table_metadata,
     put_datanode_address, test_create_logical_table_task, test_create_physical_table_task,
 };
+use crate::error::Error;
+use crate::key::table_name::TableNameKey;
 use crate::key::table_route::TableRouteValue;
 use crate::kv_backend::memory::MemoryKvBackend;
 use crate::peer::Peer;
@@ -231,6 +235,266 @@ async fn test_on_datanode_drop_regions_remaps_addresses_when_retrying() {
     assert_eq!(peers[1].addr, "new-follower");
 }
 
+#[tokio::test]
+async fn test_soft_drop_closes_regions_and_keeps_tombstone() {
+    let (tx, mut rx) = mpsc::channel(8);
+    let datanode_handler = DatanodeWatcher::new(tx);
+    let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler));
+    let mut ddl_context = new_ddl_context(node_manager);
+    ddl_context.soft_drop_enabled = true;
+    let table_id = 1024;
+    let table_name = "foo";
+    let task = test_create_table_task(table_name, table_id);
+    ddl_context
+        .table_metadata_manager
+        .create_table_metadata(
+            task.table_info.clone(),
+            TableRouteValue::physical(vec![
+                RegionRoute {
+                    region: Region::new_test(RegionId::new(table_id, 1)),
+                    leader_peer: Some(Peer::empty(1)),
+                    follower_peers: vec![Peer::empty(2)],
+                    leader_state: None,
+                    leader_down_since: None,
+                    write_route_policy: None,
+                },
+                RegionRoute {
+                    region: Region::new_test(RegionId::new(table_id, 2)),
+                    leader_peer: Some(Peer::empty(2)),
+                    follower_peers: vec![Peer::empty(1)],
+                    leader_state: None,
+                    leader_down_since: None,
+                    write_route_policy: None,
+                },
+            ]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+
+    let task = new_drop_table_task(table_name, table_id, false);
+    let mut procedure = DropTableProcedure::new(task, ddl_context.clone());
+
+    execute_procedure_until_done(&mut procedure).await;
+
+    assert!(procedure.dropping_regions.is_empty());
+    assert_eq!(ddl_context.memory_region_keeper.len(), 0);
+
+    let mut requests = Vec::new();
+    for _ in 0..4 {
+        let (peer, request) = rx.try_recv().unwrap();
+        let Some(region_request::Body::Close(req)) = request.body else {
+            unreachable!();
+        };
+        requests.push((peer.id, req.region_id));
+    }
+    requests.sort_unstable();
+    assert_eq!(
+        requests,
+        vec![
+            (1, RegionId::new(table_id, 1).as_u64()),
+            (1, RegionId::new(table_id, 2).as_u64()),
+            (2, RegionId::new(table_id, 1).as_u64()),
+            (2, RegionId::new(table_id, 2).as_u64()),
+        ]
+    );
+    assert!(rx.try_recv().is_err());
+
+    let table_name = procedure.data.task.table_name();
+    let live_table = ddl_context
+        .table_metadata_manager
+        .table_name_manager()
+        .get(TableNameKey::from(&table_name))
+        .await
+        .unwrap();
+    assert!(live_table.is_none());
+
+    let dropped_table = ddl_context
+        .table_metadata_manager
+        .get_dropped_table(&table_name)
+        .await
+        .unwrap();
+    assert!(dropped_table.is_some());
+}
+
+#[tokio::test]
+async fn test_hard_drop_keeps_delete_tombstone_flow() {
+    let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
+    let ddl_context = new_ddl_context(node_manager);
+    let table_id = 1024;
+    let table_name = "foo";
+    let task = test_create_table_task(table_name, table_id);
+    ddl_context
+        .table_metadata_manager
+        .create_table_metadata(
+            task.table_info.clone(),
+            TableRouteValue::physical(vec![RegionRoute {
+                region: Region::new_test(RegionId::new(table_id, 1)),
+                leader_peer: Some(Peer::empty(1)),
+                follower_peers: vec![],
+                leader_state: None,
+                leader_down_since: None,
+                write_route_policy: None,
+            }]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+
+    let task = new_drop_table_task(table_name, table_id, false);
+    let mut procedure = DropTableProcedure::new(task, ddl_context.clone());
+
+    execute_procedure_until(&mut procedure, |p| {
+        p.data.state == DropTableState::DeleteTombstone
+    })
+    .await;
+
+    assert_eq!(procedure.data.state, DropTableState::DeleteTombstone);
+
+    execute_procedure_until_done(&mut procedure).await;
+
+    let dropped_table = ddl_context
+        .table_metadata_manager
+        .get_dropped_table(&procedure.data.task.table_name())
+        .await
+        .unwrap();
+    assert!(dropped_table.is_none());
+}
+
+#[tokio::test]
+async fn test_create_table_succeeds_while_tombstone_exists() {
+    let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
+    let mut ddl_context = new_ddl_context(node_manager);
+    ddl_context.soft_drop_enabled = true;
+    let dropped_table_id = 1024;
+    let table_name = "foo";
+    let task = test_create_table_task(table_name, dropped_table_id);
+    ddl_context
+        .table_metadata_manager
+        .create_table_metadata(
+            task.table_info.clone(),
+            TableRouteValue::physical(vec![]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+
+    let drop_task = new_drop_table_task(table_name, dropped_table_id, false);
+    let mut drop_procedure = DropTableProcedure::new(drop_task, ddl_context.clone());
+    execute_procedure_until_done(&mut drop_procedure).await;
+
+    let mut create_task = test_create_table_task(table_name, 1025);
+    create_task.create_table.table_id = None;
+    create_task.table_info.ident.table_id = 0;
+    let mut create_procedure = CreateTableProcedure::new(create_task, ddl_context.clone()).unwrap();
+    execute_procedure_until_done(&mut create_procedure).await;
+
+    let live_table = ddl_context
+        .table_metadata_manager
+        .table_name_manager()
+        .get(TableNameKey::new(
+            DEFAULT_CATALOG_NAME,
+            DEFAULT_SCHEMA_NAME,
+            table_name,
+        ))
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(live_table.table_id(), create_procedure.table_id());
+
+    let dropped_table = ddl_context
+        .table_metadata_manager
+        .get_dropped_table(&create_procedure.data.task.table_name())
+        .await
+        .unwrap();
+    assert_eq!(dropped_table.unwrap().table_id, dropped_table_id);
+}
+
+#[tokio::test]
+async fn test_drop_recreated_table_fails_when_previous_tombstone_exists() {
+    let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
+    let mut ddl_context = new_ddl_context(node_manager);
+    ddl_context.soft_drop_enabled = true;
+    let original_table_id = 1024;
+    let recreated_table_id = 1025;
+    let table_name = "foo";
+    let task = test_create_table_task(table_name, original_table_id);
+    ddl_context
+        .table_metadata_manager
+        .create_table_metadata(
+            task.table_info.clone(),
+            TableRouteValue::physical(vec![]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+
+    let drop_task = new_drop_table_task(table_name, original_table_id, false);
+    let mut drop_procedure = DropTableProcedure::new(drop_task, ddl_context.clone());
+    execute_procedure_until_done(&mut drop_procedure).await;
+
+    ddl_context
+        .table_metadata_manager
+        .create_table_metadata(
+            test_create_table_task(table_name, recreated_table_id).table_info,
+            TableRouteValue::physical(vec![]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+
+    let mut procedure = DropTableProcedure::new(
+        new_drop_table_task(table_name, recreated_table_id, false),
+        ddl_context,
+    );
+    let err = procedure.on_prepare().await.unwrap_err();
+
+    assert_matches!(err, Error::TableNameTombstoneConflict { .. });
+}
+
+#[tokio::test]
+async fn test_hard_drop_recreated_table_ignores_previous_orphan_tombstone() {
+    let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
+    let ddl_context = new_ddl_context(node_manager);
+    let original_table_id = 1024;
+    let recreated_table_id = 1025;
+    let table_name = "foo";
+    let task = test_create_table_task(table_name, original_table_id);
+    ddl_context
+        .table_metadata_manager
+        .create_table_metadata(
+            task.table_info.clone(),
+            TableRouteValue::physical(vec![]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+
+    let drop_task = new_drop_table_task(table_name, original_table_id, false);
+    let mut drop_procedure = DropTableProcedure::new(drop_task, ddl_context.clone());
+    execute_procedure_until(&mut drop_procedure, |p| {
+        p.data.state == DropTableState::DeleteTombstone
+    })
+    .await;
+
+    ddl_context
+        .table_metadata_manager
+        .create_table_metadata(
+            test_create_table_task(table_name, recreated_table_id).table_info,
+            TableRouteValue::physical(vec![]),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+
+    let mut procedure = DropTableProcedure::new(
+        new_drop_table_task(table_name, recreated_table_id, false),
+        ddl_context,
+    );
+
+    procedure.on_prepare().await.unwrap();
+}
+
 #[tokio::test]
 async fn test_on_rollback() {
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));

src/common/meta/src/ddl_manager.rs

@@ -1238,6 +1238,7 @@ mod tests {
                 memory_region_keeper: Arc::new(MemoryRegionKeeper::default()),
                 leader_region_registry: Arc::new(LeaderRegionRegistry::default()),
                 region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl),
+                soft_drop_enabled: false,
             },
             procedure_manager.clone(),
             Arc::new(DummyRepartitionProcedureFactory),

src/common/meta/src/error.rs

@@ -379,6 +379,20 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display(
+        "Cannot drop table '{}': tombstoned table id {} already uses the same full name",
+        table_name,
+        existing_table_id
+    ))]
+    /// Raised when a live table is recreated with a name still reserved by an older tombstone.
+    TableNameTombstoneConflict {
+        table_name: String,
+        existing_table_id: TableId,
+        dropping_table_id: TableId,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("View already exists, view: {}", view_name))]
     ViewAlreadyExists {
         view_name: String,
@@ -1221,7 +1235,9 @@ impl ErrorExt for Error {
             ViewNotFound { .. } | TableNotFound { .. } | RegionNotFound { .. } => {
                 StatusCode::TableNotFound
             }
-            ViewAlreadyExists { .. } | TableAlreadyExists { .. } => StatusCode::TableAlreadyExists,
+            ViewAlreadyExists { .. }
+            | TableAlreadyExists { .. }
+            | TableNameTombstoneConflict { .. } => StatusCode::TableAlreadyExists,
 
             SubmitProcedure { source, .. }
             | QueryProcedure { source, .. }

src/common/meta/src/test_util.rs

@@ -206,6 +206,7 @@ pub fn new_ddl_context_with_kv_backend(
         flow_metadata_allocator,
         flow_metadata_manager,
         region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl),
+        soft_drop_enabled: false,
     }
 }
 

src/meta-srv/src/metasrv/builder.rs

@@ -425,6 +425,7 @@ impl MetasrvBuilder {
             flow_metadata_manager: flow_metadata_manager.clone(),
             flow_metadata_allocator: flow_metadata_allocator.clone(),
             region_failure_detector_controller,
+            soft_drop_enabled: ddl_soft_drop_enabled(&options),
         };
         let procedure_manager_c = procedure_manager.clone();
         let repartition_procedure_factory = Arc::new(DefaultRepartitionProcedureFactory::new(
@@ -673,6 +674,13 @@ fn build_procedure_manager(
     ))
 }
 
+/// Resolves if soft-drop is enabled from metasrv options.
+fn ddl_soft_drop_enabled(_options: &MetasrvOptions) -> bool {
+    // TODO(hl): add a dedicated soft-drop cluster config
+    // when wiring the user-facing option.
+    false
+}
+
 impl Default for MetasrvBuilder {
     fn default() -> Self {
         Self::new()

src/meta-srv/src/procedure/utils.rs

@@ -546,6 +546,7 @@ pub mod test_data {
             memory_region_keeper: Arc::new(MemoryRegionKeeper::new()),
             leader_region_registry: Arc::new(LeaderRegionRegistry::default()),
             region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl),
+            soft_drop_enabled: false,
         }
     }
 }

tests-integration/src/standalone.rs

@@ -250,6 +250,7 @@ impl GreptimeDbStandaloneBuilder {
                     flow_metadata_manager,
                     flow_metadata_allocator,
                     region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl),
+                    soft_drop_enabled: false,
                 },
                 procedure_manager.clone(),
                 repartition_procedure_factory,