fix(wal): handle WAL deletion on region drop (#4400)

Add LogStore trait bound to RegionWorkerLoop and handle WAL deletion on region drop.

src/mito2/src/worker/handle_drop.rs

@@ -22,6 +22,7 @@ use futures::TryStreamExt;
 use object_store::util::join_path;
 use object_store::{EntryMode, ObjectStore};
 use snafu::ResultExt;
+use store_api::logstore::LogStore;
 use store_api::region_request::AffectedRows;
 use store_api::storage::RegionId;
 use tokio::time::sleep;
@@ -34,7 +35,10 @@ use crate::worker::{RegionWorkerLoop, DROPPING_MARKER_FILE};
 const GC_TASK_INTERVAL_SEC: u64 = 5 * 60; // 5 minutes
 const MAX_RETRY_TIMES: u64 = 288; // 24 hours (5m * 288)
 
-impl<S> RegionWorkerLoop<S> {
+impl<S> RegionWorkerLoop<S>
+where
+    S: LogStore,
+{
     pub(crate) async fn handle_drop_request(
         &mut self,
         region_id: RegionId,
@@ -58,7 +62,7 @@ impl<S> RegionWorkerLoop<S> {
                 error!(e; "Failed to write the drop marker file for region {}", region_id);
 
                 // Sets the state back to writable. It's possible that the marker file has been written.
-                // We sets the state back to writable so we can retry the drop operation.
+                // We set the state back to writable so we can retry the drop operation.
                 region.switch_state_to_writable(RegionState::Dropping);
             })?;
 
@@ -66,6 +70,15 @@ impl<S> RegionWorkerLoop<S> {
         // Removes this region from region map to prevent other requests from accessing this region
         self.regions.remove_region(region_id);
         self.dropping_regions.insert_region(region.clone());
+
+        // Delete region data in WAL.
+        self.wal
+            .obsolete(
+                region_id,
+                region.version_control.current().last_entry_id,
+                &region.provider,
+            )
+            .await?;
         // Notifies flush scheduler.
         self.flush_scheduler.on_region_dropped(region_id);
         // Notifies compaction scheduler.