mito2/worker/
handle_manifest.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//! Handles manifest.
16//!
17//! It updates the manifest and applies the changes to the region in background.
18
19use std::collections::{HashMap, VecDeque};
20use std::sync::Arc;
21
22use common_telemetry::{info, warn};
23use store_api::logstore::LogStore;
24use store_api::storage::RegionId;
25
26use crate::cache::file_cache::{FileType, IndexKey};
27use crate::cache::CacheManagerRef;
28use crate::error::{RegionBusySnafu, RegionNotFoundSnafu, Result};
29use crate::manifest::action::{
30    RegionChange, RegionEdit, RegionMetaAction, RegionMetaActionList, RegionTruncate,
31};
32use crate::metrics::WRITE_CACHE_INFLIGHT_DOWNLOAD;
33use crate::region::version::VersionBuilder;
34use crate::region::{MitoRegionRef, RegionLeaderState, RegionRoleState};
35use crate::request::{
36    BackgroundNotify, OptionOutputTx, RegionChangeResult, RegionEditRequest, RegionEditResult,
37    RegionSyncRequest, TruncateResult, WorkerRequest,
38};
39use crate::sst::location;
40use crate::worker::{RegionWorkerLoop, WorkerListener};
41
42pub(crate) type RegionEditQueues = HashMap<RegionId, RegionEditQueue>;
43
44/// A queue for temporary store region edit requests, if the region is in the "Editing" state.
45/// When the current region edit request is completed, the next (if there exists) request in the
46/// queue will be processed.
47/// Everything is done in the region worker loop.
48pub(crate) struct RegionEditQueue {
49    region_id: RegionId,
50    requests: VecDeque<RegionEditRequest>,
51}
52
53impl RegionEditQueue {
54    const QUEUE_MAX_LEN: usize = 128;
55
56    fn new(region_id: RegionId) -> Self {
57        Self {
58            region_id,
59            requests: VecDeque::new(),
60        }
61    }
62
63    fn enqueue(&mut self, request: RegionEditRequest) {
64        if self.requests.len() > Self::QUEUE_MAX_LEN {
65            let _ = request.tx.send(
66                RegionBusySnafu {
67                    region_id: self.region_id,
68                }
69                .fail(),
70            );
71            return;
72        };
73        self.requests.push_back(request);
74    }
75
76    fn dequeue(&mut self) -> Option<RegionEditRequest> {
77        self.requests.pop_front()
78    }
79}
80
81impl<S: LogStore> RegionWorkerLoop<S> {
82    /// Handles region change result.
83    pub(crate) async fn handle_manifest_region_change_result(
84        &mut self,
85        change_result: RegionChangeResult,
86    ) {
87        let region = match self.regions.get_region(change_result.region_id) {
88            Some(region) => region,
89            None => {
90                self.reject_region_stalled_requests(&change_result.region_id);
91                change_result.sender.send(
92                    RegionNotFoundSnafu {
93                        region_id: change_result.region_id,
94                    }
95                    .fail(),
96                );
97                return;
98            }
99        };
100
101        if change_result.result.is_ok() {
102            // Apply the metadata to region's version.
103            region
104                .version_control
105                .alter_schema(change_result.new_meta, &region.memtable_builder);
106
107            let version = region.version();
108            info!(
109                "Region {} is altered, metadata is {:?}, options: {:?}",
110                region.region_id, version.metadata, version.options,
111            );
112        }
113
114        // Sets the region as writable.
115        region.switch_state_to_writable(RegionLeaderState::Altering);
116        // Sends the result.
117        change_result.sender.send(change_result.result.map(|_| 0));
118
119        // Handles the stalled requests.
120        self.handle_region_stalled_requests(&change_result.region_id)
121            .await;
122    }
123
124    /// Handles region sync request.
125    ///
126    /// Updates the manifest to at least the given version.
127    /// **Note**: The installed version may be greater than the given version.
128    pub(crate) async fn handle_region_sync(&mut self, request: RegionSyncRequest) {
129        let region_id = request.region_id;
130        let sender = request.sender;
131        let region = match self.regions.follower_region(region_id) {
132            Ok(region) => region,
133            Err(e) => {
134                let _ = sender.send(Err(e));
135                return;
136            }
137        };
138
139        let original_manifest_version = region.manifest_ctx.manifest_version().await;
140        let manifest = match region
141            .manifest_ctx
142            .install_manifest_to(request.manifest_version)
143            .await
144        {
145            Ok(manifest) => manifest,
146            Err(e) => {
147                let _ = sender.send(Err(e));
148                return;
149            }
150        };
151        let version = region.version();
152        if !version.memtables.is_empty() {
153            warn!(
154                "Region {} memtables is not empty, which should not happen, manifest version: {}",
155                region.region_id, manifest.manifest_version
156            );
157        }
158        let region_options = version.options.clone();
159        let new_mutable = Arc::new(
160            region
161                .version()
162                .memtables
163                .mutable
164                .new_with_part_duration(version.compaction_time_window),
165        );
166        let metadata = manifest.metadata.clone();
167        let version = VersionBuilder::new(metadata, new_mutable)
168            .add_files(region.file_purger.clone(), manifest.files.values().cloned())
169            .flushed_entry_id(manifest.flushed_entry_id)
170            .flushed_sequence(manifest.flushed_sequence)
171            .truncated_entry_id(manifest.truncated_entry_id)
172            .compaction_time_window(manifest.compaction_time_window)
173            .options(region_options)
174            .build();
175        region.version_control.overwrite_current(Arc::new(version));
176
177        let updated = manifest.manifest_version > original_manifest_version;
178        let _ = sender.send(Ok((manifest.manifest_version, updated)));
179    }
180}
181
182impl<S> RegionWorkerLoop<S> {
183    /// Handles region edit request.
184    pub(crate) async fn handle_region_edit(&mut self, request: RegionEditRequest) {
185        let region_id = request.region_id;
186        let Some(region) = self.regions.get_region(region_id) else {
187            let _ = request.tx.send(RegionNotFoundSnafu { region_id }.fail());
188            return;
189        };
190
191        if !region.is_writable() {
192            if region.state() == RegionRoleState::Leader(RegionLeaderState::Editing) {
193                self.region_edit_queues
194                    .entry(region_id)
195                    .or_insert_with(|| RegionEditQueue::new(region_id))
196                    .enqueue(request);
197            } else {
198                let _ = request.tx.send(RegionBusySnafu { region_id }.fail());
199            }
200            return;
201        }
202
203        let RegionEditRequest {
204            region_id: _,
205            edit,
206            tx: sender,
207        } = request;
208
209        // Marks the region as editing.
210        if let Err(e) = region.set_editing() {
211            let _ = sender.send(Err(e));
212            return;
213        }
214
215        let request_sender = self.sender.clone();
216        let cache_manager = self.cache_manager.clone();
217        let listener = self.listener.clone();
218        // Now the region is in editing state.
219        // Updates manifest in background.
220        common_runtime::spawn_global(async move {
221            let result = edit_region(&region, edit.clone(), cache_manager, listener).await;
222            let notify = WorkerRequest::Background {
223                region_id,
224                notify: BackgroundNotify::RegionEdit(RegionEditResult {
225                    region_id,
226                    sender,
227                    edit,
228                    result,
229                }),
230            };
231            // We don't set state back as the worker loop is already exited.
232            if let Err(res) = request_sender.send(notify).await {
233                warn!(
234                    "Failed to send region edit result back to the worker, region_id: {}, res: {:?}",
235                    region_id, res
236                );
237            }
238        });
239    }
240
241    /// Handles region edit result.
242    pub(crate) async fn handle_region_edit_result(&mut self, edit_result: RegionEditResult) {
243        let region = match self.regions.get_region(edit_result.region_id) {
244            Some(region) => region,
245            None => {
246                let _ = edit_result.sender.send(
247                    RegionNotFoundSnafu {
248                        region_id: edit_result.region_id,
249                    }
250                    .fail(),
251                );
252                return;
253            }
254        };
255
256        let need_compaction =
257            edit_result.result.is_ok() && !edit_result.edit.files_to_add.is_empty();
258
259        if edit_result.result.is_ok() {
260            // Applies the edit to the region.
261            region
262                .version_control
263                .apply_edit(edit_result.edit, &[], region.file_purger.clone());
264        }
265
266        // Sets the region as writable.
267        region.switch_state_to_writable(RegionLeaderState::Editing);
268
269        let _ = edit_result.sender.send(edit_result.result);
270
271        if let Some(edit_queue) = self.region_edit_queues.get_mut(&edit_result.region_id) {
272            if let Some(request) = edit_queue.dequeue() {
273                self.handle_region_edit(request).await;
274            }
275        }
276
277        if need_compaction {
278            self.schedule_compaction(&region).await;
279        }
280    }
281
282    /// Writes truncate action to the manifest and then applies it to the region in background.
283    pub(crate) fn handle_manifest_truncate_action(
284        &self,
285        region: MitoRegionRef,
286        truncate: RegionTruncate,
287        sender: OptionOutputTx,
288    ) {
289        // Marks the region as truncating.
290        // This prevents the region from being accessed by other write requests.
291        if let Err(e) = region.set_truncating() {
292            sender.send(Err(e));
293            return;
294        }
295        // Now the region is in truncating state.
296
297        let request_sender = self.sender.clone();
298        let manifest_ctx = region.manifest_ctx.clone();
299
300        // Updates manifest in background.
301        common_runtime::spawn_global(async move {
302            // Write region truncated to manifest.
303            let action_list =
304                RegionMetaActionList::with_action(RegionMetaAction::Truncate(truncate.clone()));
305
306            let result = manifest_ctx
307                .update_manifest(RegionLeaderState::Truncating, action_list)
308                .await
309                .map(|_| ());
310
311            // Sends the result back to the request sender.
312            let truncate_result = TruncateResult {
313                region_id: truncate.region_id,
314                sender,
315                result,
316                truncated_entry_id: truncate.truncated_entry_id,
317                truncated_sequence: truncate.truncated_sequence,
318            };
319            let _ = request_sender
320                .send(WorkerRequest::Background {
321                    region_id: truncate.region_id,
322                    notify: BackgroundNotify::Truncate(truncate_result),
323                })
324                .await
325                .inspect_err(|_| warn!("failed to send truncate result"));
326        });
327    }
328
329    /// Writes region change action to the manifest and then applies it to the region in background.
330    pub(crate) fn handle_manifest_region_change(
331        &self,
332        region: MitoRegionRef,
333        change: RegionChange,
334        sender: OptionOutputTx,
335    ) {
336        // Marks the region as altering.
337        if let Err(e) = region.set_altering() {
338            sender.send(Err(e));
339            return;
340        }
341        let listener = self.listener.clone();
342        let request_sender = self.sender.clone();
343        // Now the region is in altering state.
344        common_runtime::spawn_global(async move {
345            let new_meta = change.metadata.clone();
346            let action_list = RegionMetaActionList::with_action(RegionMetaAction::Change(change));
347
348            let result = region
349                .manifest_ctx
350                .update_manifest(RegionLeaderState::Altering, action_list)
351                .await
352                .map(|_| ());
353            let notify = WorkerRequest::Background {
354                region_id: region.region_id,
355                notify: BackgroundNotify::RegionChange(RegionChangeResult {
356                    region_id: region.region_id,
357                    sender,
358                    result,
359                    new_meta,
360                }),
361            };
362            listener
363                .on_notify_region_change_result_begin(region.region_id)
364                .await;
365
366            if let Err(res) = request_sender.send(notify).await {
367                warn!(
368                    "Failed to send region change result back to the worker, region_id: {}, res: {:?}",
369                    region.region_id, res
370                );
371            }
372        });
373    }
374}
375
376/// Checks the edit, writes and applies it.
377async fn edit_region(
378    region: &MitoRegionRef,
379    edit: RegionEdit,
380    cache_manager: CacheManagerRef,
381    listener: WorkerListener,
382) -> Result<()> {
383    let region_id = region.region_id;
384    if let Some(write_cache) = cache_manager.write_cache() {
385        for file_meta in &edit.files_to_add {
386            let write_cache = write_cache.clone();
387            let layer = region.access_layer.clone();
388            let listener = listener.clone();
389
390            let index_key = IndexKey::new(region_id, file_meta.file_id, FileType::Parquet);
391            let remote_path = location::sst_file_path(layer.region_dir(), file_meta.file_id);
392
393            let is_index_exist = file_meta.exists_index();
394            let index_file_size = file_meta.index_file_size();
395
396            let index_file_index_key =
397                IndexKey::new(region_id, file_meta.file_id, FileType::Puffin);
398            let index_remote_path =
399                location::index_file_path(layer.region_dir(), file_meta.file_id);
400
401            let file_size = file_meta.file_size;
402            common_runtime::spawn_global(async move {
403                WRITE_CACHE_INFLIGHT_DOWNLOAD.add(1);
404
405                if write_cache
406                    .download(index_key, &remote_path, layer.object_store(), file_size)
407                    .await
408                    .is_ok()
409                {
410                    // Triggers the filling of the parquet metadata cache.
411                    // The parquet file is already downloaded.
412                    let _ = write_cache
413                        .file_cache()
414                        .get_parquet_meta_data(index_key)
415                        .await;
416
417                    listener.on_file_cache_filled(index_key.file_id);
418                }
419                if is_index_exist {
420                    // also download puffin file
421                    if let Err(err) = write_cache
422                        .download(
423                            index_file_index_key,
424                            &index_remote_path,
425                            layer.object_store(),
426                            index_file_size,
427                        )
428                        .await
429                    {
430                        common_telemetry::error!(
431                            err; "Failed to download puffin file, region_id: {}, index_file_index_key: {:?}, index_remote_path: {}", region_id, index_file_index_key, index_remote_path
432                        );
433                    }
434                }
435
436                WRITE_CACHE_INFLIGHT_DOWNLOAD.sub(1);
437            });
438        }
439    }
440
441    info!("Applying {edit:?} to region {}", region_id);
442
443    let action_list = RegionMetaActionList::with_action(RegionMetaAction::Edit(edit));
444    region
445        .manifest_ctx
446        .update_manifest(RegionLeaderState::Editing, action_list)
447        .await
448        .map(|_| ())
449}