1use 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::region::MitoRegionRef;
40use crate::region::options::CompactionOptions::Twcs;
41use crate::region::options::{RegionOptions, TwcsOptions};
42use crate::region::version::VersionRef;
43use crate::request::{DdlRequest, OptionOutputTx, SenderDdlRequest};
44use crate::sst::FormatType;
45use crate::worker::RegionWorkerLoop;
46
47impl<S: LogStore> RegionWorkerLoop<S> {
48 pub(crate) async fn handle_alter_request(
49 &mut self,
50 region_id: RegionId,
51 request: RegionAlterRequest,
52 sender: OptionOutputTx,
53 ) {
54 let requested_skip_wal = skip_wal_value(&request.kind);
55 let (region, follower_skip_wal) = match self.regions.writable_non_staging_region(region_id)
56 {
57 Ok(region) => (region, None),
58 Err(_) if requested_skip_wal.is_some() => {
59 match self.regions.follower_region(region_id) {
60 Ok(region) => (region, requested_skip_wal),
61 Err(e) => {
62 sender.send(Err(e));
63 return;
64 }
65 }
66 }
67 Err(e) => {
68 sender.send(Err(e));
69 return;
70 }
71 };
72
73 info!("Try to alter region: {}, request: {:?}", region_id, request);
74
75 if let Some(skip_wal) = follower_skip_wal {
78 if let Err(e) = validate_skip_wal_change(®ion, skip_wal) {
79 sender.send(Err(e).context(InvalidMetadataSnafu));
80 return;
81 }
82 let mut options = region.version().options.clone();
83 if options.skip_wal != skip_wal {
84 info!(
85 "Set skip_wal for follower region: {}, previous: {} new: {}",
86 region_id, options.skip_wal, skip_wal
87 );
88 options.skip_wal = skip_wal;
89 region.version_control.alter_options(options);
90 }
91 sender.send(Ok(0));
92 return;
93 }
94
95 let version = region.version();
97
98 let set_options = match &request.kind {
100 AlterKind::SetRegionOptions { options } => options.clone(),
101 AlterKind::UnsetRegionOptions { keys } => {
102 keys.iter().map(Into::into).collect()
106 }
107 _ => Vec::new(),
108 };
109 let mut new_options = None;
110 if !set_options.is_empty() {
111 match self.handle_alter_region_options_fast(®ion, version.clone(), set_options) {
112 Ok(staged_options) => {
113 let Some(staged_options) = staged_options else {
114 sender.send(Ok(0));
116 return;
117 };
118 new_options = Some(staged_options);
119 }
120 Err(e) => {
121 sender.send(Err(e).context(InvalidMetadataSnafu));
122 return;
123 }
124 }
125 }
126
127 if let Err(e) = request.validate(&version.metadata) {
129 sender.send(Err(e).context(InvalidRegionRequestSnafu));
131 return;
132 }
133
134 if !request.need_alter(&version.metadata) {
136 warn!(
137 "Ignores alter request as it alters nothing, region_id: {}, request: {:?}",
138 region_id, request
139 );
140 sender.send(Ok(0));
141 return;
142 }
143
144 if !version.memtables.is_empty() {
146 info!("Flush region: {} before alteration", region_id);
149
150 let task = self.new_flush_task(®ion, FlushReason::Alter, None, self.config.clone());
152 if let Err(e) =
153 self.flush_scheduler
154 .schedule_flush(region.region_id, ®ion.version_control, task)
155 {
156 sender.send(Err(e));
158 return;
159 }
160
161 self.flush_scheduler
163 .add_ddl_request_to_pending(SenderDdlRequest {
164 region_id,
165 sender,
166 request: DdlRequest::Alter(request),
167 });
168
169 return;
170 }
171
172 info!(
173 "Try to alter region {}, version.metadata: {:?}, version.options: {:?}, request: {:?}",
174 region_id, version.metadata, version.options, request,
175 );
176 log_time_index_widening_overflow(region.region_id, &version, &request);
177 self.handle_alter_region_with_empty_memtable(region, version, request, new_options, sender);
178 }
179
180 fn handle_alter_region_with_empty_memtable(
183 &mut self,
184 region: MitoRegionRef,
185 version: VersionRef,
186 request: RegionAlterRequest,
187 new_options: Option<RegionOptions>,
188 sender: OptionOutputTx,
189 ) {
190 let need_index = need_change_index(&request.kind);
191 let new_meta = match metadata_after_alteration(&version.metadata, request) {
192 Ok(new_meta) => new_meta,
193 Err(e) => {
194 sender.send(Err(e));
195 return;
196 }
197 };
198 let options = new_options.as_ref().unwrap_or(&version.options);
200 if let Err(e) = options.validate_column_compression(&new_meta) {
201 sender.send(Err(e));
202 return;
203 }
204 let change = RegionChange {
205 metadata: new_meta,
206 sst_format: options.sst_format.unwrap_or_default(),
207 append_mode: Some(options.append_mode),
208 };
209 self.handle_manifest_region_change(region, change, need_index, new_options, sender);
210 }
211
212 fn handle_alter_region_options_fast(
223 &mut self,
224 region: &MitoRegionRef,
225 version: VersionRef,
226 options: Vec<SetRegionOption>,
227 ) -> std::result::Result<Option<RegionOptions>, MetadataError> {
228 assert!(!options.is_empty());
229
230 let mut all_options_altered = true;
231 let mut current_options = version.options.clone();
232 for option in options.iter().cloned() {
233 match option {
234 SetRegionOption::WriteBufferSize(new_write_buffer_size) => {
235 info!(
236 "Update region write_buffer_size: {}, previous: {:?} new: {:?}",
237 region.region_id, current_options.write_buffer_size, new_write_buffer_size
238 );
239 current_options.write_buffer_size = new_write_buffer_size;
240 current_options.validate().map_err(|e| {
241 store_api::metadata::InvalidRegionRequestSnafu {
242 region_id: region.region_id,
243 err: e.to_string(),
244 }
245 .build()
246 })?;
247 }
248 SetRegionOption::Ttl(new_ttl) => {
249 info!(
250 "Update region ttl: {}, previous: {:?} new: {:?}",
251 region.region_id, current_options.ttl, new_ttl
252 );
253 current_options.ttl = new_ttl;
254 }
255 SetRegionOption::Twsc(key, value) => {
256 let Twcs(options) = &mut current_options.compaction;
257 set_twcs_options(
258 options,
259 &TwcsOptions::default(),
260 &key,
261 &value,
262 region.region_id,
263 )?;
264 if !value.is_empty() {
265 current_options.compaction_override = true;
266 }
267 }
268 SetRegionOption::Format(format_str) => {
269 let new_format = format_str.parse::<FormatType>().map_err(|_| {
270 store_api::metadata::InvalidRegionRequestSnafu {
271 region_id: region.region_id,
272 err: format!("Invalid format type: {}", format_str),
273 }
274 .build()
275 })?;
276 if new_format != current_options.sst_format.unwrap_or_default() {
278 all_options_altered = false;
279 }
280 }
281 SetRegionOption::AppendMode(new_append_mode) => {
282 if new_append_mode != current_options.append_mode {
284 ensure!(
286 !current_options.append_mode && new_append_mode,
287 store_api::metadata::InvalidRegionRequestSnafu {
288 region_id: region.region_id,
289 err: "Only allow changing append_mode from false to true",
290 }
291 );
292 current_options.merge_mode = None;
294 all_options_altered = false;
295 }
296 }
297 SetRegionOption::AutoFlushInterval(new_interval) => {
298 if new_interval != current_options.auto_flush_interval {
302 info!(
303 "Update region auto_flush_interval: {}, previous: {:?} new: {:?}",
304 region.region_id, current_options.auto_flush_interval, new_interval
305 );
306 current_options.auto_flush_interval = new_interval;
307 }
308 }
309 SetRegionOption::MaxRowGroupRowCount(new_row_count) => {
310 if let Some(row_count) = new_row_count {
311 ensure!(
312 row_count > 0 && row_count <= MAX_ROW_GROUP_ROW_COUNT_LIMIT,
313 store_api::metadata::InvalidRegionRequestSnafu {
314 region_id: region.region_id,
315 err: format!(
316 "max_row_group_row_count must be in (0, \
317 {MAX_ROW_GROUP_ROW_COUNT_LIMIT}], got {row_count}"
318 ),
319 }
320 );
321 }
322 if new_row_count != current_options.max_row_group_row_count {
323 all_options_altered = false;
324 }
325 }
326 SetRegionOption::PreserveRowSequence(new_preserve) => {
327 if new_preserve != current_options.preserve_row_sequence {
328 info!(
329 "Update region preserve_row_sequence: {}, previous: {:?} new: {:?}",
330 region.region_id, current_options.preserve_row_sequence, new_preserve
331 );
332 current_options.preserve_row_sequence = new_preserve;
333 }
334 }
335 SetRegionOption::SkipWal(skip_wal) => {
336 validate_skip_wal_change(region, skip_wal)?;
337 if current_options.skip_wal != skip_wal {
338 info!(
339 "Set skip_wal for region: {}, previous: {} new: {}",
340 region.region_id, current_options.skip_wal, skip_wal
341 );
342 current_options.skip_wal = skip_wal;
343 }
344 }
345 }
346 }
347 let kind = AlterKind::SetRegionOptions { options };
348 let candidate = new_region_options_on_empty_memtable(¤t_options, &kind)
355 .unwrap_or_else(|| current_options.clone());
356 candidate.validate().map_err(|e| {
357 store_api::metadata::InvalidRegionRequestSnafu {
358 region_id: region.region_id,
359 err: e.to_string(),
360 }
361 .build()
362 })?;
363 candidate
364 .validate_column_compression(&version.metadata)
365 .map_err(|e| {
366 store_api::metadata::InvalidRegionRequestSnafu {
367 region_id: region.region_id,
368 err: e.to_string(),
369 }
370 .build()
371 })?;
372 if all_options_altered {
373 region.version_control.alter_options(candidate);
374 Ok(None)
375 } else {
376 Ok(Some(candidate))
379 }
380 }
381}
382
383fn skip_wal_value(kind: &AlterKind) -> Option<bool> {
384 let AlterKind::SetRegionOptions { options } = kind else {
385 return None;
386 };
387 let [SetRegionOption::SkipWal(skip_wal)] = options.as_slice() else {
388 return None;
389 };
390 Some(*skip_wal)
391}
392
393fn validate_skip_wal_change(
394 region: &MitoRegionRef,
395 skip_wal: bool,
396) -> std::result::Result<(), MetadataError> {
397 ensure!(
398 skip_wal || !matches!(®ion.provider, Provider::Noop),
399 store_api::metadata::InvalidRegionRequestSnafu {
400 region_id: region.region_id,
401 err: "cannot enable WAL because the region uses the Noop WAL provider".to_string(),
402 }
403 );
404 Ok(())
405}
406
407fn new_region_options_on_empty_memtable(
409 current_options: &RegionOptions,
410 kind: &AlterKind,
411) -> Option<RegionOptions> {
412 let options = match kind {
413 AlterKind::SetRegionOptions { options } => options.clone(),
414 AlterKind::UnsetRegionOptions { keys } => keys.iter().map(Into::into).collect(),
415 _ => return None,
416 };
417
418 if options.is_empty() {
419 return None;
420 }
421
422 let mut current_options = current_options.clone();
423 for option in &options {
424 match option {
425 SetRegionOption::WriteBufferSize(_)
426 | SetRegionOption::Ttl(_)
427 | SetRegionOption::Twsc(_, _)
428 | SetRegionOption::AutoFlushInterval(_)
429 | SetRegionOption::SkipWal(_) => (),
430 SetRegionOption::Format(format_str) => {
431 let new_format = format_str.parse::<FormatType>().unwrap();
433 current_options.sst_format = Some(new_format);
434 }
435 SetRegionOption::AppendMode(new_append_mode) => {
436 if *new_append_mode != current_options.append_mode {
437 current_options.append_mode = *new_append_mode;
440 current_options.merge_mode = None;
441 }
442 }
443 SetRegionOption::MaxRowGroupRowCount(new_row_count) => {
444 current_options.max_row_group_row_count = *new_row_count;
445 }
446 SetRegionOption::PreserveRowSequence(new_preserve) => {
447 current_options.preserve_row_sequence = *new_preserve;
448 }
449 }
450 }
451 Some(current_options)
452}
453
454fn metadata_after_alteration(
458 metadata: &RegionMetadata,
459 request: RegionAlterRequest,
460) -> Result<RegionMetadataRef> {
461 let mut builder = RegionMetadataBuilder::from_existing(metadata.clone());
462 builder
463 .alter(request.kind)
464 .context(InvalidRegionRequestSnafu)?
465 .bump_version();
466 let new_meta = builder.build().context(InvalidMetadataSnafu)?;
467
468 Ok(Arc::new(new_meta))
469}
470
471fn set_twcs_options(
472 options: &mut TwcsOptions,
473 default_option: &TwcsOptions,
474 key: &str,
475 value: &str,
476 region_id: RegionId,
477) -> std::result::Result<(), MetadataError> {
478 match key {
479 mito_engine_options::TWCS_TRIGGER_FILE_NUM
480 | mito_engine_options::TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM => {
481 let files = parse_usize_with_default(
482 key,
483 value,
484 default_option.active_window_trigger_file_num,
485 )?;
486 log_option_update(
487 region_id,
488 key,
489 options.active_window_trigger_file_num,
490 files,
491 );
492 options.active_window_trigger_file_num = files;
493 }
494 mito_engine_options::TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER => {
495 let files = parse_usize_with_default(
496 key,
497 value,
498 default_option.active_window_l1_merge_trigger,
499 )?;
500 ensure!(
501 files >= 2,
502 InvalidSetRegionOptionRequestSnafu { key, value }
503 );
504 log_option_update(
505 region_id,
506 key,
507 options.active_window_l1_merge_trigger,
508 files,
509 );
510 options.active_window_l1_merge_trigger = files;
511 }
512 mito_engine_options::TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM => {
513 let files = parse_usize_with_default(
514 key,
515 value,
516 default_option.inactive_window_trigger_file_num,
517 )?;
518 log_option_update(
519 region_id,
520 key,
521 options.inactive_window_trigger_file_num,
522 files,
523 );
524 options.inactive_window_trigger_file_num = files;
525 }
526 mito_engine_options::TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER => {
527 let files = parse_usize_with_default(
528 key,
529 value,
530 default_option.inactive_window_l1_merge_trigger,
531 )?;
532 ensure!(
533 files >= 2,
534 InvalidSetRegionOptionRequestSnafu { key, value }
535 );
536 log_option_update(
537 region_id,
538 key,
539 options.inactive_window_l1_merge_trigger,
540 files,
541 );
542 options.inactive_window_l1_merge_trigger = files;
543 }
544 mito_engine_options::TWCS_MAX_OUTPUT_FILE_SIZE => {
545 let size = if value.is_empty() {
546 default_option.max_output_file_size
547 } else {
548 Some(
549 ReadableSize::from_str(value)
550 .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?,
551 )
552 };
553 log_option_update(region_id, key, options.max_output_file_size, size);
554 options.max_output_file_size = size;
555 }
556 mito_engine_options::TWCS_TIME_WINDOW => {
557 let window = if value.is_empty() {
558 default_option.time_window
559 } else {
560 Some(
561 humantime::parse_duration(value)
562 .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())?,
563 )
564 };
565 log_option_update(region_id, key, options.time_window, window);
566 options.time_window = window;
567 }
568 _ => return InvalidSetRegionOptionRequestSnafu { key, value }.fail(),
569 }
570 Ok(())
571}
572
573fn parse_usize_with_default(
574 key: &str,
575 value: &str,
576 default: usize,
577) -> std::result::Result<usize, MetadataError> {
578 if value.is_empty() {
579 Ok(default)
580 } else {
581 value
582 .parse::<usize>()
583 .map_err(|_| InvalidSetRegionOptionRequestSnafu { key, value }.build())
584 }
585}
586
587fn log_option_update<T: std::fmt::Debug>(
588 region_id: RegionId,
589 option_name: &str,
590 prev_value: T,
591 cur_value: T,
592) {
593 info!(
594 "Update region {}: {}, previous: {:?}, new: {:?}",
595 option_name, region_id, prev_value, cur_value
596 );
597}
598
599fn log_time_index_widening_overflow(
604 region_id: RegionId,
605 version: &VersionRef,
606 request: &RegionAlterRequest,
607) {
608 let AlterKind::ModifyColumnTypes { columns } = &request.kind else {
609 return;
610 };
611 let time_index = version.metadata.time_index_column();
612 for column in columns {
613 if column.column_name != time_index.column_schema.name {
614 continue;
615 }
616 let Some(target_unit) = column.target_type.as_timestamp().map(|t| t.unit()) else {
619 continue;
620 };
621 for level in version.ssts.levels() {
622 for file in level.files.values() {
623 let (start, end) = file.time_range();
624 if start.convert_to(target_unit).is_none() || end.convert_to(target_unit).is_none()
625 {
626 error!(
627 "Time index widening for region {} overflows file {}: widening column \
628 '{}' to {:?}, but data spans [{}, {}] beyond the target unit's i64 \
629 range; overflowing values read back as NULL",
630 region_id,
631 file.file_id(),
632 column.column_name,
633 target_unit,
634 start.to_iso8601_string(),
635 end.to_iso8601_string(),
636 );
637 return;
638 }
639 }
640 }
641 }
642}
643
644fn need_change_index(kind: &AlterKind) -> bool {
646 match kind {
647 AlterKind::SetIndexes { options: _ } => true,
650 _ => false,
654 }
655}
656
657#[cfg(test)]
658mod tests {
659 use super::*;
660
661 #[test]
662 fn test_set_twcs_window_trigger_options() {
663 let mut options = TwcsOptions::default();
664 let defaults = options.clone();
665 let region_id = RegionId::new(1, 1);
666
667 set_twcs_options(
668 &mut options,
669 &defaults,
670 "compaction.twcs.active_window.trigger_file_num",
671 "8",
672 region_id,
673 )
674 .unwrap();
675 set_twcs_options(
676 &mut options,
677 &defaults,
678 "compaction.twcs.active_window.l1_merge_trigger",
679 "16",
680 region_id,
681 )
682 .unwrap();
683 set_twcs_options(
684 &mut options,
685 &defaults,
686 "compaction.twcs.inactive_window.trigger_file_num",
687 "3",
688 region_id,
689 )
690 .unwrap();
691 set_twcs_options(
692 &mut options,
693 &defaults,
694 "compaction.twcs.inactive_window.l1_merge_trigger",
695 "12",
696 region_id,
697 )
698 .unwrap();
699 assert_eq!(8, options.active_window_trigger_file_num);
700 assert_eq!(16, options.active_window_l1_merge_trigger);
701 assert_eq!(3, options.inactive_window_trigger_file_num);
702 assert_eq!(12, options.inactive_window_l1_merge_trigger);
703
704 set_twcs_options(
705 &mut options,
706 &defaults,
707 "compaction.twcs.active_window.trigger_file_num",
708 "",
709 region_id,
710 )
711 .unwrap();
712 set_twcs_options(
713 &mut options,
714 &defaults,
715 "compaction.twcs.active_window.l1_merge_trigger",
716 "",
717 region_id,
718 )
719 .unwrap();
720 set_twcs_options(
721 &mut options,
722 &defaults,
723 "compaction.twcs.inactive_window.trigger_file_num",
724 "",
725 region_id,
726 )
727 .unwrap();
728 set_twcs_options(
729 &mut options,
730 &defaults,
731 "compaction.twcs.inactive_window.l1_merge_trigger",
732 "",
733 region_id,
734 )
735 .unwrap();
736 assert_eq!(defaults, options);
737 }
738
739 #[test]
740 fn test_set_twcs_window_trigger_accepts_one() {
741 let defaults = TwcsOptions::default();
742 for key in [
743 "compaction.twcs.trigger_file_num",
744 "compaction.twcs.active_window.trigger_file_num",
745 "compaction.twcs.inactive_window.trigger_file_num",
746 ] {
747 let mut options = defaults.clone();
748 assert!(
749 set_twcs_options(&mut options, &defaults, key, "1", RegionId::new(1, 1)).is_ok(),
750 "{key}"
751 );
752 }
753 }
754
755 #[test]
756 fn test_set_twcs_l1_merge_triggers_reject_one() {
757 let defaults = TwcsOptions::default();
758 for key in [
759 "compaction.twcs.active_window.l1_merge_trigger",
760 "compaction.twcs.inactive_window.l1_merge_trigger",
761 ] {
762 let mut options = TwcsOptions::default();
763 assert!(
764 set_twcs_options(&mut options, &defaults, key, "1", RegionId::new(1, 1)).is_err(),
765 "{key}"
766 );
767 }
768 }
769
770 #[test]
771 fn test_new_region_options_with_idempotent_append_mode() {
772 let current_options = RegionOptions::default();
773 let kind = AlterKind::SetRegionOptions {
774 options: vec![
775 SetRegionOption::AppendMode(false),
776 SetRegionOption::MaxRowGroupRowCount(Some(1024)),
777 ],
778 };
779
780 let new_options = new_region_options_on_empty_memtable(¤t_options, &kind).unwrap();
781 assert!(!new_options.append_mode);
782 assert_eq!(Some(1024), new_options.max_row_group_row_count);
783 }
784}