mito2/worker/
handle_copy_region.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 common_telemetry::{debug, error, info};
16use snafu::OptionExt;
17use store_api::region_engine::MitoCopyRegionFromResponse;
18use store_api::storage::{FileId, RegionId};
19
20use crate::error::{InvalidRequestSnafu, MissingManifestSnafu, Result};
21use crate::manifest::action::{RegionEdit, RegionMetaAction, RegionMetaActionList};
22use crate::region::{
23    FileDescriptor, MitoRegionRef, RegionFileCopier, RegionLeaderState, RegionMetadataLoader,
24};
25use crate::request::{
26    BackgroundNotify, CopyRegionFromFinished, CopyRegionFromRequest, WorkerRequest,
27};
28use crate::sst::file::FileMeta;
29use crate::sst::location::region_dir_from_table_dir;
30use crate::worker::{RegionWorkerLoop, WorkerRequestWithTime};
31
32impl<S> RegionWorkerLoop<S> {
33    pub(crate) fn handle_copy_region_from_request(&mut self, request: CopyRegionFromRequest) {
34        let region_id = request.region_id;
35        let source_region_id = request.source_region_id;
36        let sender = request.sender;
37        let region = match self.regions.writable_non_staging_region(region_id) {
38            Ok(region) => region,
39            Err(e) => {
40                let _ = sender.send(Err(e));
41                return;
42            }
43        };
44
45        let same_table = source_region_id.table_id() == region_id.table_id();
46        if !same_table {
47            let _ = sender.send(
48                InvalidRequestSnafu {
49                    region_id,
50                    reason: format!("Source and target regions must be from the same table, source_region_id: {source_region_id}, target_region_id: {region_id}"),
51                }
52                .fail(),
53            );
54            return;
55        }
56        if source_region_id == region_id {
57            let _ = sender.send(
58                InvalidRequestSnafu {
59                    region_id,
60                    reason: format!("Source and target regions must be different, source_region_id: {source_region_id}, target_region_id: {region_id}"),
61                }
62                .fail(),
63            );
64            return;
65        }
66
67        let region_metadata_loader = RegionMetadataLoader::new(
68            self.config.clone(),
69            self.object_store_manager.clone(),
70            self.node_lease.is_enabled(),
71        );
72        let worker_sender = self.sender.clone();
73
74        common_runtime::spawn_global(async move {
75            let (region_edit, source_file_ids) = match Self::copy_region_from(
76                &region,
77                region_metadata_loader,
78                source_region_id,
79                region_id,
80                request.parallelism.max(1),
81            )
82            .await
83            {
84                Ok(region_files) => region_files,
85                Err(e) => {
86                    let _ = sender.send(Err(e));
87                    return;
88                }
89            };
90
91            match region_edit {
92                Some(region_edit) => {
93                    if let Err(e) = worker_sender
94                        .send(WorkerRequestWithTime::new(WorkerRequest::Background {
95                            region_id,
96                            notify: BackgroundNotify::CopyRegionFromFinished(
97                                CopyRegionFromFinished {
98                                    region_id,
99                                    edit: region_edit,
100                                    sender,
101                                },
102                            ),
103                        }))
104                        .await
105                    {
106                        error!(e; "Failed to send copy region from finished notification to worker, region_id: {}", region_id);
107                    }
108                }
109                None => {
110                    let _ = sender.send(Ok(MitoCopyRegionFromResponse {
111                        copied_file_ids: source_file_ids,
112                    }));
113                }
114            }
115        });
116    }
117
118    pub(crate) fn handle_copy_region_from_finished(&mut self, request: CopyRegionFromFinished) {
119        let region_id = request.region_id;
120        let sender = request.sender;
121        let region = match self.regions.writable_region(region_id) {
122            Ok(region) => region,
123            Err(e) => {
124                let _ = sender.send(Err(e));
125                return;
126            }
127        };
128
129        let copied_file_ids = request
130            .edit
131            .files_to_add
132            .iter()
133            .map(|file_meta| file_meta.file_id)
134            .collect();
135
136        region
137            .version_control
138            .apply_edit(Some(request.edit), &[], region.file_purger.clone());
139
140        let _ = sender.send(Ok(MitoCopyRegionFromResponse { copied_file_ids }));
141    }
142
143    /// Returns the region edit and the file ids that were copied from the source region to the target region.
144    ///
145    /// If no need to copy files, returns (None, source_file_ids).
146    async fn copy_region_from(
147        region: &MitoRegionRef,
148        region_metadata_loader: RegionMetadataLoader,
149        source_region_id: RegionId,
150        target_region_id: RegionId,
151        parallelism: usize,
152    ) -> Result<(Option<RegionEdit>, Vec<FileId>)> {
153        let table_dir = region.table_dir();
154        let path_type = region.path_type();
155        let region_dir = region_dir_from_table_dir(table_dir, source_region_id, path_type);
156        info!(
157            "Loading source region manifest from region dir: {region_dir}, target region: {target_region_id}"
158        );
159        let source_region_manifest = region_metadata_loader
160            .load_manifest(&region_dir, &region.version().options)
161            .await?
162            .context(MissingManifestSnafu {
163                region_id: source_region_id,
164            })?;
165        let mut new_file_metas = vec![];
166        let target_region_manifest = region.manifest_ctx.manifest().await;
167        let source_file_ids = source_region_manifest
168            .files
169            .keys()
170            .cloned()
171            .collect::<Vec<_>>();
172        debug!(
173            "source region files: {:?}, source region id: {}",
174            source_region_manifest.files, source_region_id
175        );
176        for (file_id, file_meta) in &source_region_manifest.files {
177            if !target_region_manifest.files.contains_key(file_id) {
178                new_file_metas.push(remap_copied_file_meta(file_meta, target_region_id));
179            }
180        }
181        if new_file_metas.is_empty() {
182            return Ok((None, source_file_ids));
183        }
184
185        let file_descriptors = new_file_metas
186            .iter()
187            .flat_map(file_descriptors_for_meta)
188            .collect();
189        debug!("File descriptors to copy: {:?}", file_descriptors);
190        let copier = RegionFileCopier::new(region.access_layer());
191        // TODO(weny): ensure the target region is empty.
192        copier
193            .copy_files(
194                source_region_id,
195                target_region_id,
196                file_descriptors,
197                parallelism,
198            )
199            .await?;
200        let edit = RegionEdit {
201            files_to_add: new_file_metas,
202            files_to_remove: vec![],
203            timestamp_ms: Some(chrono::Utc::now().timestamp_millis()),
204            compaction_time_window: None,
205            flushed_entry_id: None,
206            flushed_sequence: None,
207            committed_sequence: None,
208        };
209        let action_list = RegionMetaActionList::with_action(RegionMetaAction::Edit(edit.clone()));
210        info!(
211            "Applying {:?} to region {target_region_id}, reason: CopyRegionFrom",
212            edit
213        );
214        let version = region
215            .manifest_ctx
216            .update_manifest(RegionLeaderState::Writable, action_list, false)
217            .await?;
218        info!(
219            "Successfully update manifest version to {version}, region: {target_region_id}, reason: CopyRegionFrom"
220        );
221
222        Ok((Some(edit), source_file_ids))
223    }
224}
225
226fn remap_copied_file_meta(file_meta: &FileMeta, target_region_id: RegionId) -> FileMeta {
227    let mut new_file_meta = file_meta.clone();
228    new_file_meta.region_id = target_region_id;
229    // The target region has an independent sequence domain: the physical
230    // per-row sequences in the copied file belong to the source region, so they
231    // must not be trusted for exact sequence-range reads on the target. Clear
232    // the `preserve_row_sequence` marker and source-domain max sequence to fail
233    // closed until the scan provably cannot intersect the copied rows (see
234    // `files_allow_exact_sequence_range`); otherwise an exact request would
235    // replay source-domain rows as target sequences, and an unmarked file with
236    // a stale source-domain `sequence` hint could be silently skipped as
237    // "proven disjoint".
238    new_file_meta.preserve_row_sequence = false;
239    new_file_meta.sequence = None;
240    new_file_meta
241}
242
243fn file_descriptors_for_meta(file_meta: &FileMeta) -> Vec<FileDescriptor> {
244    let data = FileDescriptor::Data {
245        file_id: file_meta.file_id,
246        size: file_meta.file_size,
247    };
248    if file_meta.exists_index() {
249        let region_index_id = file_meta.index_id();
250        vec![
251            data,
252            FileDescriptor::Index {
253                file_id: region_index_id.file_id.file_id(),
254                version: region_index_id.version,
255                size: file_meta.index_file_size(),
256            },
257        ]
258    } else {
259        vec![data]
260    }
261}
mito2/worker/handle_copy_region.rs

mito2/worker/
handle_copy_region.rs
