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        // Whether this execution reduced the SST file count. The scheduler keeps
135        // draining while compaction makes progress; a rewrite that did not reduce
136        // files (e.g. its output was split into more files than its input) must end
137        // the chain to avoid a no-progress loop.
138        let made_progress = request.edit.files_to_remove.len() > request.edit.files_to_add.len();
139
140        region.version_control.apply_edit(
141            Some(request.edit.clone()),
142            &[],
143            region.file_purger.clone(),
144        );
145
146        let index_build_file_metas = std::mem::take(&mut request.edit.files_to_add);
147
148        // compaction finished.
149        request.on_success();
150        self.listener.on_compaction_result_notified(region_id).await;
151
152        // In async mode, create indexes after compact if new files are created.
153        if self.config.index.build_mode == IndexBuildMode::Async
154            && !index_build_file_metas.is_empty()
155        {
156            self.handle_rebuild_index(
157                BuildIndexRequest {
158                    region_id,
159                    build_type: IndexBuildType::Compact,
160                    file_metas: index_build_file_metas,
161                },
162                OptionOutputTx::new(None),
163            )
164            .await;
165        }
166
167        // Schedule next compaction if necessary.
168        let transition = self
169            .compaction_scheduler
170            .on_execution_finished(
171                region_id,
172                &execution,
173                &region.manifest_ctx,
174                self.schema_metadata_manager.clone(),
175                made_progress,
176            )
177            .await;
178        match transition {
179            CompactionTransition::AutomaticFollowupScheduled => {
180                region.update_schedule_compaction_millis();
181            }
182            CompactionTransition::NoAction => {}
183            CompactionTransition::DdlReady(mut pending_ddls) => {
184                self.handle_ddl_requests(&mut pending_ddls).await;
185            }
186        }
187    }
188
189    pub(crate) async fn handle_compaction_cancelled(
190        &mut self,
191        region_id: RegionId,
192        request: CompactionCancelled,
193    ) where
194        S: LogStore,
195    {
196        let execution = request.execution.clone();
197        let is_current = self.regions.get_region(region_id).is_some_and(|_| {
198            self.compaction_scheduler
199                .is_current_execution(region_id, &execution)
200        });
201        request.on_success();
202
203        if !is_current {
204            return;
205        }
206
207        // Reuse the scheduler's finish path to wake pending DDLs after a cooperative stop.
208        let mut pending_ddls = self
209            .compaction_scheduler
210            .on_execution_cancelled(region_id, &execution)
211            .await;
212        if !pending_ddls.is_empty() {
213            self.listener.on_compaction_result_notified(region_id).await;
214        }
215
216        self.handle_ddl_requests(&mut pending_ddls).await;
217    }
218
219    /// When compaction fails, we simply log the error.
220    pub(crate) async fn handle_compaction_failure(&mut self, req: CompactionFailed) {
221        if self.regions.get_region(req.region_id).is_none() {
222            return;
223        }
224        if !self
225            .compaction_scheduler
226            .is_current_execution(req.region_id, &req.execution)
227        {
228            debug!(
229                "Ignores stale compaction failure for region {}: {:?}",
230                req.region_id, req.err
231            );
232            return;
233        }
234
235        error!(req.err; "Failed to compact region: {}", req.region_id);
236        self.compaction_scheduler
237            .on_execution_failed(req.region_id, &req.execution, req.err);
238    }
239
240    /// Schedule compaction for the region if necessary.
241    pub(crate) async fn schedule_compaction(&mut self, region: &MitoRegionRef) {
242        if region.is_staging() || region.is_enter_staging() {
243            info!(
244                "Region {} is staging or entering staging, skip compaction",
245                region.region_id
246            );
247            return;
248        }
249        let now = self.time_provider.current_time_millis();
250        if now - region.last_schedule_compaction_millis()
251            >= self.config.min_compaction_interval.as_millis() as i64
252        {
253            debug!(
254                "minimal compaction interval time {:?} has passed, scheduling next compaction",
255                self.config.min_compaction_interval
256            );
257            match self.compaction_scheduler.schedule_automatic_compaction(
258                compact_request::Options::Regular(Default::default()),
259                &region.version_control,
260                &region.access_layer,
261                &region.manifest_ctx,
262                self.schema_metadata_manager.clone(),
263            ) {
264                Ok(true) => region.update_schedule_compaction_millis(),
265                Ok(false) => {}
266                Err(e) => {
267                    error!(e; "Failed to schedule compaction for region: {}", region.region_id)
268                }
269            }
270        }
271    }
272}
mito2/worker/handle_compaction.rs

mito2/worker/
handle_compaction.rs
