1use std::collections::HashMap;
20use std::time::Duration;
21
22use api::v1::SemanticType;
23use common_base::readable_size::ReadableSize;
24use common_telemetry::info;
25use common_time::TimeToLive;
26use common_wal::options::{WAL_OPTIONS_KEY, WalOptions};
27use datatypes::schema::COMPRESSION_KEY;
28use serde::de::Error as _;
29use serde::{Deserialize, Deserializer, Serialize};
30use serde_json::Value;
31use serde_with::{DisplayFromStr, NoneAsEmptyString, serde_as, with_prefix};
32use snafu::{ResultExt, ensure};
33use store_api::codec::PrimaryKeyEncoding;
34use store_api::metadata::RegionMetadata;
35use store_api::metric_engine_consts::{
36 MEMTABLE_PARTITION_TREE_PRIMARY_KEY_ENCODING, PRIMARY_KEY_ENCODING,
37};
38use store_api::mito_engine_options::{
39 COMPACTION_OVERRIDE, COMPACTION_TWCS_PREFIX, FloatFieldEncoding, MAX_ROW_GROUP_ROW_COUNT_LIMIT,
40};
41use store_api::storage::{ColumnId, RegionId};
42use strum::EnumString;
43
44use crate::error::{InvalidRegionOptionsSnafu, JsonOptionsSnafu, Result};
45use crate::memtable::bulk::BulkMemtableConfig;
46use crate::sst::FormatType;
47use crate::sst::parquet::DEFAULT_ROW_GROUP_SIZE;
48
49const DEFAULT_INDEX_SEGMENT_ROW_COUNT: usize = 1024;
50const MEMTABLE_PARTITION_TREE_PREFIX: &str = "memtable.partition_tree.";
51const MEMTABLE_BULK_PREFIX: &str = "memtable.bulk.";
52const MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD: &str = "memtable.bulk.encode_bytes_threshold";
53
54const LEGACY_PARTITION_TREE_MEMTABLE_TYPE: &str = "partition_tree";
58
59pub(crate) fn parse_wal_options(
60 options_map: &HashMap<String, String>,
61) -> std::result::Result<WalOptions, serde_json::Error> {
62 options_map
63 .get(WAL_OPTIONS_KEY)
64 .map_or(Ok(WalOptions::default()), |encoded_wal_options| {
65 serde_json::from_str(encoded_wal_options)
66 })
67}
68
69#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, EnumString)]
71#[serde(rename_all = "snake_case")]
72#[strum(serialize_all = "snake_case")]
73pub enum MergeMode {
74 #[default]
76 LastRow,
77 LastNonNull,
79}
80
81#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
87#[serde(default)]
88pub struct RegionOptions {
89 pub ttl: Option<TimeToLive>,
91 #[serde(with = "humantime_serde")]
94 pub auto_flush_interval: Option<Duration>,
95 pub compaction: CompactionOptions,
97 pub compaction_override: bool,
98 pub storage: Option<String>,
100 pub append_mode: bool,
102 pub skip_wal: bool,
104 pub wal_options: WalOptions,
106 pub index_options: IndexOptions,
108 pub memtable: Option<MemtableOptions>,
110 pub merge_mode: Option<MergeMode>,
113 pub sst_format: Option<FormatType>,
115 #[serde(skip_serializing_if = "Option::is_none")]
117 pub max_row_group_row_count: Option<usize>,
118 #[serde(skip_serializing_if = "Option::is_none")]
120 pub primary_key_encoding: Option<PrimaryKeyEncoding>,
121 #[serde(skip_serializing_if = "Option::is_none")]
125 pub write_buffer_size: Option<ReadableSize>,
126 #[serde(default, skip_serializing_if = "is_false")]
135 pub preserve_row_sequence: bool,
136 #[serde(default)]
138 pub float_field_encoding: FloatFieldEncoding,
139}
140
141fn is_false(value: &bool) -> bool {
142 !*value
143}
144
145impl RegionOptions {
146 pub fn validate(&self) -> Result<()> {
148 if self.append_mode {
149 ensure!(
150 self.merge_mode
151 .is_none_or(|mode| mode == MergeMode::LastRow),
152 InvalidRegionOptionsSnafu {
153 reason: "only last_row merge_mode is allowed when append_mode is enabled",
154 }
155 );
156 }
157 if let Some(auto_flush_interval) = self.auto_flush_interval {
158 ensure!(
159 auto_flush_interval > Duration::ZERO,
160 InvalidRegionOptionsSnafu {
161 reason: "auto_flush_interval must be greater than 0",
162 }
163 );
164 }
165 if let Some(row_count) = self.max_row_group_row_count {
166 ensure!(
167 row_count > 0 && row_count <= MAX_ROW_GROUP_ROW_COUNT_LIMIT,
168 InvalidRegionOptionsSnafu {
169 reason: format!(
170 "max_row_group_row_count must be in (0, {MAX_ROW_GROUP_ROW_COUNT_LIMIT}], got {row_count}",
171 ),
172 }
173 );
174 }
175 if self.preserve_row_sequence {
176 ensure!(
177 self.append_mode,
178 InvalidRegionOptionsSnafu {
179 reason: "preserve_row_sequence is only supported for append-only tables (append_mode must be true)",
180 }
181 );
182 }
183 let CompactionOptions::Twcs(options) = &self.compaction;
184 ensure!(
185 options.active_window_l1_merge_trigger >= 2,
186 InvalidRegionOptionsSnafu {
187 reason: "active_window.l1_merge_trigger must be at least 2",
188 }
189 );
190 ensure!(
191 options.inactive_window_l1_merge_trigger >= 2,
192 InvalidRegionOptionsSnafu {
193 reason: "inactive_window.l1_merge_trigger must be at least 2",
194 }
195 );
196 Ok(())
197 }
198
199 pub(crate) fn validate_column_compression(&self, metadata: &RegionMetadata) -> Result<()> {
202 if self.sst_format == Some(FormatType::Flat)
203 && metadata.primary_key_encoding == PrimaryKeyEncoding::Dense
204 {
205 return Ok(());
206 }
207 for column in &metadata.column_metadatas {
208 ensure!(
209 column.semantic_type != SemanticType::Tag
210 || !column
211 .column_schema
212 .metadata()
213 .contains_key(COMPRESSION_KEY),
214 InvalidRegionOptionsSnafu {
215 reason: format!(
216 "COMPRESSION on tag column {} requires flat SST format and dense primary key encoding",
217 column.column_schema.name
218 ),
219 }
220 );
221 }
222 Ok(())
223 }
224
225 pub fn row_group_size(&self) -> usize {
227 self.max_row_group_row_count
228 .unwrap_or(DEFAULT_ROW_GROUP_SIZE)
229 }
230
231 pub fn need_dedup(&self) -> bool {
233 !self.append_mode
234 }
235
236 pub fn merge_mode(&self) -> MergeMode {
238 self.merge_mode.unwrap_or_default()
239 }
240
241 pub fn auto_flush_interval_or(&self, default: Duration) -> Duration {
243 self.auto_flush_interval.unwrap_or(default)
244 }
245
246 pub fn primary_key_encoding(&self) -> PrimaryKeyEncoding {
248 self.primary_key_encoding.unwrap_or_default()
249 }
250}
251
252impl RegionOptions {
253 pub fn try_from_options(
255 region_id: RegionId,
256 options_map: &HashMap<String, String>,
257 ) -> Result<Self> {
258 let value = options_map_to_value(options_map);
259 let json = serde_json::to_string(&value).context(JsonOptionsSnafu)?;
260
261 let options: RegionOptionsWithoutEnum =
265 serde_json::from_str(&json).context(JsonOptionsSnafu)?;
266 let has_compaction_type =
267 validate_enum_options(options_map, "compaction.type", &[COMPACTION_TWCS_PREFIX])?;
268 let compaction = if has_compaction_type {
269 serde_json::from_str(&json).context(JsonOptionsSnafu)?
270 } else {
271 CompactionOptions::default()
272 };
273
274 let wal_options = parse_wal_options(options_map).context(JsonOptionsSnafu)?;
275
276 let index_options: IndexOptions = serde_json::from_str(&json).context(JsonOptionsSnafu)?;
277 let is_legacy_partition_tree = options_map
278 .get("memtable.type")
279 .map(|s| s.eq_ignore_ascii_case(LEGACY_PARTITION_TREE_MEMTABLE_TYPE))
280 .unwrap_or(false);
281 let memtable = if validate_enum_options(
282 options_map,
283 "memtable.type",
284 &[MEMTABLE_PARTITION_TREE_PREFIX, MEMTABLE_BULK_PREFIX],
285 )? {
286 if is_legacy_partition_tree {
287 None
291 } else {
292 Some(serde_json::from_str(&json).context(JsonOptionsSnafu)?)
293 }
294 } else {
295 None
296 };
297
298 let mut sst_format = options.sst_format;
301 if is_legacy_partition_tree {
302 info!(
303 "Region {} specified the removed partition_tree memtable; \
304 overriding memtable to the default and SST format to flat",
305 region_id
306 );
307 sst_format = Some(FormatType::Flat);
308 }
309
310 if matches!(memtable, Some(MemtableOptions::Bulk(_))) {
313 if let Some(format) = sst_format
314 && format != FormatType::Flat
315 {
316 info!(
317 "Region {} uses bulk memtable; overriding sst_format from {:?} to flat",
318 region_id, format
319 );
320 }
321 sst_format = Some(FormatType::Flat);
322 }
323
324 let float_field_encoding = options.float_field_encoding.unwrap_or_default();
325
326 let compaction_override_flag = options_map
327 .get(COMPACTION_OVERRIDE)
328 .map(|v| matches!(v.to_lowercase().as_str(), "true" | "1"))
329 .unwrap_or(false);
330 let compaction_override = has_compaction_type || compaction_override_flag;
331 let primary_key_encoding = options_map
332 .get(PRIMARY_KEY_ENCODING)
333 .or_else(|| options_map.get(MEMTABLE_PARTITION_TREE_PRIMARY_KEY_ENCODING))
334 .map(|v| match v.to_lowercase().as_str() {
335 "dense" => Ok(PrimaryKeyEncoding::Dense),
336 "sparse" => Ok(PrimaryKeyEncoding::Sparse),
337 _ => Err(InvalidRegionOptionsSnafu {
338 reason: format!("Invalid primary key encoding: {v}"),
339 }
340 .build()),
341 })
342 .transpose()?;
343
344 let opts = RegionOptions {
345 ttl: options.ttl,
346 auto_flush_interval: options.auto_flush_interval,
347 compaction,
348 compaction_override,
349 storage: options.storage,
350 append_mode: options.append_mode,
351 skip_wal: options.skip_wal,
352 wal_options,
353 index_options,
354 memtable,
355 merge_mode: options.merge_mode,
356 sst_format,
357 max_row_group_row_count: options.max_row_group_row_count,
358 primary_key_encoding,
359 write_buffer_size: options.write_buffer_size,
360 preserve_row_sequence: options.preserve_row_sequence,
361 float_field_encoding,
362 };
363 opts.validate()?;
364
365 Ok(opts)
366 }
367
368 pub(crate) fn try_from_options_with_bulk_config(
370 region_id: RegionId,
371 options_map: &HashMap<String, String>,
372 default_bulk_config: &BulkMemtableConfig,
373 ) -> Result<Self> {
374 let mut options = Self::try_from_options(region_id, options_map)?;
375 if !options_map.contains_key(MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD)
376 && let Some(MemtableOptions::Bulk(config)) = &mut options.memtable
377 {
378 config.encode_bytes_threshold = default_bulk_config.encode_bytes_threshold;
379 }
380 Ok(options)
381 }
382}
383
384#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
386#[serde(tag = "compaction.type")]
387#[serde(rename_all = "snake_case")]
388pub enum CompactionOptions {
389 #[serde(with = "prefix_twcs")]
391 Twcs(TwcsOptions),
392}
393
394impl CompactionOptions {
395 pub(crate) fn time_window(&self) -> Option<Duration> {
396 match self {
397 CompactionOptions::Twcs(opts) => opts.time_window,
398 }
399 }
400
401 pub(crate) fn remote_compaction(&self) -> bool {
402 match self {
403 CompactionOptions::Twcs(opts) => opts.remote_compaction,
404 }
405 }
406
407 pub(crate) fn fallback_to_local(&self) -> bool {
408 match self {
409 CompactionOptions::Twcs(opts) => opts.fallback_to_local,
410 }
411 }
412}
413
414impl Default for CompactionOptions {
415 fn default() -> Self {
416 Self::Twcs(TwcsOptions::default())
417 }
418}
419
420#[serde_as]
422#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
423#[serde(default)]
424pub struct TwcsOptions {
425 #[serde_as(as = "DisplayFromStr")]
427 #[serde(rename = "active_window.trigger_file_num", alias = "trigger_file_num")]
428 pub active_window_trigger_file_num: usize,
429 #[serde_as(as = "DisplayFromStr")]
431 #[serde(rename = "active_window.l1_merge_trigger")]
432 pub active_window_l1_merge_trigger: usize,
433 #[serde_as(as = "DisplayFromStr")]
435 #[serde(rename = "inactive_window.trigger_file_num")]
436 pub inactive_window_trigger_file_num: usize,
437 #[serde_as(as = "DisplayFromStr")]
439 #[serde(rename = "inactive_window.l1_merge_trigger")]
440 pub inactive_window_l1_merge_trigger: usize,
441 #[serde(with = "humantime_serde")]
443 pub time_window: Option<Duration>,
444 pub max_output_file_size: Option<ReadableSize>,
448 #[serde_as(as = "DisplayFromStr")]
450 pub remote_compaction: bool,
451 #[serde_as(as = "DisplayFromStr")]
453 pub fallback_to_local: bool,
454}
455
456with_prefix!(prefix_twcs "compaction.twcs.");
457
458impl TwcsOptions {
459 pub fn time_window_seconds(&self) -> Option<i64> {
461 self.time_window.and_then(|window| {
462 let window_secs = window.as_secs();
463 if window_secs == 0 {
464 None
465 } else {
466 window_secs.try_into().ok()
467 }
468 })
469 }
470}
471
472impl Default for TwcsOptions {
473 fn default() -> Self {
474 Self {
475 active_window_trigger_file_num: 4,
476 active_window_l1_merge_trigger: 16,
477 inactive_window_trigger_file_num: 2,
478 inactive_window_l1_merge_trigger: 8,
479 time_window: None,
480 max_output_file_size: Some(ReadableSize::mb(512)),
481 remote_compaction: false,
482 fallback_to_local: true,
483 }
484 }
485}
486
487#[serde_as]
490#[derive(Debug, Deserialize)]
491#[serde(default)]
492struct RegionOptionsWithoutEnum {
493 write_buffer_size: Option<ReadableSize>,
494 ttl: Option<TimeToLive>,
496 #[serde(with = "humantime_serde")]
497 auto_flush_interval: Option<Duration>,
498 storage: Option<String>,
499 #[serde_as(as = "DisplayFromStr")]
500 append_mode: bool,
501 #[serde_as(as = "DisplayFromStr")]
502 skip_wal: bool,
503 #[serde_as(as = "NoneAsEmptyString")]
504 merge_mode: Option<MergeMode>,
505 #[serde_as(as = "NoneAsEmptyString")]
506 sst_format: Option<FormatType>,
507 #[serde_as(as = "NoneAsEmptyString")]
508 max_row_group_row_count: Option<usize>,
509 #[serde_as(as = "DisplayFromStr")]
510 preserve_row_sequence: bool,
511 #[serde(rename = "experimental_sst_float_field_encoding")]
512 #[serde_as(as = "NoneAsEmptyString")]
513 float_field_encoding: Option<FloatFieldEncoding>,
514}
515
516impl Default for RegionOptionsWithoutEnum {
517 fn default() -> Self {
518 let options = RegionOptions::default();
519 RegionOptionsWithoutEnum {
520 write_buffer_size: options.write_buffer_size,
521 ttl: options.ttl,
522 auto_flush_interval: options.auto_flush_interval,
523 storage: options.storage,
524 append_mode: options.append_mode,
525 skip_wal: options.skip_wal,
526 merge_mode: options.merge_mode,
527 sst_format: options.sst_format,
528 max_row_group_row_count: options.max_row_group_row_count,
529 preserve_row_sequence: options.preserve_row_sequence,
530 float_field_encoding: Some(options.float_field_encoding),
531 }
532 }
533}
534
535with_prefix!(prefix_inverted_index "index.inverted_index.");
536
537#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
539#[serde(default)]
540pub struct IndexOptions {
541 #[serde(flatten, with = "prefix_inverted_index")]
543 pub inverted_index: InvertedIndexOptions,
544}
545
546#[serde_as]
548#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
549#[serde(default)]
550pub struct InvertedIndexOptions {
551 #[serde(deserialize_with = "deserialize_ignore_column_ids")]
554 #[serde(serialize_with = "serialize_ignore_column_ids")]
555 pub ignore_column_ids: Vec<ColumnId>,
556
557 #[serde_as(as = "DisplayFromStr")]
559 pub segment_row_count: usize,
560}
561
562impl Default for InvertedIndexOptions {
563 fn default() -> Self {
564 Self {
565 ignore_column_ids: Vec::new(),
566 segment_row_count: DEFAULT_INDEX_SEGMENT_ROW_COUNT,
567 }
568 }
569}
570
571#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
573#[serde(tag = "memtable.type", rename_all = "snake_case")]
574pub enum MemtableOptions {
575 TimeSeries,
576 #[serde(with = "prefix_bulk")]
577 Bulk(BulkMemtableConfig),
578}
579
580with_prefix!(prefix_bulk "memtable.bulk.");
581
582fn deserialize_ignore_column_ids<'de, D>(deserializer: D) -> Result<Vec<ColumnId>, D::Error>
583where
584 D: Deserializer<'de>,
585{
586 let s: String = Deserialize::deserialize(deserializer)?;
587 let mut column_ids = Vec::new();
588 if s.is_empty() {
589 return Ok(column_ids);
590 }
591 for item in s.split(',') {
592 let column_id = item.parse().map_err(D::Error::custom)?;
593 column_ids.push(column_id);
594 }
595 Ok(column_ids)
596}
597
598fn serialize_ignore_column_ids<S>(column_ids: &[ColumnId], serializer: S) -> Result<S::Ok, S::Error>
599where
600 S: serde::Serializer,
601{
602 let s = column_ids
603 .iter()
604 .map(|id| id.to_string())
605 .collect::<Vec<_>>()
606 .join(",");
607 serializer.serialize_str(&s)
608}
609
610fn options_map_to_value(options: &HashMap<String, String>) -> Value {
614 let map = options
615 .iter()
616 .map(|(key, value)| {
617 if value.eq_ignore_ascii_case("null") {
619 (key.clone(), Value::Null)
620 } else {
621 (key.clone(), Value::from(value.clone()))
622 }
623 })
624 .collect();
625 Value::Object(map)
626}
627
628fn validate_enum_options(
636 options_map: &HashMap<String, String>,
637 enum_tag_key: &str,
638 enum_option_prefixes: &[&str],
639) -> Result<bool> {
640 let mut has_enum_options = false;
641 let mut has_tag = false;
642 for key in options_map.keys() {
643 if key == enum_tag_key {
644 has_tag = true;
645 } else if !has_enum_options
646 && enum_option_prefixes
647 .iter()
648 .any(|prefix| key.starts_with(prefix))
649 {
650 has_enum_options = true;
651 }
652
653 if has_tag && has_enum_options {
654 break;
655 }
656 }
657
658 ensure!(
660 has_tag || !has_enum_options,
661 InvalidRegionOptionsSnafu {
662 reason: format!("missing key {} in options", enum_tag_key),
663 }
664 );
665
666 Ok(has_tag)
667}
668
669#[cfg(test)]
670mod tests {
671 use common_error::ext::ErrorExt;
672 use common_error::status_code::StatusCode;
673 use common_wal::options::KafkaWalOptions;
674 use store_api::mito_engine_options::{
675 EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, SKIP_WAL_KEY, WRITE_BUFFER_SIZE_KEY,
676 };
677
678 use super::*;
679
680 fn make_map(options: &[(&str, &str)]) -> HashMap<String, String> {
681 options
682 .iter()
683 .map(|(k, v)| (k.to_string(), v.to_string()))
684 .collect()
685 }
686
687 #[test]
688 fn test_empty_region_options() {
689 let map = make_map(&[]);
690 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
691 assert_eq!(RegionOptions::default(), options);
692 }
693
694 #[test]
695 fn test_float_field_encoding_defaults_and_parses() {
696 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &make_map(&[])).unwrap();
697 assert_eq!(FloatFieldEncoding::Default, options.float_field_encoding);
698
699 let map = make_map(&[(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, "default")]);
700 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
701 assert_eq!(FloatFieldEncoding::Default, options.float_field_encoding);
702
703 let map = make_map(&[(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, "byte_stream_split")]);
704 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
705 assert_eq!(
706 FloatFieldEncoding::ByteStreamSplit,
707 options.float_field_encoding
708 );
709
710 let map = make_map(&[(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, "invalid")]);
711 assert!(RegionOptions::try_from_options(RegionId::new(0, 0), &map).is_err());
712 }
713
714 #[test]
715 fn test_with_ttl() {
716 let map = make_map(&[("ttl", "7d")]);
717 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
718 let expect = RegionOptions {
719 ttl: Some(Duration::from_secs(3600 * 24 * 7).into()),
720 ..Default::default()
721 };
722 assert_eq!(expect, options);
723 }
724
725 #[test]
726 fn test_with_skip_wal() {
727 let map = make_map(&[(SKIP_WAL_KEY, "true")]);
728 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
729 assert!(options.skip_wal);
730 }
731
732 #[test]
733 fn test_with_auto_flush_interval() {
734 let map = make_map(&[("auto_flush_interval", "5m")]);
735 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
736 let expect = RegionOptions {
737 auto_flush_interval: Some(Duration::from_secs(5 * 60)),
738 ..Default::default()
739 };
740 assert_eq!(expect, options);
741 }
742
743 #[test]
744 fn test_with_zero_auto_flush_interval() {
745 let map = make_map(&[("auto_flush_interval", "0s")]);
746 let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err();
747 assert!(
748 err.to_string().contains("auto_flush_interval"),
749 "unexpected error: {err}"
750 );
751 }
752
753 #[test]
754 fn test_with_write_buffer_size() {
755 let map = make_map(&[(WRITE_BUFFER_SIZE_KEY, "128MiB")]);
756 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
757 let expect = RegionOptions {
758 write_buffer_size: Some(ReadableSize::mb(128)),
759 ..Default::default()
760 };
761 assert_eq!(expect, options);
762 }
763
764 #[test]
765 fn test_with_zero_write_buffer_size() {
766 let map = make_map(&[(WRITE_BUFFER_SIZE_KEY, "0")]);
767 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
768 let expect = RegionOptions {
769 write_buffer_size: Some(ReadableSize::mb(0)),
770 ..Default::default()
771 };
772 assert_eq!(expect, options);
773 }
774
775 #[test]
776 fn test_with_storage() {
777 let map = make_map(&[("storage", "S3")]);
778 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
779 let expect = RegionOptions {
780 storage: Some("S3".to_string()),
781 ..Default::default()
782 };
783 assert_eq!(expect, options);
784 }
785
786 #[test]
787 fn test_without_compaction_type() {
788 let map = make_map(&[
789 ("compaction.twcs.trigger_file_num", "8"),
790 ("compaction.twcs.time_window", "2h"),
791 ]);
792 let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err();
793 assert_eq!(StatusCode::InvalidArguments, err.status_code());
794 }
795
796 #[test]
797 fn test_with_compaction_type() {
798 let map = make_map(&[
799 ("compaction.twcs.active_window.trigger_file_num", "8"),
800 ("compaction.twcs.active_window.l1_merge_trigger", "16"),
801 ("compaction.twcs.inactive_window.trigger_file_num", "2"),
802 ("compaction.twcs.inactive_window.l1_merge_trigger", "12"),
803 ("compaction.twcs.time_window", "2h"),
804 ("compaction.type", "twcs"),
805 ]);
806 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
807 let expect = RegionOptions {
808 compaction: CompactionOptions::Twcs(TwcsOptions {
809 active_window_trigger_file_num: 8,
810 active_window_l1_merge_trigger: 16,
811 inactive_window_trigger_file_num: 2,
812 inactive_window_l1_merge_trigger: 12,
813 time_window: Some(Duration::from_secs(3600 * 2)),
814 ..Default::default()
815 }),
816 compaction_override: true,
817 ..Default::default()
818 };
819 assert_eq!(expect, options);
820 }
821
822 #[test]
823 fn test_twcs_window_trigger_below_two_is_accepted_for_compatibility() {
824 let map = make_map(&[
827 ("compaction.twcs.active_window.trigger_file_num", "1"),
828 ("compaction.type", "twcs"),
829 ]);
830 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
831 let CompactionOptions::Twcs(twcs) = &options.compaction;
832 assert_eq!(1, twcs.active_window_trigger_file_num);
833
834 let map = make_map(&[
835 ("compaction.twcs.inactive_window.trigger_file_num", "1"),
836 ("compaction.type", "twcs"),
837 ]);
838 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
839 let CompactionOptions::Twcs(twcs) = &options.compaction;
840 assert_eq!(1, twcs.inactive_window_trigger_file_num);
841 }
842
843 #[test]
844 fn test_active_window_l1_merge_trigger_below_two_is_rejected() {
845 let map = make_map(&[
846 ("compaction.twcs.active_window.l1_merge_trigger", "1"),
847 ("compaction.type", "twcs"),
848 ]);
849
850 let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err();
851 assert_eq!(StatusCode::InvalidArguments, err.status_code());
852 }
853
854 #[test]
855 fn test_inactive_window_l1_merge_trigger_defaults_to_eight() {
856 let value = serde_json::to_value(TwcsOptions::default()).unwrap();
857 assert_eq!(
858 Some("8"),
859 value
860 .get("inactive_window.l1_merge_trigger")
861 .and_then(|value| value.as_str())
862 );
863 }
864
865 #[test]
866 fn test_inactive_window_l1_merge_trigger_below_two_is_rejected() {
867 let map = make_map(&[
868 ("compaction.twcs.inactive_window.l1_merge_trigger", "1"),
869 ("compaction.type", "twcs"),
870 ]);
871
872 let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err();
873 assert_eq!(StatusCode::InvalidArguments, err.status_code());
874 }
875
876 #[test]
877 fn test_with_compaction_override_true_without_compaction_type() {
878 let map = make_map(&[(COMPACTION_OVERRIDE, "true")]);
879 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
880 let expect = RegionOptions {
881 compaction_override: true,
882 ..Default::default()
883 };
884 assert_eq!(expect, options);
885 }
886
887 #[test]
888 fn test_with_compaction_override_false_without_compaction_type() {
889 let map = make_map(&[(COMPACTION_OVERRIDE, "false")]);
890 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
891 assert_eq!(RegionOptions::default(), options);
892 }
893
894 #[test]
895 fn test_compaction_twcs_options_still_require_compaction_type_with_override() {
896 let map = make_map(&[
897 (COMPACTION_OVERRIDE, "true"),
898 ("compaction.twcs.time_window", "2h"),
899 ]);
900 let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err();
901 assert_eq!(StatusCode::InvalidArguments, err.status_code());
902 }
903
904 fn test_with_wal_options(wal_options: &WalOptions) -> bool {
905 let encoded_wal_options = serde_json::to_string(&wal_options).unwrap();
906 let map = make_map(&[(WAL_OPTIONS_KEY, &encoded_wal_options)]);
907 let got = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
908 let expect = RegionOptions {
909 wal_options: wal_options.clone(),
910 ..Default::default()
911 };
912 expect == got
913 }
914
915 #[test]
916 fn test_with_index() {
917 let map = make_map(&[
918 ("index.inverted_index.ignore_column_ids", "1,2,3"),
919 ("index.inverted_index.segment_row_count", "512"),
920 ]);
921 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
922 let expect = RegionOptions {
923 index_options: IndexOptions {
924 inverted_index: InvertedIndexOptions {
925 ignore_column_ids: vec![1, 2, 3],
926 segment_row_count: 512,
927 },
928 },
929 ..Default::default()
930 };
931 assert_eq!(expect, options);
932 }
933
934 #[test]
936 fn test_with_any_wal_options() {
937 let all_wal_options = [
938 WalOptions::RaftEngine,
939 WalOptions::Kafka(KafkaWalOptions::new("test_topic".to_string())),
940 ];
941 all_wal_options.iter().all(test_with_wal_options);
942 }
943
944 #[test]
945 fn test_with_memtable() {
946 let map = make_map(&[("memtable.type", "time_series")]);
947 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
948 let expect = RegionOptions {
949 memtable: Some(MemtableOptions::TimeSeries),
950 ..Default::default()
951 };
952 assert_eq!(expect, options);
953
954 let map = make_map(&[("memtable.type", "bulk")]);
955 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
956 let expect = RegionOptions {
957 memtable: Some(MemtableOptions::Bulk(BulkMemtableConfig::default())),
958 sst_format: Some(FormatType::Flat),
959 ..Default::default()
960 };
961 assert_eq!(expect, options);
962
963 let map = make_map(&[
964 ("memtable.type", "bulk"),
965 ("memtable.bulk.merge_threshold", "7"),
966 ("memtable.bulk.encode_row_threshold", "11"),
967 ("memtable.bulk.encode_bytes_threshold", "13"),
968 ("memtable.bulk.max_merge_groups", "17"),
969 ]);
970 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
971 let expect = RegionOptions {
972 memtable: Some(MemtableOptions::Bulk(BulkMemtableConfig {
973 merge_threshold: 7,
974 encode_row_threshold: 11,
975 encode_bytes_threshold: 13,
976 max_merge_groups: 17,
977 })),
978 sst_format: Some(FormatType::Flat),
979 ..Default::default()
980 };
981 assert_eq!(expect, options);
982
983 let map = make_map(&[("memtable.type", "partition_tree")]);
986 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
987 let expect = RegionOptions {
988 memtable: None,
989 sst_format: Some(FormatType::Flat),
990 ..Default::default()
991 };
992 assert_eq!(expect, options);
993
994 let map = make_map(&[
996 ("memtable.type", "partition_tree"),
997 ("memtable.partition_tree.index_max_keys_per_shard", "2048"),
998 ("memtable.partition_tree.fork_dictionary_bytes", "128M"),
999 ]);
1000 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
1001 let expect = RegionOptions {
1002 memtable: None,
1003 sst_format: Some(FormatType::Flat),
1004 ..Default::default()
1005 };
1006 assert_eq!(expect, options);
1007 }
1008
1009 #[test]
1010 fn test_primary_key_encoding() {
1011 let map = make_map(&[("primary_key_encoding", "sparse")]);
1013 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
1014 assert_eq!(options.primary_key_encoding(), PrimaryKeyEncoding::Sparse);
1015 assert_eq!(
1016 options.primary_key_encoding,
1017 Some(PrimaryKeyEncoding::Sparse)
1018 );
1019
1020 let map = make_map(&[
1022 ("memtable.type", "partition_tree"),
1023 ("memtable.partition_tree.primary_key_encoding", "sparse"),
1024 ]);
1025 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
1026 assert_eq!(options.memtable, None);
1027 assert_eq!(options.sst_format, Some(FormatType::Flat));
1028 assert_eq!(options.primary_key_encoding(), PrimaryKeyEncoding::Sparse);
1029
1030 let map = make_map(&[("primary_key_encoding", "bogus")]);
1032 let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err();
1033 assert_eq!(StatusCode::InvalidArguments, err.status_code());
1034 }
1035
1036 #[test]
1037 fn test_legacy_partition_tree_overrides_sst_format() {
1038 let map = make_map(&[
1041 ("memtable.type", "partition_tree"),
1042 ("sst_format", "primary_key"),
1043 ]);
1044 let options = RegionOptions::try_from_options(RegionId::new(1, 1), &map).unwrap();
1045 assert_eq!(options.memtable, None);
1046 assert_eq!(options.sst_format, Some(FormatType::Flat));
1047 }
1048
1049 #[test]
1050 fn test_bulk_memtable_overrides_sst_format() {
1051 let map = make_map(&[("memtable.type", "bulk"), ("sst_format", "primary_key")]);
1055 let options = RegionOptions::try_from_options(RegionId::new(1, 1), &map).unwrap();
1056 assert_eq!(
1057 options.memtable,
1058 Some(MemtableOptions::Bulk(BulkMemtableConfig::default()))
1059 );
1060 assert_eq!(options.sst_format, Some(FormatType::Flat));
1061 }
1062
1063 #[test]
1064 fn test_unknown_memtable_type() {
1065 let map = make_map(&[("memtable.type", "no_such_memtable")]);
1066 let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err();
1067 assert_eq!(StatusCode::InvalidArguments, err.status_code());
1068 }
1069
1070 #[test]
1071 fn test_without_memtable_type() {
1072 let map = make_map(&[("memtable.partition_tree.index_max_keys_per_shard", "2048")]);
1073 let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err();
1074 assert_eq!(StatusCode::InvalidArguments, err.status_code());
1075
1076 let map = make_map(&[("memtable.bulk.merge_threshold", "7")]);
1077 let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err();
1078 assert_eq!(StatusCode::InvalidArguments, err.status_code());
1079 }
1080
1081 #[test]
1082 fn test_with_merge_mode() {
1083 let map = make_map(&[("merge_mode", "last_row")]);
1084 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
1085 assert_eq!(MergeMode::LastRow, options.merge_mode());
1086
1087 let map = make_map(&[("merge_mode", "last_non_null")]);
1088 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
1089 assert_eq!(MergeMode::LastNonNull, options.merge_mode());
1090
1091 let map = make_map(&[("merge_mode", "unknown")]);
1092 let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err();
1093 assert_eq!(StatusCode::InvalidArguments, err.status_code());
1094 }
1095
1096 #[test]
1097 fn test_append_mode_allows_last_row_merge_mode() {
1098 let map = make_map(&[("append_mode", "true"), ("merge_mode", "last_row")]);
1099 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
1100 assert!(options.append_mode);
1101 assert_eq!(MergeMode::LastRow, options.merge_mode());
1102
1103 let map = make_map(&[("append_mode", "true"), ("merge_mode", "last_non_null")]);
1104 let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err();
1105 assert_eq!(StatusCode::InvalidArguments, err.status_code());
1106 }
1107
1108 #[test]
1109 fn test_with_all() {
1110 let wal_options = WalOptions::Kafka(KafkaWalOptions::new("test_topic".to_string()));
1111 let map = make_map(&[
1112 ("ttl", "7d"),
1113 ("compaction.twcs.trigger_file_num", "8"),
1114 ("compaction.twcs.max_output_file_size", "1GB"),
1115 ("compaction.twcs.time_window", "2h"),
1116 ("compaction.type", "twcs"),
1117 ("compaction.twcs.remote_compaction", "false"),
1118 ("compaction.twcs.fallback_to_local", "true"),
1119 ("storage", "S3"),
1120 ("append_mode", "false"),
1121 ("index.inverted_index.ignore_column_ids", "1,2,3"),
1122 ("index.inverted_index.segment_row_count", "512"),
1123 (
1124 WAL_OPTIONS_KEY,
1125 &serde_json::to_string(&wal_options).unwrap(),
1126 ),
1127 ("memtable.type", "bulk"),
1128 ("memtable.bulk.merge_threshold", "7"),
1129 ("memtable.bulk.encode_row_threshold", "11"),
1130 ("memtable.bulk.encode_bytes_threshold", "13"),
1131 ("memtable.bulk.max_merge_groups", "17"),
1132 ("merge_mode", "last_non_null"),
1133 ]);
1134 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
1135 let expect = RegionOptions {
1136 ttl: Some(Duration::from_secs(3600 * 24 * 7).into()),
1137 auto_flush_interval: None,
1138 compaction: CompactionOptions::Twcs(TwcsOptions {
1139 active_window_trigger_file_num: 8,
1140 active_window_l1_merge_trigger: 16,
1141 inactive_window_trigger_file_num: 2,
1142 inactive_window_l1_merge_trigger: 8,
1143 time_window: Some(Duration::from_secs(3600 * 2)),
1144 max_output_file_size: Some(ReadableSize::gb(1)),
1145 remote_compaction: false,
1146 fallback_to_local: true,
1147 }),
1148 compaction_override: true,
1149 storage: Some("S3".to_string()),
1150 append_mode: false,
1151 skip_wal: false,
1152 wal_options,
1153 index_options: IndexOptions {
1154 inverted_index: InvertedIndexOptions {
1155 ignore_column_ids: vec![1, 2, 3],
1156 segment_row_count: 512,
1157 },
1158 },
1159 memtable: Some(MemtableOptions::Bulk(BulkMemtableConfig {
1160 merge_threshold: 7,
1161 encode_row_threshold: 11,
1162 encode_bytes_threshold: 13,
1163 max_merge_groups: 17,
1164 })),
1165 merge_mode: Some(MergeMode::LastNonNull),
1166 sst_format: Some(FormatType::Flat),
1167 max_row_group_row_count: None,
1168 primary_key_encoding: None,
1169 write_buffer_size: None,
1170 preserve_row_sequence: false,
1171 float_field_encoding: FloatFieldEncoding::default(),
1172 };
1173 assert_eq!(expect, options);
1174 }
1175
1176 #[test]
1177 fn test_with_preserve_row_sequence() {
1178 let map = make_map(&[("append_mode", "false"), ("preserve_row_sequence", "true")]);
1179 let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err();
1180 assert_eq!(StatusCode::InvalidArguments, err.status_code());
1181 assert!(err.to_string().contains("preserve_row_sequence"));
1182
1183 let map = make_map(&[("append_mode", "true"), ("preserve_row_sequence", "true")]);
1184 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
1185 assert!(options.append_mode);
1186 assert!(options.preserve_row_sequence);
1187
1188 let map = make_map(&[("append_mode", "true")]);
1189 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
1190 assert!(!options.preserve_row_sequence);
1191
1192 let map = make_map(&[]);
1193 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
1194 assert_eq!(RegionOptions::default(), options);
1195 }
1196
1197 #[test]
1198 fn test_region_options_serde() {
1199 let options = RegionOptions {
1200 ttl: Some(Duration::from_secs(3600 * 24 * 7).into()),
1201 auto_flush_interval: None,
1202 compaction: CompactionOptions::Twcs(TwcsOptions {
1203 active_window_trigger_file_num: 8,
1204 active_window_l1_merge_trigger: 8,
1205 inactive_window_trigger_file_num: 2,
1206 inactive_window_l1_merge_trigger: 8,
1207 time_window: Some(Duration::from_secs(3600 * 2)),
1208 max_output_file_size: None,
1209 remote_compaction: false,
1210 fallback_to_local: true,
1211 }),
1212 compaction_override: false,
1213 storage: Some("S3".to_string()),
1214 append_mode: false,
1215 skip_wal: false,
1216 wal_options: WalOptions::Kafka(KafkaWalOptions::new("test_topic".to_string())),
1217 index_options: IndexOptions {
1218 inverted_index: InvertedIndexOptions {
1219 ignore_column_ids: vec![1, 2, 3],
1220 segment_row_count: 512,
1221 },
1222 },
1223 memtable: Some(MemtableOptions::Bulk(BulkMemtableConfig::default())),
1224 merge_mode: Some(MergeMode::LastNonNull),
1225 sst_format: None,
1226 max_row_group_row_count: None,
1227 primary_key_encoding: None,
1228 write_buffer_size: Some(ReadableSize::mb(128)),
1229 preserve_row_sequence: true,
1230 float_field_encoding: FloatFieldEncoding::default(),
1231 };
1232 let region_options_json_str = serde_json::to_string(&options).unwrap();
1233 assert!(region_options_json_str.contains("preserve_row_sequence"));
1234 let got: RegionOptions = serde_json::from_str(®ion_options_json_str).unwrap();
1235 assert_eq!(options, got);
1236
1237 let old_region_options_json_str = r#"{"ttl":null}"#;
1239 let got: RegionOptions = serde_json::from_str(old_region_options_json_str).unwrap();
1240 assert_eq!(None, got.write_buffer_size);
1241 assert!(!got.preserve_row_sequence);
1242 assert_eq!(FloatFieldEncoding::Default, got.float_field_encoding);
1243 let CompactionOptions::Twcs(twcs) = got.compaction;
1244 assert_eq!(16, twcs.active_window_l1_merge_trigger);
1245 assert_eq!(8, twcs.inactive_window_l1_merge_trigger);
1246
1247 let default_json = serde_json::to_value(RegionOptions::default()).unwrap();
1248 assert!(default_json.get(WRITE_BUFFER_SIZE_KEY).is_none());
1249 }
1250
1251 #[test]
1252 fn test_region_options_str_serde() {
1253 let region_options_json_str = r#"{
1255 "ttl": "7days",
1256 "compaction": {
1257 "compaction.type": "twcs",
1258 "compaction.twcs.trigger_file_num": "8",
1259 "compaction.twcs.max_output_file_size": "7MB",
1260 "compaction.twcs.time_window": "2h"
1261 },
1262 "storage": "S3",
1263 "append_mode": false,
1264 "wal_options": {
1265 "wal.provider": "kafka",
1266 "wal.kafka.topic": "test_topic"
1267 },
1268 "index_options": {
1269 "index.inverted_index.ignore_column_ids": "",
1270 "index.inverted_index.segment_row_count": "512"
1271 },
1272 "memtable": {
1273 "memtable.type": "bulk"
1274 },
1275 "merge_mode": "last_non_null"
1276}"#;
1277 let got: RegionOptions = serde_json::from_str(region_options_json_str).unwrap();
1278 let options = RegionOptions {
1279 ttl: Some(Duration::from_secs(3600 * 24 * 7).into()),
1280 auto_flush_interval: None,
1281 compaction: CompactionOptions::Twcs(TwcsOptions {
1282 active_window_trigger_file_num: 8,
1283 active_window_l1_merge_trigger: 16,
1284 inactive_window_trigger_file_num: 2,
1285 inactive_window_l1_merge_trigger: 8,
1286 time_window: Some(Duration::from_secs(3600 * 2)),
1287 max_output_file_size: Some(ReadableSize::mb(7)),
1288 remote_compaction: false,
1289 fallback_to_local: true,
1290 }),
1291 compaction_override: false,
1292 storage: Some("S3".to_string()),
1293 append_mode: false,
1294 skip_wal: false,
1295 wal_options: WalOptions::Kafka(KafkaWalOptions::new("test_topic".to_string())),
1296 index_options: IndexOptions {
1297 inverted_index: InvertedIndexOptions {
1298 ignore_column_ids: vec![],
1299 segment_row_count: 512,
1300 },
1301 },
1302 memtable: Some(MemtableOptions::Bulk(BulkMemtableConfig::default())),
1303 merge_mode: Some(MergeMode::LastNonNull),
1304 sst_format: None,
1305 max_row_group_row_count: None,
1306 primary_key_encoding: None,
1307 write_buffer_size: None,
1308 preserve_row_sequence: false,
1309 float_field_encoding: FloatFieldEncoding::default(),
1310 };
1311 assert_eq!(options, got);
1312 }
1313
1314 #[test]
1315 fn test_max_row_group_row_count() {
1316 assert_eq!(None, RegionOptions::default().max_row_group_row_count);
1318 assert_eq!(
1319 DEFAULT_ROW_GROUP_SIZE,
1320 RegionOptions::default().row_group_size()
1321 );
1322
1323 let map = make_map(&[("max_row_group_row_count", "51200")]);
1325 let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap();
1326 assert_eq!(Some(51200), options.max_row_group_row_count);
1327 assert_eq!(51200, options.row_group_size());
1328
1329 let map = make_map(&[("max_row_group_row_count", "0")]);
1331 assert!(RegionOptions::try_from_options(RegionId::new(0, 0), &map).is_err());
1332 }
1333}