mito2/worker/
handle_alter.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 alter related requests.
16
17use std::str::FromStr;
18use std::sync::Arc;
19
20use common_base::readable_size::ReadableSize;
21use common_telemetry::info;
22use common_telemetry::tracing::warn;
23use humantime_serde::re::humantime;
24use snafu::{ResultExt, ensure};
25use store_api::metadata::{
26    InvalidSetRegionOptionRequestSnafu, MetadataError, RegionMetadata, RegionMetadataBuilder,
27    RegionMetadataRef,
28};
29use store_api::mito_engine_options;
30use store_api::region_request::{AlterKind, RegionAlterRequest, SetRegionOption};
31use store_api::storage::RegionId;
32
33use crate::error::{InvalidMetadataSnafu, InvalidRegionRequestSnafu, Result};
34use crate::flush::FlushReason;
35use crate::manifest::action::RegionChange;
36use crate::region::MitoRegionRef;
37use crate::region::options::CompactionOptions::Twcs;
38use crate::region::options::{RegionOptions, TwcsOptions};
39use crate::region::version::VersionRef;
40use crate::request::{DdlRequest, OptionOutputTx, SenderDdlRequest};
41use crate::sst::FormatType;
42use crate::worker::RegionWorkerLoop;
43
44impl<S> RegionWorkerLoop<S> {
45    pub(crate) async fn handle_alter_request(
46        &mut self,
47        region_id: RegionId,
48        request: RegionAlterRequest,
49        sender: OptionOutputTx,
50    ) {
51        let region = match self.regions.writable_non_staging_region(region_id) {
52            Ok(region) => region,
53            Err(e) => {
54                sender.send(Err(e));
55                return;
56            }
57        };
58
59        info!("Try to alter region: {}, request: {:?}", region_id, request);
60
61        // Gets the version before alter.
62        let mut version = region.version();
63
64        // fast path for memory state changes like options.
65        let set_options = match &request.kind {
66            AlterKind::SetRegionOptions { options } => options.clone(),
67            AlterKind::UnsetRegionOptions { keys } => {
68                // Converts the keys to SetRegionOption.
69                //
70                // It passes an empty string to achieve the purpose of unset
71                keys.iter().map(Into::into).collect()
72            }
73            _ => Vec::new(),
74        };
75        if !set_options.is_empty() {
76            match self.handle_alter_region_options_fast(&region, version, set_options) {
77                Ok(new_version) => {
78                    let Some(new_version) = new_version else {
79                        // We don't have options to alter after flush.
80                        sender.send(Ok(0));
81                        return;
82                    };
83                    version = new_version;
84                }
85                Err(e) => {
86                    sender.send(Err(e).context(InvalidMetadataSnafu));
87                    return;
88                }
89            }
90        }
91
92        // Validates request.
93        if let Err(e) = request.validate(&version.metadata) {
94            // Invalid request.
95            sender.send(Err(e).context(InvalidRegionRequestSnafu));
96            return;
97        }
98
99        // Checks whether we need to alter the region.
100        if !request.need_alter(&version.metadata) {
101            warn!(
102                "Ignores alter request as it alters nothing, region_id: {}, request: {:?}",
103                region_id, request
104            );
105            sender.send(Ok(0));
106            return;
107        }
108
109        // Checks whether we can alter the region directly.
110        if !version.memtables.is_empty() {
111            // If memtable is not empty, we can't alter it directly and need to flush
112            // all memtables first.
113            info!("Flush region: {} before alteration", region_id);
114
115            // Try to submit a flush task.
116            let task = self.new_flush_task(&region, FlushReason::Alter, None, self.config.clone());
117            if let Err(e) =
118                self.flush_scheduler
119                    .schedule_flush(region.region_id, &region.version_control, task)
120            {
121                // Unable to flush the region, send error to waiter.
122                sender.send(Err(e));
123                return;
124            }
125
126            // Safety: We have requested flush.
127            self.flush_scheduler
128                .add_ddl_request_to_pending(SenderDdlRequest {
129                    region_id,
130                    sender,
131                    request: DdlRequest::Alter(request),
132                });
133
134            return;
135        }
136
137        info!(
138            "Try to alter region {}, version.metadata: {:?}, version.options: {:?}, request: {:?}",
139            region_id, version.metadata, version.options, request,
140        );
141        self.handle_alter_region_with_empty_memtable(region, version, request, sender);
142    }
143
144    // TODO(yingwen): Optional new options and sst format.
145    /// Handles region metadata and format changes when the region memtable is empty.
146    fn handle_alter_region_with_empty_memtable(
147        &mut self,
148        region: MitoRegionRef,
149        version: VersionRef,
150        request: RegionAlterRequest,
151        sender: OptionOutputTx,
152    ) {
153        let need_index = need_change_index(&request.kind);
154        let new_options = new_region_options_on_empty_memtable(&version.options, &request.kind);
155        let new_meta = match metadata_after_alteration(&version.metadata, request) {
156            Ok(new_meta) => new_meta,
157            Err(e) => {
158                sender.send(Err(e));
159                return;
160            }
161        };
162        // Persist the metadata to region's manifest.
163        let change = RegionChange {
164            metadata: new_meta,
165            sst_format: new_options
166                .as_ref()
167                .unwrap_or(&version.options)
168                .sst_format
169                .unwrap_or_default(),
170        };
171        self.handle_manifest_region_change(region, change, need_index, new_options, sender);
172    }
173
174    /// Handles requests that changes region options, like TTL. It only affects memory state
175    /// since changes are persisted in the `DatanodeTableValue` in metasrv.
176    ///
177    /// If the options require empty memtable, it only does validation.
178    ///
179    /// Returns a new version with the updated options if it needs further alteration.
180    fn handle_alter_region_options_fast(
181        &mut self,
182        region: &MitoRegionRef,
183        version: VersionRef,
184        options: Vec<SetRegionOption>,
185    ) -> std::result::Result<Option<VersionRef>, MetadataError> {
186        assert!(!options.is_empty());
187
188        let mut all_options_altered = true;
189        let mut current_options = version.options.clone();
190        for option in options {
191            match option {
192                SetRegionOption::Ttl(new_ttl) => {
193                    info!(
194                        "Update region ttl: {}, previous: {:?} new: {:?}",
195                        region.region_id, current_options.ttl, new_ttl
196                    );
197                    current_options.ttl = new_ttl;
198                }
199                SetRegionOption::Twsc(key, value) => {
200                    let Twcs(options) = &mut current_options.compaction;
201                    set_twcs_options(
202                        options,
203                        &TwcsOptions::default(),
204                        &key,
205                        &value,
206                        region.region_id,
207                    )?;
208                }
209                SetRegionOption::Format(format_str) => {
210                    let new_format = format_str.parse::<FormatType>().map_err(|_| {
211                        store_api::metadata::InvalidRegionRequestSnafu {
212                            region_id: region.region_id,
213                            err: format!("Invalid format type: {}", format_str),
214                        }
215                        .build()
216                    })?;
217                    // If the format is unchanged, we also consider the option is altered.
218                    if new_format != current_options.sst_format.unwrap_or_default() {
219                        all_options_altered = false;
220
221                        // Validates the format type.
222                        ensure!(
223                            new_format == FormatType::Flat,
224                            store_api::metadata::InvalidRegionRequestSnafu {
225                                region_id: region.region_id,
226                                err: "Only allow changing format type to flat",
227                            }
228                        );
229                    }
230                }
231            }
232        }
233        region.version_control.alter_options(current_options);
234        if all_options_altered {
235            Ok(None)
236        } else {
237            Ok(Some(region.version()))
238        }
239    }
240}
241
242/// Returns the new region options if there are updates to the options.
243fn new_region_options_on_empty_memtable(
244    current_options: &RegionOptions,
245    kind: &AlterKind,
246) -> Option<RegionOptions> {
247    let AlterKind::SetRegionOptions { options } = kind else {
248        return None;
249    };
250
251    if options.is_empty() {
252        return None;
253    }
254
255    let mut current_options = current_options.clone();
256    for option in options {
257        match option {
258            SetRegionOption::Ttl(_) | SetRegionOption::Twsc(_, _) => (),
259            SetRegionOption::Format(format_str) => {
260                // Safety: handle_alter_region_options_fast() has validated this.
261                let new_format = format_str.parse::<FormatType>().unwrap();
262                assert_eq!(FormatType::Flat, new_format);
263
264                current_options.sst_format = Some(new_format);
265            }
266        }
267    }
268    Some(current_options)
269}
270
271/// Creates a metadata after applying the alter `request` to the old `metadata`.
272///
273/// Returns an error if the `request` is invalid.
274fn metadata_after_alteration(
275    metadata: &RegionMetadata,
276    request: RegionAlterRequest,
277) -> Result<RegionMetadataRef> {
278    let mut builder = RegionMetadataBuilder::from_existing(metadata.clone());
279    builder
280        .alter(request.kind)
281        .context(InvalidRegionRequestSnafu)?
282        .bump_version();
283    let new_meta = builder.build().context(InvalidMetadataSnafu)?;
284
285    Ok(Arc::new(new_meta))
286}
287
288fn set_twcs_options(
289    options: &mut TwcsOptions,
290    default_option: &TwcsOptions,
291    key: &str,
292    value: &str,
293    region_id: RegionId,
294) -> std::result::Result<(), MetadataError> {
295    match key {
296        mito_engine_options::TWCS_TRIGGER_FILE_NUM => {
297            let files = parse_usize_with_default(key, value, default_option.trigger_file_num)?;
298            log_option_update(region_id, key, options.trigger_file_num, files);
299            options.trigger_file_num = files;
300        }
301        mito_engine_options::TWCS_MAX_OUTPUT_FILE_SIZE => {
302            let size = if value.is_empty() {
303                default_option.max_output_file_size
304            } else {
305                Some(
306                    ReadableSize::from_str(value)
307                        .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?,
308                )
309            };
310            log_option_update(region_id, key, options.max_output_file_size, size);
311            options.max_output_file_size = size;
312        }
313        mito_engine_options::TWCS_TIME_WINDOW => {
314            let window = if value.is_empty() {
315                default_option.time_window
316            } else {
317                Some(
318                    humantime::parse_duration(value)
319                        .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?,
320                )
321            };
322            log_option_update(region_id, key, options.time_window, window);
323            options.time_window = window;
324        }
325        _ => return InvalidSetRegionOptionRequestSnafu { key, value }.fail(),
326    }
327    Ok(())
328}
329
330fn parse_usize_with_default(
331    key: &str,
332    value: &str,
333    default: usize,
334) -> std::result::Result<usize, MetadataError> {
335    if value.is_empty() {
336        Ok(default)
337    } else {
338        value
339            .parse::<usize>()
340            .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())
341    }
342}
343
344fn log_option_update<T: std::fmt::Debug>(
345    region_id: RegionId,
346    option_name: &str,
347    prev_value: T,
348    cur_value: T,
349) {
350    info!(
351        "Update region {}: {}, previous: {:?}, new: {:?}",
352        option_name, region_id, prev_value, cur_value
353    );
354}
355
356/// Used to determine whether we can build index directly after schema change.
357fn need_change_index(kind: &AlterKind) -> bool {
358    match kind {
359        // `SetIndexes` is a fast-path operation because it can build indexes for existing SSTs
360        // in the background, without needing to wait for a flush or compaction cycle.
361        AlterKind::SetIndexes { options: _ } => true,
362        // For AddColumns, DropColumns, UnsetIndexes and ModifyColumnTypes, we don't treat them as index changes.
363        // Index files still need to be rebuilt after schema changes,
364        // but this will happen automatically during flush or compaction.
365        _ => false,
366    }
367}
mito2/worker/handle_alter.rs

mito2/worker/
handle_alter.rs
