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 the close request asks for a flush, or the region skips WAL,
40        // and has data in memtable and region is flushable (like, not in follower state),
41        // we should flush it before closing to ensure durability.
42        if (request.flush_on_close || region.skip_wal())
43            && !region
44                .version_control
45                .current()
46                .version
47                .memtables
48                .is_empty()
49            && region.is_flushable()
50        {
51            info!("Region {} has pending data, waiting for flush", region_id);
52            if self.flush_scheduler.is_flush_requested(region_id) {
53                self.flush_scheduler
54                    .add_ddl_request_to_pending(SenderDdlRequest {
55                        region_id,
56                        sender,
57                        request: DdlRequest::Close(request),
58                    });
59                return;
60            }
61            self.handle_flush_request(
62                region_id,
63                RegionFlushRequest {
64                    reason: Some(RegionFlushReason::Closing),
65                    ..Default::default()
66                },
67                sender,
68            );
69            return;
70        }
71
72        // WAL configured or memtable is empty, flush is not necessary.
73        self.remove_region(region_id).await;
74        info!("Region {} closed, worker: {}", region_id, self.id);
75        sender.send(Ok(0))
76    }
77
78    /// Remove a region and stop all related tasks.
79    pub(crate) async fn remove_region(&mut self, region_id: RegionId) {
80        let Some(region) = self.regions.remove_region(region_id) else {
81            return;
82        };
83        region.stop().await;
84        self.fail_region_stalled_requests_as_not_found(&region_id);
85        self.reject_region_edit_queue_as_not_found(region_id);
86        // Clean flush status.
87        self.flush_scheduler.on_region_closed(region_id);
88        // Clean compaction status.
89        self.compaction_scheduler.on_region_closed(region_id);
90        // clean index build status.
91        self.index_build_scheduler.on_region_closed(region_id).await;
92        self.region_count.dec();
93
94        // Notify the region hook that the region has been closed. The region is
95        // fully stopped and unregistered, but its files/manifest are preserved.
96        // Runs inline; the hook contract requires it to be fast.
97        if let Some(hook) = region.manifest_ctx.hook() {
98            let metadata = region.metadata();
99            hook.on_region_closed(region_id, &metadata).await;
100        }
101    }
102}
mito2/worker/handle_close.rs

mito2/worker/
handle_close.rs
