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(
118                &region,
119                FlushReason::Alter,
120                None,
121                self.config.clone(),
122                region.is_staging(),
123            );
124            if let Err(e) =
125                self.flush_scheduler
126                    .schedule_flush(region.region_id, &region.version_control, task)
127            {
128                // Unable to flush the region, send error to waiter.
129                sender.send(Err(e));
130                return;
131            }
132
133            // Safety: We have requested flush.
134            self.flush_scheduler
135                .add_ddl_request_to_pending(SenderDdlRequest {
136                    region_id,
137                    sender,
138                    request: DdlRequest::Alter(request),
139                });
140
141            return;
142        }
143
144        info!(
145            "Try to alter region {}, version.metadata: {:?}, version.options: {:?}, request: {:?}",
146            region_id, version.metadata, version.options, request,
147        );
148        self.handle_alter_region_with_empty_memtable(region, version, request, sender);
149    }
150
151    // TODO(yingwen): Optional new options and sst format.
152    /// Handles region metadata and format changes when the region memtable is empty.
153    fn handle_alter_region_with_empty_memtable(
154        &mut self,
155        region: MitoRegionRef,
156        version: VersionRef,
157        request: RegionAlterRequest,
158        sender: OptionOutputTx,
159    ) {
160        let need_index = need_change_index(&request.kind);
161        let new_options = new_region_options_on_empty_memtable(&version.options, &request.kind);
162        let new_meta = match metadata_after_alteration(&version.metadata, request) {
163            Ok(new_meta) => new_meta,
164            Err(e) => {
165                sender.send(Err(e));
166                return;
167            }
168        };
169        // Persist the metadata to region's manifest.
170        let change = RegionChange {
171            metadata: new_meta,
172            sst_format: new_options
173                .as_ref()
174                .unwrap_or(&version.options)
175                .sst_format
176                .unwrap_or_default(),
177        };
178        self.handle_manifest_region_change(region, change, need_index, new_options, sender);
179    }
180
181    /// Handles requests that changes region options, like TTL. It only affects memory state
182    /// since changes are persisted in the `DatanodeTableValue` in metasrv.
183    ///
184    /// If the options require empty memtable, it only does validation.
185    ///
186    /// Returns a new version with the updated options if it needs further alteration.
187    fn handle_alter_region_options_fast(
188        &mut self,
189        region: &MitoRegionRef,
190        version: VersionRef,
191        options: Vec<SetRegionOption>,
192    ) -> std::result::Result<Option<VersionRef>, MetadataError> {
193        assert!(!options.is_empty());
194
195        let mut all_options_altered = true;
196        let mut current_options = version.options.clone();
197        for option in options {
198            match option {
199                SetRegionOption::Ttl(new_ttl) => {
200                    info!(
201                        "Update region ttl: {}, previous: {:?} new: {:?}",
202                        region.region_id, current_options.ttl, new_ttl
203                    );
204                    current_options.ttl = new_ttl;
205                }
206                SetRegionOption::Twsc(key, value) => {
207                    let Twcs(options) = &mut current_options.compaction;
208                    set_twcs_options(
209                        options,
210                        &TwcsOptions::default(),
211                        &key,
212                        &value,
213                        region.region_id,
214                    )?;
215                }
216                SetRegionOption::Format(format_str) => {
217                    let new_format = format_str.parse::<FormatType>().map_err(|_| {
218                        store_api::metadata::InvalidRegionRequestSnafu {
219                            region_id: region.region_id,
220                            err: format!("Invalid format type: {}", format_str),
221                        }
222                        .build()
223                    })?;
224                    // If the format is unchanged, we also consider the option is altered.
225                    if new_format != current_options.sst_format.unwrap_or_default() {
226                        all_options_altered = false;
227
228                        // Validates the format type.
229                        ensure!(
230                            new_format == FormatType::Flat,
231                            store_api::metadata::InvalidRegionRequestSnafu {
232                                region_id: region.region_id,
233                                err: "Only allow changing format type to flat",
234                            }
235                        );
236                    }
237                }
238            }
239        }
240        region.version_control.alter_options(current_options);
241        if all_options_altered {
242            Ok(None)
243        } else {
244            Ok(Some(region.version()))
245        }
246    }
247}
248
249/// Returns the new region options if there are updates to the options.
250fn new_region_options_on_empty_memtable(
251    current_options: &RegionOptions,
252    kind: &AlterKind,
253) -> Option<RegionOptions> {
254    let AlterKind::SetRegionOptions { options } = kind else {
255        return None;
256    };
257
258    if options.is_empty() {
259        return None;
260    }
261
262    let mut current_options = current_options.clone();
263    for option in options {
264        match option {
265            SetRegionOption::Ttl(_) | SetRegionOption::Twsc(_, _) => (),
266            SetRegionOption::Format(format_str) => {
267                // Safety: handle_alter_region_options_fast() has validated this.
268                let new_format = format_str.parse::<FormatType>().unwrap();
269                assert_eq!(FormatType::Flat, new_format);
270
271                current_options.sst_format = Some(new_format);
272            }
273        }
274    }
275    Some(current_options)
276}
277
278/// Creates a metadata after applying the alter `request` to the old `metadata`.
279///
280/// Returns an error if the `request` is invalid.
281fn metadata_after_alteration(
282    metadata: &RegionMetadata,
283    request: RegionAlterRequest,
284) -> Result<RegionMetadataRef> {
285    let mut builder = RegionMetadataBuilder::from_existing(metadata.clone());
286    builder
287        .alter(request.kind)
288        .context(InvalidRegionRequestSnafu)?
289        .bump_version();
290    let new_meta = builder.build().context(InvalidMetadataSnafu)?;
291
292    Ok(Arc::new(new_meta))
293}
294
295fn set_twcs_options(
296    options: &mut TwcsOptions,
297    default_option: &TwcsOptions,
298    key: &str,
299    value: &str,
300    region_id: RegionId,
301) -> std::result::Result<(), MetadataError> {
302    match key {
303        mito_engine_options::TWCS_TRIGGER_FILE_NUM => {
304            let files = parse_usize_with_default(key, value, default_option.trigger_file_num)?;
305            log_option_update(region_id, key, options.trigger_file_num, files);
306            options.trigger_file_num = files;
307        }
308        mito_engine_options::TWCS_MAX_OUTPUT_FILE_SIZE => {
309            let size = if value.is_empty() {
310                default_option.max_output_file_size
311            } else {
312                Some(
313                    ReadableSize::from_str(value)
314                        .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?,
315                )
316            };
317            log_option_update(region_id, key, options.max_output_file_size, size);
318            options.max_output_file_size = size;
319        }
320        mito_engine_options::TWCS_TIME_WINDOW => {
321            let window = if value.is_empty() {
322                default_option.time_window
323            } else {
324                Some(
325                    humantime::parse_duration(value)
326                        .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?,
327                )
328            };
329            log_option_update(region_id, key, options.time_window, window);
330            options.time_window = window;
331        }
332        _ => return InvalidSetRegionOptionRequestSnafu { key, value }.fail(),
333    }
334    Ok(())
335}
336
337fn parse_usize_with_default(
338    key: &str,
339    value: &str,
340    default: usize,
341) -> std::result::Result<usize, MetadataError> {
342    if value.is_empty() {
343        Ok(default)
344    } else {
345        value
346            .parse::<usize>()
347            .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())
348    }
349}
350
351fn log_option_update<T: std::fmt::Debug>(
352    region_id: RegionId,
353    option_name: &str,
354    prev_value: T,
355    cur_value: T,
356) {
357    info!(
358        "Update region {}: {}, previous: {:?}, new: {:?}",
359        option_name, region_id, prev_value, cur_value
360    );
361}
362
363/// Used to determine whether we can build index directly after schema change.
364fn need_change_index(kind: &AlterKind) -> bool {
365    match kind {
366        // `SetIndexes` is a fast-path operation because it can build indexes for existing SSTs
367        // in the background, without needing to wait for a flush or compaction cycle.
368        AlterKind::SetIndexes { options: _ } => true,
369        // For AddColumns, DropColumns, UnsetIndexes and ModifyColumnTypes, we don't treat them as index changes.
370        // Index files still need to be rebuilt after schema changes,
371        // but this will happen automatically during flush or compaction.
372        _ => false,
373    }
374}
mito2/worker/handle_alter.rs

mito2/worker/
handle_alter.rs
