diff --git a/src/mito2/src/worker/handle_close.rs b/src/mito2/src/worker/handle_close.rs index a616a0e32dc..a37400956e9 100644 --- a/src/mito2/src/worker/handle_close.rs +++ b/src/mito2/src/worker/handle_close.rs @@ -17,9 +17,10 @@ use common_telemetry::info; use store_api::logstore::LogStore; use store_api::logstore::provider::Provider; -use store_api::region_request::{RegionCloseRequest, RegionFlushReason, RegionFlushRequest}; +use store_api::region_request::RegionCloseRequest; use store_api::storage::RegionId; +use crate::flush::FlushReason; use crate::request::{DdlRequest, OptionOutputTx, SenderDdlRequest}; use crate::worker::RegionWorkerLoop; @@ -57,14 +58,27 @@ impl RegionWorkerLoop { }); return; } - self.handle_flush_request( - region_id, - RegionFlushRequest { - reason: Some(RegionFlushReason::Closing), - ..Default::default() - }, - sender, - ); + let task = self.new_flush_task(®ion, FlushReason::Manual, None, self.config.clone()); + if let Err(err) = + self.flush_scheduler + .schedule_flush(region_id, ®ion.version_control, task) + { + sender.send(Err(err)); + return; + } + + if self.flush_scheduler.is_flush_requested(region_id) { + self.flush_scheduler + .add_ddl_request_to_pending(SenderDdlRequest { + region_id, + sender, + request: DdlRequest::Close(RegionCloseRequest {}), + }); + } else { + self.remove_region(region_id).await; + info!("Region {} closed, worker: {}", region_id, self.id); + sender.send(Ok(0)); + } return; }