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::tracing::warn;
22use common_telemetry::{error, info};
23use humantime_serde::re::humantime;
24use snafu::{ResultExt, ensure};
25use store_api::logstore::LogStore;
26use store_api::logstore::provider::Provider;
27use store_api::metadata::{
28    InvalidSetRegionOptionRequestSnafu, MetadataError, RegionMetadata, RegionMetadataBuilder,
29    RegionMetadataRef,
30};
31use store_api::mito_engine_options;
32use store_api::mito_engine_options::MAX_ROW_GROUP_ROW_COUNT_LIMIT;
33use store_api::region_request::{AlterKind, RegionAlterRequest, SetRegionOption};
34use store_api::storage::RegionId;
35
36use crate::error::{InvalidMetadataSnafu, InvalidRegionRequestSnafu, Result};
37use crate::flush::FlushReason;
38use crate::manifest::action::RegionChange;
39use crate::memtable::ensure_json2_uses_flat_format;
40use crate::region::MitoRegionRef;
41use crate::region::options::CompactionOptions::Twcs;
42use crate::region::options::{RegionOptions, TwcsOptions};
43use crate::region::version::VersionRef;
44use crate::request::{DdlRequest, OptionOutputTx, SenderDdlRequest};
45use crate::sst::FormatType;
46use crate::worker::RegionWorkerLoop;
47
48pub(crate) type PendingAlters = std::collections::HashMap<
49    RegionId,
50    std::collections::VecDeque<(RegionAlterRequest, OptionOutputTx)>,
51>;
52
53impl<S: LogStore> RegionWorkerLoop<S> {
54    pub(crate) async fn handle_alter_request(
55        &mut self,
56        region_id: RegionId,
57        request: RegionAlterRequest,
58        sender: OptionOutputTx,
59    ) {
60        if self.flush_scheduler.has_pending_alter(region_id)
61            || self.pending_alters.contains_key(&region_id)
62            || self.regions.get_region(region_id).is_some_and(|region| {
63                matches!(
64                    region.state(),
65                    crate::region::RegionRoleState::Leader(
66                        crate::region::RegionLeaderState::Editing
67                            | crate::region::RegionLeaderState::Altering
68                    )
69                )
70            })
71        {
72            let queue = self.pending_alters.entry(region_id).or_default();
73            if queue.len() >= 128 {
74                sender.send(crate::error::RegionBusySnafu { region_id }.fail());
75            } else {
76                queue.push_back((request, sender));
77            }
78            self.dispatch_pending_region_requests(region_id).await;
79            return;
80        }
81        self.execute_alter_request(region_id, request, sender).await;
82    }
83
84    pub(crate) async fn execute_alter_request(
85        &mut self,
86        region_id: RegionId,
87        request: RegionAlterRequest,
88        sender: OptionOutputTx,
89    ) {
90        let requested_skip_wal = skip_wal_value(&request.kind);
91        let (region, follower_skip_wal) = match self.regions.writable_non_staging_region(region_id)
92        {
93            Ok(region) => (region, None),
94            Err(_) if requested_skip_wal.is_some() => {
95                match self.regions.follower_region(region_id) {
96                    Ok(region) => (region, requested_skip_wal),
97                    Err(e) => {
98                        sender.send(Err(e));
99                        return;
100                    }
101                }
102            }
103            Err(e) => {
104                sender.send(Err(e));
105                return;
106            }
107        };
108
109        info!("Try to alter region: {}, request: {:?}", region_id, request);
110
111        // Followers only accept skip-WAL changes, which are in-memory option changes and must
112        // never enter the leader path that may flush memtables.
113        if let Some(skip_wal) = follower_skip_wal {
114            if let Err(e) = validate_skip_wal_change(&region, skip_wal) {
115                sender.send(Err(e).context(InvalidMetadataSnafu));
116                return;
117            }
118            let mut options = region.version().options.clone();
119            if options.skip_wal != skip_wal {
120                info!(
121                    "Set skip_wal for follower region: {}, previous: {} new: {}",
122                    region_id, options.skip_wal, skip_wal
123                );
124                options.skip_wal = skip_wal;
125                region.version_control.alter_options(options);
126            }
127            sender.send(Ok(0));
128            return;
129        }
130
131        // Gets the version before alter.
132        let version = region.version();
133
134        // fast path for memory state changes like options.
135        let set_options = match &request.kind {
136            AlterKind::SetRegionOptions { options } => options.clone(),
137            AlterKind::UnsetRegionOptions { keys } => {
138                // Converts the keys to SetRegionOption.
139                //
140                // It passes an empty string to achieve the purpose of unset
141                keys.iter().map(Into::into).collect()
142            }
143            _ => Vec::new(),
144        };
145        let mut new_options = None;
146        if !set_options.is_empty() {
147            match self.handle_alter_region_options_fast(&region, version.clone(), set_options) {
148                Ok(staged_options) => {
149                    let Some(staged_options) = staged_options else {
150                        // We don't have options to alter after flush.
151                        sender.send(Ok(0));
152                        return;
153                    };
154                    new_options = Some(staged_options);
155                }
156                Err(e) => {
157                    sender.send(Err(e).context(InvalidMetadataSnafu));
158                    return;
159                }
160            }
161        }
162
163        // Validates request.
164        if let Err(e) = request.validate(&version.metadata) {
165            // Invalid request.
166            sender.send(Err(e).context(InvalidRegionRequestSnafu));
167            return;
168        }
169
170        // Checks whether we need to alter the region.
171        if !request.need_alter(&version.metadata) {
172            warn!(
173                "Ignores alter request as it alters nothing, region_id: {}, request: {:?}",
174                region_id, request
175            );
176            sender.send(Ok(0));
177            return;
178        }
179
180        // Checks whether we can alter the region directly.
181        if !version.memtables.is_empty() {
182            // If memtable is not empty, we can't alter it directly and need to flush
183            // all memtables first.
184            info!("Flush region: {} before alteration", region_id);
185
186            // Try to submit a flush task.
187            let task = self.new_flush_task(&region, FlushReason::Alter, None, self.config.clone());
188            if let Err(e) =
189                self.flush_scheduler
190                    .schedule_flush(region.region_id, &region.version_control, task)
191            {
192                // Unable to flush the region, send error to waiter.
193                sender.send(Err(e));
194                return;
195            }
196
197            // Safety: We have requested flush.
198            self.flush_scheduler
199                .add_ddl_request_to_pending(SenderDdlRequest {
200                    region_id,
201                    sender,
202                    request: DdlRequest::Alter(request),
203                });
204
205            return;
206        }
207
208        info!(
209            "Try to alter region {}, version.metadata: {:?}, version.options: {:?}, request: {:?}",
210            region_id, version.metadata, version.options, request,
211        );
212        log_time_index_widening_overflow(region.region_id, &version, &request);
213        self.handle_alter_region_with_empty_memtable(region, version, request, new_options, sender);
214    }
215
216    // TODO(yingwen): Optional new options and sst format.
217    /// Handles region metadata and format changes when the region memtable is empty.
218    fn handle_alter_region_with_empty_memtable(
219        &mut self,
220        region: MitoRegionRef,
221        version: VersionRef,
222        request: RegionAlterRequest,
223        new_options: Option<RegionOptions>,
224        sender: OptionOutputTx,
225    ) {
226        let need_index = need_change_index(&request.kind);
227        let new_meta = match metadata_after_alteration(&version.metadata, request) {
228            Ok(new_meta) => new_meta,
229            Err(e) => {
230                sender.send(Err(e));
231                return;
232            }
233        };
234        let options = new_options.as_ref().unwrap_or(&version.options);
235        if let Err(e) = ensure_json2_uses_flat_format(&new_meta, options) {
236            sender.send(Err(e));
237            return;
238        }
239        if let Err(e) = options.validate_column_compression(&new_meta) {
240            sender.send(Err(e));
241            return;
242        }
243        // Persist the metadata to region's manifest.
244        let change = RegionChange {
245            metadata: new_meta,
246            sst_format: options.sst_format.unwrap_or_default(),
247            append_mode: Some(options.append_mode),
248        };
249        self.handle_manifest_region_change(region, change, need_index, new_options, sender);
250    }
251
252    /// Handles requests that changes region options, like TTL. It only affects memory state
253    /// since changes are persisted in the `DatanodeTableValue` in metasrv.
254    ///
255    /// Options that can be applied without an empty memtable (e.g. TTL,
256    /// `preserve_row_sequence`) are applied directly through the fast path.
257    /// If another option in the same ALTER requires an empty memtable (e.g.
258    /// `append_mode`), only the complete final options are staged and the
259    /// existing flush path is followed.
260    ///
261    /// Returns the staged options if they need further alteration.
262    fn handle_alter_region_options_fast(
263        &mut self,
264        region: &MitoRegionRef,
265        version: VersionRef,
266        options: Vec<SetRegionOption>,
267    ) -> std::result::Result<Option<RegionOptions>, MetadataError> {
268        assert!(!options.is_empty());
269
270        let mut all_options_altered = true;
271        let mut current_options = version.options.clone();
272        for option in options.iter().cloned() {
273            match option {
274                SetRegionOption::WriteBufferSize(new_write_buffer_size) => {
275                    info!(
276                        "Update region write_buffer_size: {}, previous: {:?} new: {:?}",
277                        region.region_id, current_options.write_buffer_size, new_write_buffer_size
278                    );
279                    current_options.write_buffer_size = new_write_buffer_size;
280                    current_options.validate().map_err(|e| {
281                        store_api::metadata::InvalidRegionRequestSnafu {
282                            region_id: region.region_id,
283                            err: e.to_string(),
284                        }
285                        .build()
286                    })?;
287                }
288                SetRegionOption::Ttl(new_ttl) => {
289                    info!(
290                        "Update region ttl: {}, previous: {:?} new: {:?}",
291                        region.region_id, current_options.ttl, new_ttl
292                    );
293                    current_options.ttl = new_ttl;
294                }
295                SetRegionOption::Twsc(key, value) => {
296                    let Twcs(options) = &mut current_options.compaction;
297                    set_twcs_options(
298                        options,
299                        &TwcsOptions::default(),
300                        &key,
301                        &value,
302                        region.region_id,
303                    )?;
304                    if !value.is_empty() {
305                        current_options.compaction_override = true;
306                    }
307                }
308                SetRegionOption::Format(format_str) => {
309                    let new_format = format_str.parse::<FormatType>().map_err(|_| {
310                        store_api::metadata::InvalidRegionRequestSnafu {
311                            region_id: region.region_id,
312                            err: format!("Invalid format type: {}", format_str),
313                        }
314                        .build()
315                    })?;
316                    // If the format is unchanged, we also consider the option is altered.
317                    if new_format != current_options.sst_format.unwrap_or_default() {
318                        all_options_altered = false;
319                    }
320                }
321                SetRegionOption::AppendMode(new_append_mode) => {
322                    // If the append mode is unchanged, we consider the option is altered.
323                    if new_append_mode != current_options.append_mode {
324                        // Validates: only allow changing from false to true.
325                        ensure!(
326                            !current_options.append_mode && new_append_mode,
327                            store_api::metadata::InvalidRegionRequestSnafu {
328                                region_id: region.region_id,
329                                err: "Only allow changing append_mode from false to true",
330                            }
331                        );
332                        // Clear merge_mode since it's incompatible with append_mode.
333                        current_options.merge_mode = None;
334                        all_options_altered = false;
335                    }
336                }
337                SetRegionOption::AutoFlushInterval(new_interval) => {
338                    // The flush logic reads the effective interval from the region's
339                    // current version each cycle, so the change takes effect on the
340                    // next flush without a memtable flush.
341                    if new_interval != current_options.auto_flush_interval {
342                        info!(
343                            "Update region auto_flush_interval: {}, previous: {:?} new: {:?}",
344                            region.region_id, current_options.auto_flush_interval, new_interval
345                        );
346                        current_options.auto_flush_interval = new_interval;
347                    }
348                }
349                SetRegionOption::CompactionOutputParallelism(parallelism) => {
350                    // New compactions use the updated options; running ones retain
351                    // their snapshot. No memtable flush is needed.
352                    current_options.compaction_output_parallelism = parallelism;
353                }
354                SetRegionOption::MaxRowGroupRowCount(new_row_count) => {
355                    if let Some(row_count) = new_row_count {
356                        ensure!(
357                            row_count > 0 && row_count <= MAX_ROW_GROUP_ROW_COUNT_LIMIT,
358                            store_api::metadata::InvalidRegionRequestSnafu {
359                                region_id: region.region_id,
360                                err: format!(
361                                    "max_row_group_row_count must be in (0, \
362                                     {MAX_ROW_GROUP_ROW_COUNT_LIMIT}], got {row_count}"
363                                ),
364                            }
365                        );
366                    }
367                    if new_row_count != current_options.max_row_group_row_count {
368                        all_options_altered = false;
369                    }
370                }
371                SetRegionOption::PreserveRowSequence(new_preserve) => {
372                    if new_preserve != current_options.preserve_row_sequence {
373                        info!(
374                            "Update region preserve_row_sequence: {}, previous: {:?} new: {:?}",
375                            region.region_id, current_options.preserve_row_sequence, new_preserve
376                        );
377                        current_options.preserve_row_sequence = new_preserve;
378                    }
379                }
380                SetRegionOption::SkipWal(skip_wal) => {
381                    validate_skip_wal_change(region, skip_wal)?;
382                    if current_options.skip_wal != skip_wal {
383                        info!(
384                            "Set skip_wal for region: {}, previous: {} new: {}",
385                            region.region_id, current_options.skip_wal, skip_wal
386                        );
387                        current_options.skip_wal = skip_wal;
388                    }
389                }
390            }
391        }
392        let kind = AlterKind::SetRegionOptions { options };
393        // Validate the complete final options. The loop above validates
394        // per-option, but a combined ALTER may flip append_mode and toggle
395        // preserve_row_sequence in the same request, which is only valid when
396        // viewed together (order-independent). Validating the staged outcome
397        // also guarantees that a preserve-only toggle applied via the fast path
398        // below cannot create an invalid state (preserve requires append_mode).
399        let candidate = new_region_options_on_empty_memtable(&current_options, &kind)
400            .unwrap_or_else(|| current_options.clone());
401        candidate.validate().map_err(|e| {
402            store_api::metadata::InvalidRegionRequestSnafu {
403                region_id: region.region_id,
404                err: e.to_string(),
405            }
406            .build()
407        })?;
408        candidate
409            .validate_column_compression(&version.metadata)
410            .map_err(|e| {
411                store_api::metadata::InvalidRegionRequestSnafu {
412                    region_id: region.region_id,
413                    err: e.to_string(),
414                }
415                .build()
416            })?;
417        if all_options_altered {
418            region.version_control.alter_options(candidate);
419            Ok(None)
420        } else {
421            // Some options require an empty memtable (e.g. append_mode,
422            // sst_format, or max_row_group_row_count).
423            Ok(Some(candidate))
424        }
425    }
426}
427
428fn skip_wal_value(kind: &AlterKind) -> Option<bool> {
429    let AlterKind::SetRegionOptions { options } = kind else {
430        return None;
431    };
432    let [SetRegionOption::SkipWal(skip_wal)] = options.as_slice() else {
433        return None;
434    };
435    Some(*skip_wal)
436}
437
438fn validate_skip_wal_change(
439    region: &MitoRegionRef,
440    skip_wal: bool,
441) -> std::result::Result<(), MetadataError> {
442    ensure!(
443        skip_wal || !matches!(&region.provider, Provider::Noop),
444        store_api::metadata::InvalidRegionRequestSnafu {
445            region_id: region.region_id,
446            err: "cannot enable WAL because the region uses the Noop WAL provider".to_string(),
447        }
448    );
449    Ok(())
450}
451
452/// Returns the new region options if there are updates to the options.
453fn new_region_options_on_empty_memtable(
454    current_options: &RegionOptions,
455    kind: &AlterKind,
456) -> Option<RegionOptions> {
457    let options = match kind {
458        AlterKind::SetRegionOptions { options } => options.clone(),
459        AlterKind::UnsetRegionOptions { keys } => keys.iter().map(Into::into).collect(),
460        _ => return None,
461    };
462
463    if options.is_empty() {
464        return None;
465    }
466
467    let mut current_options = current_options.clone();
468    for option in &options {
469        match option {
470            SetRegionOption::WriteBufferSize(_)
471            | SetRegionOption::Ttl(_)
472            | SetRegionOption::Twsc(_, _)
473            | SetRegionOption::AutoFlushInterval(_)
474            | SetRegionOption::CompactionOutputParallelism(_)
475            | SetRegionOption::SkipWal(_) => (),
476            SetRegionOption::Format(format_str) => {
477                // Safety: handle_alter_region_options_fast() has validated this.
478                let new_format = format_str.parse::<FormatType>().unwrap();
479                current_options.sst_format = Some(new_format);
480            }
481            SetRegionOption::AppendMode(new_append_mode) => {
482                if *new_append_mode != current_options.append_mode {
483                    // Safety: handle_alter_region_options_fast() has validated that the only
484                    // supported transition is from false to true.
485                    current_options.append_mode = *new_append_mode;
486                    current_options.merge_mode = None;
487                }
488            }
489            SetRegionOption::MaxRowGroupRowCount(new_row_count) => {
490                current_options.max_row_group_row_count = *new_row_count;
491            }
492            SetRegionOption::PreserveRowSequence(new_preserve) => {
493                current_options.preserve_row_sequence = *new_preserve;
494            }
495        }
496    }
497    Some(current_options)
498}
499
500/// Creates a metadata after applying the alter `request` to the old `metadata`.
501///
502/// Returns an error if the `request` is invalid.
503fn metadata_after_alteration(
504    metadata: &RegionMetadata,
505    request: RegionAlterRequest,
506) -> Result<RegionMetadataRef> {
507    let mut builder = RegionMetadataBuilder::from_existing(metadata.clone());
508    builder
509        .alter(request.kind)
510        .context(InvalidRegionRequestSnafu)?
511        .bump_version();
512    let new_meta = builder.build().context(InvalidMetadataSnafu)?;
513
514    Ok(Arc::new(new_meta))
515}
516
517fn set_twcs_options(
518    options: &mut TwcsOptions,
519    default_option: &TwcsOptions,
520    key: &str,
521    value: &str,
522    region_id: RegionId,
523) -> std::result::Result<(), MetadataError> {
524    match key {
525        mito_engine_options::TWCS_TRIGGER_FILE_NUM
526        | mito_engine_options::TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM => {
527            let files = parse_usize_with_default(
528                key,
529                value,
530                default_option.active_window_trigger_file_num,
531            )?;
532            log_option_update(
533                region_id,
534                key,
535                options.active_window_trigger_file_num,
536                files,
537            );
538            options.active_window_trigger_file_num = files;
539        }
540        mito_engine_options::TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER => {
541            let files = parse_usize_with_default(
542                key,
543                value,
544                default_option.active_window_l1_merge_trigger,
545            )?;
546            ensure!(
547                files >= 2,
548                InvalidSetRegionOptionRequestSnafu { key, value }
549            );
550            log_option_update(
551                region_id,
552                key,
553                options.active_window_l1_merge_trigger,
554                files,
555            );
556            options.active_window_l1_merge_trigger = files;
557        }
558        mito_engine_options::TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM => {
559            let files = parse_usize_with_default(
560                key,
561                value,
562                default_option.inactive_window_trigger_file_num,
563            )?;
564            log_option_update(
565                region_id,
566                key,
567                options.inactive_window_trigger_file_num,
568                files,
569            );
570            options.inactive_window_trigger_file_num = files;
571        }
572        mito_engine_options::TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER => {
573            let files = parse_usize_with_default(
574                key,
575                value,
576                default_option.inactive_window_l1_merge_trigger,
577            )?;
578            ensure!(
579                files >= 2,
580                InvalidSetRegionOptionRequestSnafu { key, value }
581            );
582            log_option_update(
583                region_id,
584                key,
585                options.inactive_window_l1_merge_trigger,
586                files,
587            );
588            options.inactive_window_l1_merge_trigger = files;
589        }
590        mito_engine_options::TWCS_MAX_OUTPUT_FILE_SIZE => {
591            let size = if value.is_empty() {
592                default_option.max_output_file_size
593            } else {
594                Some(
595                    ReadableSize::from_str(value)
596                        .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?,
597                )
598            };
599            log_option_update(region_id, key, options.max_output_file_size, size);
600            options.max_output_file_size = size;
601        }
602        mito_engine_options::TWCS_TIME_WINDOW => {
603            let window = if value.is_empty() {
604                default_option.time_window
605            } else {
606                Some(
607                    humantime::parse_duration(value)
608                        .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?,
609                )
610            };
611            log_option_update(region_id, key, options.time_window, window);
612            options.time_window = window;
613        }
614        _ => return InvalidSetRegionOptionRequestSnafu { key, value }.fail(),
615    }
616    Ok(())
617}
618
619fn parse_usize_with_default(
620    key: &str,
621    value: &str,
622    default: usize,
623) -> std::result::Result<usize, MetadataError> {
624    if value.is_empty() {
625        Ok(default)
626    } else {
627        value
628            .parse::<usize>()
629            .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())
630    }
631}
632
633fn log_option_update<T: std::fmt::Debug>(
634    region_id: RegionId,
635    option_name: &str,
636    prev_value: T,
637    cur_value: T,
638) {
639    info!(
640        "Update region {}: {}, previous: {:?}, new: {:?}",
641        option_name, region_id, prev_value, cur_value
642    );
643}
644
645/// Logs a time index unit widening whose rescaled values would overflow the
646/// target unit's `i64` range in existing data. Memtables are flushed before
647/// an alter, so each SST's time range bounds its values; overflowing values
648/// cast to NULL on read, which is deemed acceptable.
649fn log_time_index_widening_overflow(
650    region_id: RegionId,
651    version: &VersionRef,
652    request: &RegionAlterRequest,
653) {
654    let AlterKind::ModifyColumnTypes { columns } = &request.kind else {
655        return;
656    };
657    let time_index = version.metadata.time_index_column();
658    for column in columns {
659        if column.column_name != time_index.column_schema.name {
660            continue;
661        }
662        // Non-timestamp targets and narrowing changes are rejected by
663        // `RegionAlterRequest::validate` before the request reaches here.
664        let Some(target_unit) = column.target_type.as_timestamp().map(|t| t.unit()) else {
665            continue;
666        };
667        for level in version.ssts.levels() {
668            for file in level.files.values() {
669                let (start, end) = file.time_range();
670                if start.convert_to(target_unit).is_none() || end.convert_to(target_unit).is_none()
671                {
672                    error!(
673                        "Time index widening for region {} overflows file {}: widening column \
674                         '{}' to {:?}, but data spans [{}, {}] beyond the target unit's i64 \
675                         range; overflowing values read back as NULL",
676                        region_id,
677                        file.file_id(),
678                        column.column_name,
679                        target_unit,
680                        start.to_iso8601_string(),
681                        end.to_iso8601_string(),
682                    );
683                    return;
684                }
685            }
686        }
687    }
688}
689
690/// Used to determine whether we can build index directly after schema change.
691fn need_change_index(kind: &AlterKind) -> bool {
692    match kind {
693        // `SetIndexes` is a fast-path operation because it can build indexes for existing SSTs
694        // in the background, without needing to wait for a flush or compaction cycle.
695        AlterKind::SetIndexes { options: _ } => true,
696        // For AddColumns, DropColumns, UnsetIndexes and ModifyColumnTypes, we don't treat them as index changes.
697        // Index files still need to be rebuilt after schema changes,
698        // but this will happen automatically during flush or compaction.
699        _ => false,
700    }
701}
702
703#[cfg(test)]
704mod tests {
705    use super::*;
706
707    #[test]
708    fn test_set_twcs_window_trigger_options() {
709        let mut options = TwcsOptions::default();
710        let defaults = options.clone();
711        let region_id = RegionId::new(1, 1);
712
713        set_twcs_options(
714            &mut options,
715            &defaults,
716            "compaction.twcs.active_window.trigger_file_num",
717            "8",
718            region_id,
719        )
720        .unwrap();
721        set_twcs_options(
722            &mut options,
723            &defaults,
724            "compaction.twcs.active_window.l1_merge_trigger",
725            "16",
726            region_id,
727        )
728        .unwrap();
729        set_twcs_options(
730            &mut options,
731            &defaults,
732            "compaction.twcs.inactive_window.trigger_file_num",
733            "3",
734            region_id,
735        )
736        .unwrap();
737        set_twcs_options(
738            &mut options,
739            &defaults,
740            "compaction.twcs.inactive_window.l1_merge_trigger",
741            "12",
742            region_id,
743        )
744        .unwrap();
745        assert_eq!(8, options.active_window_trigger_file_num);
746        assert_eq!(16, options.active_window_l1_merge_trigger);
747        assert_eq!(3, options.inactive_window_trigger_file_num);
748        assert_eq!(12, options.inactive_window_l1_merge_trigger);
749
750        set_twcs_options(
751            &mut options,
752            &defaults,
753            "compaction.twcs.active_window.trigger_file_num",
754            "",
755            region_id,
756        )
757        .unwrap();
758        set_twcs_options(
759            &mut options,
760            &defaults,
761            "compaction.twcs.active_window.l1_merge_trigger",
762            "",
763            region_id,
764        )
765        .unwrap();
766        set_twcs_options(
767            &mut options,
768            &defaults,
769            "compaction.twcs.inactive_window.trigger_file_num",
770            "",
771            region_id,
772        )
773        .unwrap();
774        set_twcs_options(
775            &mut options,
776            &defaults,
777            "compaction.twcs.inactive_window.l1_merge_trigger",
778            "",
779            region_id,
780        )
781        .unwrap();
782        assert_eq!(defaults, options);
783    }
784
785    #[test]
786    fn test_set_twcs_window_trigger_accepts_one() {
787        let defaults = TwcsOptions::default();
788        for key in [
789            "compaction.twcs.trigger_file_num",
790            "compaction.twcs.active_window.trigger_file_num",
791            "compaction.twcs.inactive_window.trigger_file_num",
792        ] {
793            let mut options = defaults.clone();
794            assert!(
795                set_twcs_options(&mut options, &defaults, key, "1", RegionId::new(1, 1)).is_ok(),
796                "{key}"
797            );
798        }
799    }
800
801    #[test]
802    fn test_set_twcs_l1_merge_triggers_reject_one() {
803        let defaults = TwcsOptions::default();
804        for key in [
805            "compaction.twcs.active_window.l1_merge_trigger",
806            "compaction.twcs.inactive_window.l1_merge_trigger",
807        ] {
808            let mut options = TwcsOptions::default();
809            assert!(
810                set_twcs_options(&mut options, &defaults, key, "1", RegionId::new(1, 1)).is_err(),
811                "{key}"
812            );
813        }
814    }
815
816    #[test]
817    fn test_new_region_options_with_idempotent_append_mode() {
818        let current_options = RegionOptions::default();
819        let kind = AlterKind::SetRegionOptions {
820            options: vec![
821                SetRegionOption::AppendMode(false),
822                SetRegionOption::MaxRowGroupRowCount(Some(1024)),
823            ],
824        };
825
826        let new_options = new_region_options_on_empty_memtable(&current_options, &kind).unwrap();
827        assert!(!new_options.append_mode);
828        assert_eq!(Some(1024), new_options.max_row_group_row_count);
829    }
830}
mito2/worker/handle_alter.rs

mito2/worker/
handle_alter.rs
