1use std::cmp;
18use std::path::Path;
19use std::time::Duration;
20
21use common_base::memory_limit::MemoryLimit;
22use common_base::readable_size::ReadableSize;
23use common_memory_manager::OnExhaustedPolicy;
24use common_stat::{get_total_cpu_cores, get_total_memory_readable};
25use common_telemetry::warn;
26use serde::{Deserialize, Serialize};
27use serde_with::serde_as;
28
29use crate::cache::file_cache::DEFAULT_INDEX_CACHE_PERCENT;
30use crate::error::Result;
31use crate::gc::GcConfig;
32use crate::sst::DEFAULT_WRITE_BUFFER_SIZE;
33
34const MULTIPART_UPLOAD_MINIMUM_SIZE: ReadableSize = ReadableSize::mb(5);
35pub(crate) const DEFAULT_MAX_CONCURRENT_SCAN_FILES: usize = 384;
37
38const GLOBAL_WRITE_BUFFER_SIZE_FACTOR: u64 = 8;
40const SST_META_CACHE_SIZE_FACTOR: u64 = 8;
42const PREFILTER_RESULT_CACHE_SIZE_FACTOR: u64 = 32;
44const INDEX_METADATA_CACHE_SIZE_FACTOR: u64 = 32;
46const MEM_CACHE_SIZE_FACTOR: u64 = 16;
48const PAGE_CACHE_SIZE_FACTOR: u64 = 8;
50const INDEX_CREATE_MEM_THRESHOLD_FACTOR: u64 = 16;
52
53pub(crate) const FETCH_OPTION_TIMEOUT: Duration = Duration::from_secs(3);
55
56#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
59#[serde(default)]
60pub struct MitoConfig {
61 pub num_workers: usize,
65 pub worker_channel_size: usize,
67 pub worker_request_batch_size: usize,
69
70 pub manifest_checkpoint_distance: u64,
74 pub experimental_manifest_keep_removed_file_count: usize,
79 #[serde(with = "humantime_serde")]
84 pub experimental_manifest_keep_removed_file_ttl: Duration,
85 pub compress_manifest: bool,
87
88 pub max_background_index_builds: usize,
91 pub max_background_flushes: usize,
93 pub max_background_compactions: usize,
95 pub max_background_purges: usize,
97 pub experimental_compaction_memory_limit: MemoryLimit,
101 pub experimental_compaction_on_exhausted: OnExhaustedPolicy,
103
104 #[serde(with = "humantime_serde")]
107 pub auto_flush_interval: Duration,
108 pub global_write_buffer_size: ReadableSize,
110 pub global_write_buffer_reject_size: ReadableSize,
112 pub default_region_write_buffer_size: ReadableSize,
116
117 pub sst_meta_cache_size: ReadableSize,
120 pub vector_cache_size: ReadableSize,
122 pub page_cache_size: ReadableSize,
124 pub selector_result_cache_size: ReadableSize,
126 pub range_result_cache_size: ReadableSize,
128 pub prefilter_result_cache_size: ReadableSize,
130 pub enable_write_cache: bool,
132 pub write_cache_path: String,
134 pub write_cache_size: ReadableSize,
136 #[serde(with = "humantime_serde")]
138 pub write_cache_ttl: Option<Duration>,
139 pub preload_index_cache: bool,
141 pub index_cache_percent: u8,
145 pub enable_refill_cache_on_read: bool,
148 pub manifest_cache_size: ReadableSize,
150
151 pub sst_write_buffer_size: ReadableSize,
154 pub max_concurrent_scan_files: usize,
156 pub allow_stale_entries: bool,
158 pub scan_memory_limit: MemoryLimit,
162 pub scan_memory_on_exhausted: OnExhaustedPolicy,
167
168 pub index: IndexConfig,
170 pub inverted_index: InvertedIndexConfig,
172 pub fulltext_index: FulltextIndexConfig,
174 pub bloom_filter_index: BloomFilterConfig,
176 #[cfg(feature = "vector_index")]
178 pub vector_index: VectorIndexConfig,
179
180 #[serde(with = "humantime_serde")]
183 pub min_compaction_interval: Duration,
184 pub schedule_compaction_after_edit: bool,
186
187 pub default_flat_format: bool,
190
191 pub experimental_series_scan_v2: bool,
193
194 pub gc: GcConfig,
195}
196
197impl Default for MitoConfig {
198 fn default() -> Self {
199 let mut mito_config = MitoConfig {
200 num_workers: divide_num_cpus(2),
201 worker_channel_size: 128,
202 worker_request_batch_size: 64,
203 manifest_checkpoint_distance: 10,
204 experimental_manifest_keep_removed_file_count: 256,
205 experimental_manifest_keep_removed_file_ttl: Duration::from_secs(60 * 60),
206 compress_manifest: false,
207 max_background_index_builds: divide_num_cpus(8),
208 max_background_flushes: divide_num_cpus(2),
209 max_background_compactions: divide_num_cpus(4),
210 max_background_purges: get_total_cpu_cores(),
211 experimental_compaction_memory_limit: MemoryLimit::Unlimited,
212 experimental_compaction_on_exhausted: OnExhaustedPolicy::default(),
213 auto_flush_interval: Duration::from_secs(10 * 60),
214 global_write_buffer_size: ReadableSize::gb(1),
215 global_write_buffer_reject_size: ReadableSize::gb(2),
216 default_region_write_buffer_size: ReadableSize::mb(0),
217 sst_meta_cache_size: ReadableSize::mb(128),
218 vector_cache_size: ReadableSize::mb(512),
219 page_cache_size: ReadableSize::mb(512),
220 selector_result_cache_size: ReadableSize::mb(512),
221 range_result_cache_size: ReadableSize::mb(512),
222 prefilter_result_cache_size: ReadableSize::mb(128),
223 enable_write_cache: false,
224 write_cache_path: String::new(),
225 write_cache_size: ReadableSize::gb(5),
226 write_cache_ttl: None,
227 preload_index_cache: true,
228 index_cache_percent: DEFAULT_INDEX_CACHE_PERCENT,
229 enable_refill_cache_on_read: true,
230 manifest_cache_size: ReadableSize::mb(256),
231 sst_write_buffer_size: DEFAULT_WRITE_BUFFER_SIZE,
232 max_concurrent_scan_files: DEFAULT_MAX_CONCURRENT_SCAN_FILES,
233 allow_stale_entries: false,
234 scan_memory_limit: MemoryLimit::default(),
235 scan_memory_on_exhausted: OnExhaustedPolicy::Fail,
236 index: IndexConfig::default(),
237 inverted_index: InvertedIndexConfig::default(),
238 fulltext_index: FulltextIndexConfig::default(),
239 bloom_filter_index: BloomFilterConfig::default(),
240 #[cfg(feature = "vector_index")]
241 vector_index: VectorIndexConfig::default(),
242 min_compaction_interval: Duration::from_secs(0),
243 schedule_compaction_after_edit: true,
244 default_flat_format: true,
245 experimental_series_scan_v2: true,
246 gc: GcConfig::default(),
247 };
248
249 if let Some(sys_memory) = get_total_memory_readable() {
251 mito_config.adjust_buffer_and_cache_size(sys_memory);
252 }
253
254 mito_config
255 }
256}
257
258impl MitoConfig {
259 pub fn sanitize(&mut self, data_home: &str) -> Result<()> {
263 if self.num_workers == 0 {
265 self.num_workers = divide_num_cpus(2);
266 }
267
268 if self.worker_channel_size == 0 {
270 warn!("Sanitize channel size 0 to 1");
271 self.worker_channel_size = 1;
272 }
273
274 if self.max_background_flushes == 0 {
275 warn!(
276 "Sanitize max background flushes 0 to {}",
277 divide_num_cpus(2)
278 );
279 self.max_background_flushes = divide_num_cpus(2);
280 }
281 if self.max_background_compactions == 0 {
282 warn!(
283 "Sanitize max background compactions 0 to {}",
284 divide_num_cpus(4)
285 );
286 self.max_background_compactions = divide_num_cpus(4);
287 }
288 if self.max_background_purges == 0 {
289 let cpu_cores = get_total_cpu_cores();
290 warn!("Sanitize max background purges 0 to {}", cpu_cores);
291 self.max_background_purges = cpu_cores;
292 }
293
294 if self.global_write_buffer_reject_size <= self.global_write_buffer_size {
295 self.global_write_buffer_reject_size = self.global_write_buffer_size * 2;
296 warn!(
297 "Sanitize global write buffer reject size to {}",
298 self.global_write_buffer_reject_size
299 );
300 }
301
302 if self.sst_write_buffer_size < MULTIPART_UPLOAD_MINIMUM_SIZE {
303 self.sst_write_buffer_size = MULTIPART_UPLOAD_MINIMUM_SIZE;
304 warn!(
305 "Sanitize sst write buffer size to {}",
306 self.sst_write_buffer_size
307 );
308 }
309
310 if self.write_cache_path.trim().is_empty() {
312 self.write_cache_path = data_home.to_string();
313 }
314
315 if self.index_cache_percent == 0 || self.index_cache_percent >= 100 {
317 warn!(
318 "Invalid index_cache_percent {}, resetting to default {}",
319 self.index_cache_percent, DEFAULT_INDEX_CACHE_PERCENT
320 );
321 self.index_cache_percent = DEFAULT_INDEX_CACHE_PERCENT;
322 }
323
324 self.index.sanitize(data_home, &self.inverted_index)?;
325
326 Ok(())
327 }
328
329 fn adjust_buffer_and_cache_size(&mut self, sys_memory: ReadableSize) {
330 let global_write_buffer_size = cmp::min(
332 sys_memory / GLOBAL_WRITE_BUFFER_SIZE_FACTOR,
333 ReadableSize::gb(1),
334 );
335 let global_write_buffer_reject_size = global_write_buffer_size * 2;
337 let sst_meta_cache_size = cmp::min(
340 sys_memory / SST_META_CACHE_SIZE_FACTOR,
341 ReadableSize::mb(512),
342 );
343 let prefilter_result_cache_size = cmp::min(
344 sys_memory / PREFILTER_RESULT_CACHE_SIZE_FACTOR,
345 ReadableSize::mb(128),
346 );
347 let mem_cache_size = cmp::min(sys_memory / MEM_CACHE_SIZE_FACTOR, ReadableSize::mb(512));
349 let page_cache_size = sys_memory / PAGE_CACHE_SIZE_FACTOR;
350
351 self.global_write_buffer_size = global_write_buffer_size;
352 self.global_write_buffer_reject_size = global_write_buffer_reject_size;
353 self.sst_meta_cache_size = sst_meta_cache_size;
354 self.vector_cache_size = mem_cache_size;
355 self.page_cache_size = page_cache_size;
356 self.selector_result_cache_size = mem_cache_size;
357 self.range_result_cache_size = mem_cache_size;
358 self.prefilter_result_cache_size = prefilter_result_cache_size;
360
361 self.index.adjust_buffer_and_cache_size(sys_memory);
362 }
363
364 #[cfg(test)]
366 pub fn enable_write_cache(
367 mut self,
368 path: String,
369 size: ReadableSize,
370 ttl: Option<Duration>,
371 ) -> Self {
372 self.enable_write_cache = true;
373 self.write_cache_path = path;
374 self.write_cache_size = size;
375 self.write_cache_ttl = ttl;
376 self
377 }
378}
379
380#[cfg(test)]
381mod tests {
382 use super::*;
383
384 #[test]
385 fn test_default_auto_flush_interval() {
386 assert_eq!(
387 Duration::from_secs(10 * 60),
388 MitoConfig::default().auto_flush_interval
389 );
390 }
391
392 #[test]
393 fn test_adjust_sst_metadata_and_prefilter_cache_caps_independently() {
394 let mut config = MitoConfig::default();
395
396 config.adjust_buffer_and_cache_size(ReadableSize::gb(1));
397 assert_eq!(ReadableSize::mb(128), config.sst_meta_cache_size);
398 assert_eq!(ReadableSize::mb(32), config.prefilter_result_cache_size);
399
400 config.adjust_buffer_and_cache_size(ReadableSize::gb(64));
401 assert_eq!(ReadableSize::mb(512), config.sst_meta_cache_size);
402 assert_eq!(ReadableSize::mb(128), config.prefilter_result_cache_size);
403 }
404
405 #[test]
406 fn test_experimental_series_scan_v2_config() {
407 assert!(MitoConfig::default().experimental_series_scan_v2);
408
409 let config: MitoConfig = toml::from_str("experimental_series_scan_v2 = false").unwrap();
410 assert!(!config.experimental_series_scan_v2);
411 }
412}
413
414#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Default)]
416#[serde(rename_all = "snake_case")]
417pub enum IndexBuildMode {
418 #[default]
420 Sync,
421 Async,
423}
424
425#[serde_as]
426#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
427#[serde(default)]
428pub struct IndexConfig {
429 pub aux_path: String,
439
440 pub staging_size: ReadableSize,
442 #[serde(with = "humantime_serde")]
446 pub staging_ttl: Option<Duration>,
447
448 pub build_mode: IndexBuildMode,
450
451 pub write_buffer_size: ReadableSize,
453
454 pub metadata_cache_size: ReadableSize,
456 pub content_cache_size: ReadableSize,
458 pub content_cache_page_size: ReadableSize,
460 pub result_cache_size: ReadableSize,
462}
463
464impl Default for IndexConfig {
465 fn default() -> Self {
466 Self {
467 aux_path: String::new(),
468 staging_size: ReadableSize::gb(2),
469 staging_ttl: Some(Duration::from_secs(7 * 24 * 60 * 60)),
470 build_mode: IndexBuildMode::default(),
471 write_buffer_size: ReadableSize::mb(8),
472 metadata_cache_size: ReadableSize::mb(64),
473 content_cache_size: ReadableSize::mb(128),
474 content_cache_page_size: ReadableSize::kb(64),
475 result_cache_size: ReadableSize::mb(128),
476 }
477 }
478}
479
480impl IndexConfig {
481 pub fn sanitize(
482 &mut self,
483 data_home: &str,
484 inverted_index: &InvertedIndexConfig,
485 ) -> Result<()> {
486 #[allow(deprecated)]
487 if self.aux_path.is_empty() && !inverted_index.intermediate_path.is_empty() {
488 self.aux_path.clone_from(&inverted_index.intermediate_path);
489 warn!(
490 "`inverted_index.intermediate_path` is deprecated, use
491 `index.aux_path` instead. Set `index.aux_path` to {}",
492 &inverted_index.intermediate_path
493 )
494 }
495 if self.aux_path.is_empty() {
496 let path = Path::new(data_home).join("index_intermediate");
497 self.aux_path = path.as_os_str().to_string_lossy().to_string();
498 }
499
500 if self.write_buffer_size < MULTIPART_UPLOAD_MINIMUM_SIZE {
501 self.write_buffer_size = MULTIPART_UPLOAD_MINIMUM_SIZE;
502 warn!(
503 "Sanitize index write buffer size to {}",
504 self.write_buffer_size
505 );
506 }
507
508 if self.staging_ttl.map(|ttl| ttl.is_zero()).unwrap_or(false) {
509 self.staging_ttl = None;
510 }
511
512 Ok(())
513 }
514
515 pub fn adjust_buffer_and_cache_size(&mut self, sys_memory: ReadableSize) {
516 let cache_size = cmp::min(sys_memory / MEM_CACHE_SIZE_FACTOR, ReadableSize::mb(128));
517 self.result_cache_size = cmp::min(self.result_cache_size, cache_size);
518 self.content_cache_size = cmp::min(self.content_cache_size, cache_size);
519
520 let metadata_cache_size = cmp::min(
521 sys_memory / INDEX_METADATA_CACHE_SIZE_FACTOR,
522 ReadableSize::mb(64),
523 );
524 self.metadata_cache_size = cmp::min(self.metadata_cache_size, metadata_cache_size);
525 }
526}
527
528#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Default)]
530#[serde(rename_all = "snake_case")]
531pub enum Mode {
532 #[default]
534 Auto,
535 Disable,
537}
538
539impl Mode {
540 pub fn disabled(&self) -> bool {
542 matches!(self, Mode::Disable)
543 }
544
545 pub fn auto(&self) -> bool {
547 matches!(self, Mode::Auto)
548 }
549}
550
551#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
553#[serde(rename_all = "snake_case")]
554pub enum MemoryThreshold {
555 #[default]
557 Auto,
558 Unlimited,
560 #[serde(untagged)]
562 Size(ReadableSize),
563}
564
565#[serde_as]
567#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
568#[serde(default)]
569pub struct InvertedIndexConfig {
570 pub create_on_flush: Mode,
572 pub create_on_compaction: Mode,
574 pub apply_on_query: Mode,
576
577 pub mem_threshold_on_create: MemoryThreshold,
579
580 #[deprecated = "use [IndexConfig::aux_path] instead"]
581 #[serde(skip_serializing)]
582 pub intermediate_path: String,
583
584 #[deprecated = "use [IndexConfig::write_buffer_size] instead"]
585 #[serde(skip_serializing)]
586 pub write_buffer_size: ReadableSize,
587}
588
589impl Default for InvertedIndexConfig {
590 #[allow(deprecated)]
591 fn default() -> Self {
592 Self {
593 create_on_flush: Mode::Auto,
594 create_on_compaction: Mode::Auto,
595 apply_on_query: Mode::Auto,
596 mem_threshold_on_create: MemoryThreshold::Auto,
597 write_buffer_size: ReadableSize::mb(8),
598 intermediate_path: String::new(),
599 }
600 }
601}
602
603impl InvertedIndexConfig {
604 pub fn mem_threshold_on_create(&self) -> Option<usize> {
605 match self.mem_threshold_on_create {
606 MemoryThreshold::Auto => {
607 if let Some(sys_memory) = get_total_memory_readable() {
608 Some((sys_memory / INDEX_CREATE_MEM_THRESHOLD_FACTOR).as_bytes() as usize)
609 } else {
610 Some(ReadableSize::mb(64).as_bytes() as usize)
611 }
612 }
613 MemoryThreshold::Unlimited => None,
614 MemoryThreshold::Size(size) => Some(size.as_bytes() as usize),
615 }
616 }
617}
618
619#[serde_as]
621#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
622#[serde(default)]
623pub struct FulltextIndexConfig {
624 pub create_on_flush: Mode,
626 pub create_on_compaction: Mode,
628 pub apply_on_query: Mode,
630 pub mem_threshold_on_create: MemoryThreshold,
632 pub compress: bool,
634}
635
636impl Default for FulltextIndexConfig {
637 fn default() -> Self {
638 Self {
639 create_on_flush: Mode::Auto,
640 create_on_compaction: Mode::Auto,
641 apply_on_query: Mode::Auto,
642 mem_threshold_on_create: MemoryThreshold::Auto,
643 compress: true,
644 }
645 }
646}
647
648impl FulltextIndexConfig {
649 pub fn mem_threshold_on_create(&self) -> usize {
650 match self.mem_threshold_on_create {
651 MemoryThreshold::Auto => {
652 if let Some(sys_memory) = get_total_memory_readable() {
653 (sys_memory / INDEX_CREATE_MEM_THRESHOLD_FACTOR).as_bytes() as _
654 } else {
655 ReadableSize::mb(64).as_bytes() as _
656 }
657 }
658 MemoryThreshold::Unlimited => usize::MAX,
659 MemoryThreshold::Size(size) => size.as_bytes() as _,
660 }
661 }
662}
663
664#[serde_as]
666#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
667#[serde(default)]
668pub struct BloomFilterConfig {
669 pub create_on_flush: Mode,
671 pub create_on_compaction: Mode,
673 pub apply_on_query: Mode,
675 pub mem_threshold_on_create: MemoryThreshold,
677}
678
679impl Default for BloomFilterConfig {
680 fn default() -> Self {
681 Self {
682 create_on_flush: Mode::Auto,
683 create_on_compaction: Mode::Auto,
684 apply_on_query: Mode::Auto,
685 mem_threshold_on_create: MemoryThreshold::Auto,
686 }
687 }
688}
689
690impl BloomFilterConfig {
691 pub fn mem_threshold_on_create(&self) -> Option<usize> {
692 match self.mem_threshold_on_create {
693 MemoryThreshold::Auto => {
694 if let Some(sys_memory) = get_total_memory_readable() {
695 Some((sys_memory / INDEX_CREATE_MEM_THRESHOLD_FACTOR).as_bytes() as usize)
696 } else {
697 Some(ReadableSize::mb(64).as_bytes() as usize)
698 }
699 }
700 MemoryThreshold::Unlimited => None,
701 MemoryThreshold::Size(size) => Some(size.as_bytes() as usize),
702 }
703 }
704}
705
706#[cfg(feature = "vector_index")]
708#[serde_as]
709#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
710#[serde(default)]
711pub struct VectorIndexConfig {
712 pub create_on_flush: Mode,
714 pub create_on_compaction: Mode,
716 pub apply_on_query: Mode,
718 pub mem_threshold_on_create: MemoryThreshold,
720}
721
722#[cfg(feature = "vector_index")]
723impl Default for VectorIndexConfig {
724 fn default() -> Self {
725 Self {
726 create_on_flush: Mode::Auto,
727 create_on_compaction: Mode::Auto,
728 apply_on_query: Mode::Auto,
729 mem_threshold_on_create: MemoryThreshold::Auto,
730 }
731 }
732}
733
734#[cfg(feature = "vector_index")]
735impl VectorIndexConfig {
736 pub fn mem_threshold_on_create(&self) -> Option<usize> {
737 match self.mem_threshold_on_create {
738 MemoryThreshold::Auto => {
739 if let Some(sys_memory) = get_total_memory_readable() {
740 Some((sys_memory / INDEX_CREATE_MEM_THRESHOLD_FACTOR).as_bytes() as usize)
741 } else {
742 Some(ReadableSize::mb(64).as_bytes() as usize)
743 }
744 }
745 MemoryThreshold::Unlimited => None,
746 MemoryThreshold::Size(size) => Some(size.as_bytes() as usize),
747 }
748 }
749}
750
751fn divide_num_cpus(divisor: usize) -> usize {
753 debug_assert!(divisor > 0);
754 let cores = get_total_cpu_cores();
755 debug_assert!(cores > 0);
756
757 cores.div_ceil(divisor)
758}