mito2/worker/
handle_compaction.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
15use api::v1::region::compact_request;
16use common_telemetry::{debug, error, info};
17use store_api::logstore::LogStore;
18use store_api::region_request::RegionCompactRequest;
19use store_api::storage::RegionId;
20
21use crate::compaction::{CompactionPickFinished, CompactionTransition};
22use crate::config::IndexBuildMode;
23use crate::error::{RegionNotFoundSnafu, StaleCompactionExecutionSnafu};
24use crate::metrics::COMPACTION_REQUEST_COUNT;
25use crate::region::MitoRegionRef;
26use crate::request::{
27    BuildIndexRequest, CompactionCancelled, CompactionFailed, CompactionFinished, OnFailure,
28    OptionOutputTx,
29};
30use crate::sst::index::IndexBuildType;
31use crate::worker::RegionWorkerLoop;
32
33impl<S> RegionWorkerLoop<S> {
34    pub(crate) async fn handle_compaction_pick_finished(
35        &mut self,
36        region_id: RegionId,
37        request: CompactionPickFinished,
38    ) where
39        S: LogStore,
40    {
41        let Some(region) = self.regions.get_region(region_id) else {
42            return;
43        };
44        // A terminal pick (canceled, no plan, or failed) may remove the
45        // compaction status and release DDLs fenced behind its picking phase.
46        // Such a pick never produces an execution callback, so the worker must
47        // execute the returned DDLs from this notification.
48        let transition = self
49            .compaction_scheduler
50            .handle_compaction_pick_finished(
51                request,
52                &region.manifest_ctx,
53                self.schema_metadata_manager.clone(),
54            )
55            .await;
56        match transition {
57            CompactionTransition::AutomaticFollowupScheduled => {
58                // There will be a followup compaction so we need to update the
59                // last schedule time to avoid frequent compactions.
60                region.update_schedule_compaction_millis();
61            }
62            CompactionTransition::NoAction => {}
63            CompactionTransition::DdlReady(mut pending_ddls) => {
64                if !pending_ddls.is_empty() {
65                    // Preserve the lifecycle order observed by listeners: compaction
66                    // termination is visible before its dependent DDLs are dispatched.
67                    self.listener.on_compaction_result_notified(region_id).await;
68                    self.handle_ddl_requests(&mut pending_ddls).await;
69                }
70            }
71        }
72    }
73
74    /// Handles compaction request submitted to region worker.
75    pub(crate) async fn handle_compaction_request(
76        &mut self,
77        region_id: RegionId,
78        req: RegionCompactRequest,
79        mut sender: OptionOutputTx,
80    ) {
81        let Some(region) = self.regions.writable_region_or(region_id, &mut sender) else {
82            return;
83        };
84        COMPACTION_REQUEST_COUNT.inc();
85        let parallelism = req.parallelism.unwrap_or(1) as usize;
86        match self.compaction_scheduler.schedule_manual_compaction(
87            req.options,
88            &region.version_control,
89            &region.access_layer,
90            sender,
91            &region.manifest_ctx,
92            self.schema_metadata_manager.clone(),
93            parallelism,
94            req.time_range,
95        ) {
96            // Ok(false) means the request was merged, queued or rejected; the
97            // waiter was already notified or will be completed by the queued
98            // cycle, so only a newly scheduled task is logged as success.
99            Ok(true) => info!(
100                "Successfully scheduled compaction task for region: {}",
101                region_id
102            ),
103            Ok(false) => {}
104            Err(e) => {
105                error!(e; "Failed to schedule compaction task for region: {}", region_id);
106            }
107        }
108    }
109
110    /// Handles compaction finished, update region version and manifest, deleted compacted files.
111    pub(crate) async fn handle_compaction_finished(
112        &mut self,
113        region_id: RegionId,
114        mut request: CompactionFinished,
115    ) where
116        S: LogStore,
117    {
118        let region = match self.regions.get_region(region_id) {
119            Some(region) => region,
120            None => {
121                request.on_failure(RegionNotFoundSnafu { region_id }.build());
122                return;
123            }
124        };
125        // Reject stale terminal results before applying their manifest edit.
126        if !self
127            .compaction_scheduler
128            .is_current_execution(region_id, &request.execution)
129        {
130            request.on_failure(StaleCompactionExecutionSnafu { region_id }.build());
131            return;
132        }
133        let execution = request.execution.clone();
134
135        region.version_control.apply_edit(
136            Some(request.edit.clone()),
137            &[],
138            region.file_purger.clone(),
139        );
140
141        let index_build_file_metas = std::mem::take(&mut request.edit.files_to_add);
142
143        // compaction finished.
144        request.on_success();
145        self.listener.on_compaction_result_notified(region_id).await;
146
147        // In async mode, create indexes after compact if new files are created.
148        if self.config.index.build_mode == IndexBuildMode::Async
149            && !index_build_file_metas.is_empty()
150        {
151            self.handle_rebuild_index(
152                BuildIndexRequest {
153                    region_id,
154                    build_type: IndexBuildType::Compact,
155                    file_metas: index_build_file_metas,
156                },
157                OptionOutputTx::new(None),
158            )
159            .await;
160        }
161
162        // Schedule next compaction if necessary.
163        let transition = self
164            .compaction_scheduler
165            .on_execution_finished(
166                region_id,
167                &execution,
168                &region.manifest_ctx,
169                self.schema_metadata_manager.clone(),
170            )
171            .await;
172        match transition {
173            CompactionTransition::AutomaticFollowupScheduled => {
174                region.update_schedule_compaction_millis();
175            }
176            CompactionTransition::NoAction => {}
177            CompactionTransition::DdlReady(mut pending_ddls) => {
178                self.handle_ddl_requests(&mut pending_ddls).await;
179            }
180        }
181    }
182
183    pub(crate) async fn handle_compaction_cancelled(
184        &mut self,
185        region_id: RegionId,
186        request: CompactionCancelled,
187    ) where
188        S: LogStore,
189    {
190        let execution = request.execution.clone();
191        let is_current = self.regions.get_region(region_id).is_some_and(|_| {
192            self.compaction_scheduler
193                .is_current_execution(region_id, &execution)
194        });
195        request.on_success();
196
197        if !is_current {
198            return;
199        }
200
201        // Reuse the scheduler's finish path to wake pending DDLs after a cooperative stop.
202        let mut pending_ddls = self
203            .compaction_scheduler
204            .on_execution_cancelled(region_id, &execution)
205            .await;
206        if !pending_ddls.is_empty() {
207            self.listener.on_compaction_result_notified(region_id).await;
208        }
209
210        self.handle_ddl_requests(&mut pending_ddls).await;
211    }
212
213    /// When compaction fails, we simply log the error.
214    pub(crate) async fn handle_compaction_failure(&mut self, req: CompactionFailed) {
215        if self.regions.get_region(req.region_id).is_none() {
216            return;
217        }
218        if !self
219            .compaction_scheduler
220            .is_current_execution(req.region_id, &req.execution)
221        {
222            debug!(
223                "Ignores stale compaction failure for region {}: {:?}",
224                req.region_id, req.err
225            );
226            return;
227        }
228
229        error!(req.err; "Failed to compact region: {}", req.region_id);
230        self.compaction_scheduler
231            .on_execution_failed(req.region_id, &req.execution, req.err);
232    }
233
234    /// Schedule compaction for the region if necessary.
235    pub(crate) async fn schedule_compaction(&mut self, region: &MitoRegionRef) {
236        if region.is_staging() || region.is_enter_staging() {
237            info!(
238                "Region {} is staging or entering staging, skip compaction",
239                region.region_id
240            );
241            return;
242        }
243        let now = self.time_provider.current_time_millis();
244        if now - region.last_schedule_compaction_millis()
245            >= self.config.min_compaction_interval.as_millis() as i64
246        {
247            debug!(
248                "minimal compaction interval time {:?} has passed, scheduling next compaction",
249                self.config.min_compaction_interval
250            );
251            match self.compaction_scheduler.schedule_automatic_compaction(
252                compact_request::Options::Regular(Default::default()),
253                &region.version_control,
254                &region.access_layer,
255                &region.manifest_ctx,
256                self.schema_metadata_manager.clone(),
257            ) {
258                Ok(true) => region.update_schedule_compaction_millis(),
259                Ok(false) => {}
260                Err(e) => {
261                    error!(e; "Failed to schedule compaction for region: {}", region.region_id)
262                }
263            }
264        }
265    }
266}
mito2/worker/handle_compaction.rs

mito2/worker/
handle_compaction.rs
