mito2/worker/
handle_close.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Handling close request.
16
17use common_telemetry::info;
18use store_api::logstore::LogStore;
19use store_api::region_request::{RegionCloseRequest, RegionFlushReason, RegionFlushRequest};
20use store_api::storage::RegionId;
21
22use crate::request::{DdlRequest, OptionOutputTx, SenderDdlRequest};
23use crate::worker::RegionWorkerLoop;
24
25impl<S: LogStore> RegionWorkerLoop<S> {
26    pub(crate) async fn handle_close_request(
27        &mut self,
28        region_id: RegionId,
29        request: RegionCloseRequest,
30        sender: OptionOutputTx,
31    ) {
32        let Some(region) = self.regions.get_region(region_id) else {
33            sender.send(Ok(0));
34            return;
35        };
36
37        info!("Try to close region {}, worker: {}", region_id, self.id);
38
39        if region.is_staging_or_staged_edit() {
40            // A staging region cannot flush, so it leaves staging before the close-time flush
41            // below. The manifest lock lets a successful end of staging in progress finish first,
42            // after which the region is already out of staging.
43            let _manager = region.manifest_ctx.manifest_manager.write().await;
44            region.abort_staging();
45            // Edits queued behind a staged edit belong to the staging that has just ended.
46            self.reject_region_edit_queue_as_not_found(region_id);
47        }
48
49        // Memtables containing skip-WAL writes must be flushed before closing.
50        // Followers cannot flush; explicit flush-on-close and region WAL options
51        // retain their existing behavior.
52        let version = region.version();
53        if (request.flush_on_close || region.skip_wal() || version.memtables.has_skip_wal_writes())
54            && !version.memtables.is_empty()
55            && region.is_flushable()
56        {
57            info!("Region {} has pending data, waiting for flush", region_id);
58            if self.flush_scheduler.is_flush_requested(region_id) {
59                self.flush_scheduler
60                    .add_ddl_request_to_pending(SenderDdlRequest {
61                        region_id,
62                        sender,
63                        request: DdlRequest::Close(request),
64                    });
65                return;
66            }
67            self.handle_flush_request(
68                region_id,
69                RegionFlushRequest {
70                    reason: Some(RegionFlushReason::Closing),
71                    ..Default::default()
72                },
73                sender,
74            );
75            return;
76        }
77
78        // No close-time flush is required, or the region cannot flush.
79        self.remove_region(region_id).await;
80        info!("Region {} closed, worker: {}", region_id, self.id);
81        sender.send(Ok(0))
82    }
83
84    /// Remove a region and stop all related tasks.
85    pub(crate) async fn remove_region(&mut self, region_id: RegionId) {
86        let Some(region) = self.regions.remove_region(region_id) else {
87            return;
88        };
89        region.deactivate_region_manifest_size_metric();
90        region.stop().await;
91        self.fail_region_stalled_requests_as_not_found(&region_id);
92        self.reject_region_edit_queue_as_not_found(region_id);
93        // Clean flush status.
94        self.flush_scheduler.on_region_closed(region_id);
95        // Clean compaction status.
96        self.compaction_scheduler.on_region_closed(region_id);
97        // clean index build status.
98        self.index_build_scheduler.on_region_closed(region_id).await;
99        self.region_count.dec();
100
101        // Notify the region hook that the region has been closed. The region is
102        // fully stopped and unregistered, but its files/manifest are preserved.
103        // Runs inline; the hook contract requires it to be fast.
104        if let Some(hook) = region.manifest_ctx.hook() {
105            let metadata = region.metadata();
106            hook.on_region_closed(region_id, &metadata).await;
107        }
108    }
109}
mito2/worker/handle_close.rs

mito2/worker/
handle_close.rs
