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

mito2/worker/
handle_alter.rs
