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 gc: GcConfig,
192}
193
194impl Default for MitoConfig {
195 fn default() -> Self {
196 let mut mito_config = MitoConfig {
197 num_workers: divide_num_cpus(2),
198 worker_channel_size: 128,
199 worker_request_batch_size: 64,
200 manifest_checkpoint_distance: 10,
201 experimental_manifest_keep_removed_file_count: 256,
202 experimental_manifest_keep_removed_file_ttl: Duration::from_secs(60 * 60),
203 compress_manifest: false,
204 max_background_index_builds: divide_num_cpus(8),
205 max_background_flushes: divide_num_cpus(2),
206 max_background_compactions: divide_num_cpus(4),
207 max_background_purges: get_total_cpu_cores(),
208 experimental_compaction_memory_limit: MemoryLimit::Unlimited,
209 experimental_compaction_on_exhausted: OnExhaustedPolicy::default(),
210 auto_flush_interval: Duration::from_secs(30 * 60),
211 global_write_buffer_size: ReadableSize::gb(1),
212 global_write_buffer_reject_size: ReadableSize::gb(2),
213 default_region_write_buffer_size: ReadableSize::mb(0),
214 sst_meta_cache_size: ReadableSize::mb(128),
215 vector_cache_size: ReadableSize::mb(512),
216 page_cache_size: ReadableSize::mb(512),
217 selector_result_cache_size: ReadableSize::mb(512),
218 range_result_cache_size: ReadableSize::mb(512),
219 prefilter_result_cache_size: ReadableSize::mb(128),
220 enable_write_cache: false,
221 write_cache_path: String::new(),
222 write_cache_size: ReadableSize::gb(5),
223 write_cache_ttl: None,
224 preload_index_cache: true,
225 index_cache_percent: DEFAULT_INDEX_CACHE_PERCENT,
226 enable_refill_cache_on_read: true,
227 manifest_cache_size: ReadableSize::mb(256),
228 sst_write_buffer_size: DEFAULT_WRITE_BUFFER_SIZE,
229 max_concurrent_scan_files: DEFAULT_MAX_CONCURRENT_SCAN_FILES,
230 allow_stale_entries: false,
231 scan_memory_limit: MemoryLimit::default(),
232 scan_memory_on_exhausted: OnExhaustedPolicy::Fail,
233 index: IndexConfig::default(),
234 inverted_index: InvertedIndexConfig::default(),
235 fulltext_index: FulltextIndexConfig::default(),
236 bloom_filter_index: BloomFilterConfig::default(),
237 #[cfg(feature = "vector_index")]
238 vector_index: VectorIndexConfig::default(),
239 min_compaction_interval: Duration::from_secs(0),
240 schedule_compaction_after_edit: true,
241 default_flat_format: true,
242 gc: GcConfig::default(),
243 };
244
245 if let Some(sys_memory) = get_total_memory_readable() {
247 mito_config.adjust_buffer_and_cache_size(sys_memory);
248 }
249
250 mito_config
251 }
252}
253
254impl MitoConfig {
255 pub fn sanitize(&mut self, data_home: &str) -> Result<()> {
259 if self.num_workers == 0 {
261 self.num_workers = divide_num_cpus(2);
262 }
263
264 if self.worker_channel_size == 0 {
266 warn!("Sanitize channel size 0 to 1");
267 self.worker_channel_size = 1;
268 }
269
270 if self.max_background_flushes == 0 {
271 warn!(
272 "Sanitize max background flushes 0 to {}",
273 divide_num_cpus(2)
274 );
275 self.max_background_flushes = divide_num_cpus(2);
276 }
277 if self.max_background_compactions == 0 {
278 warn!(
279 "Sanitize max background compactions 0 to {}",
280 divide_num_cpus(4)
281 );
282 self.max_background_compactions = divide_num_cpus(4);
283 }
284 if self.max_background_purges == 0 {
285 let cpu_cores = get_total_cpu_cores();
286 warn!("Sanitize max background purges 0 to {}", cpu_cores);
287 self.max_background_purges = cpu_cores;
288 }
289
290 if self.global_write_buffer_reject_size <= self.global_write_buffer_size {
291 self.global_write_buffer_reject_size = self.global_write_buffer_size * 2;
292 warn!(
293 "Sanitize global write buffer reject size to {}",
294 self.global_write_buffer_reject_size
295 );
296 }
297
298 if self.sst_write_buffer_size < MULTIPART_UPLOAD_MINIMUM_SIZE {
299 self.sst_write_buffer_size = MULTIPART_UPLOAD_MINIMUM_SIZE;
300 warn!(
301 "Sanitize sst write buffer size to {}",
302 self.sst_write_buffer_size
303 );
304 }
305
306 if self.write_cache_path.trim().is_empty() {
308 self.write_cache_path = data_home.to_string();
309 }
310
311 if self.index_cache_percent == 0 || self.index_cache_percent >= 100 {
313 warn!(
314 "Invalid index_cache_percent {}, resetting to default {}",
315 self.index_cache_percent, DEFAULT_INDEX_CACHE_PERCENT
316 );
317 self.index_cache_percent = DEFAULT_INDEX_CACHE_PERCENT;
318 }
319
320 self.index.sanitize(data_home, &self.inverted_index)?;
321
322 Ok(())
323 }
324
325 fn adjust_buffer_and_cache_size(&mut self, sys_memory: ReadableSize) {
326 let global_write_buffer_size = cmp::min(
328 sys_memory / GLOBAL_WRITE_BUFFER_SIZE_FACTOR,
329 ReadableSize::gb(1),
330 );
331 let global_write_buffer_reject_size = global_write_buffer_size * 2;
333 let sst_meta_cache_size = cmp::min(
336 sys_memory / SST_META_CACHE_SIZE_FACTOR,
337 ReadableSize::mb(512),
338 );
339 let prefilter_result_cache_size = cmp::min(
340 sys_memory / PREFILTER_RESULT_CACHE_SIZE_FACTOR,
341 ReadableSize::mb(128),
342 );
343 let mem_cache_size = cmp::min(sys_memory / MEM_CACHE_SIZE_FACTOR, ReadableSize::mb(512));
345 let page_cache_size = sys_memory / PAGE_CACHE_SIZE_FACTOR;
346
347 self.global_write_buffer_size = global_write_buffer_size;
348 self.global_write_buffer_reject_size = global_write_buffer_reject_size;
349 self.sst_meta_cache_size = sst_meta_cache_size;
350 self.vector_cache_size = mem_cache_size;
351 self.page_cache_size = page_cache_size;
352 self.selector_result_cache_size = mem_cache_size;
353 self.range_result_cache_size = mem_cache_size;
354 self.prefilter_result_cache_size = prefilter_result_cache_size;
356
357 self.index.adjust_buffer_and_cache_size(sys_memory);
358 }
359
360 #[cfg(test)]
362 pub fn enable_write_cache(
363 mut self,
364 path: String,
365 size: ReadableSize,
366 ttl: Option<Duration>,
367 ) -> Self {
368 self.enable_write_cache = true;
369 self.write_cache_path = path;
370 self.write_cache_size = size;
371 self.write_cache_ttl = ttl;
372 self
373 }
374}
375
376#[cfg(test)]
377mod tests {
378 use super::*;
379
380 #[test]
381 fn test_adjust_sst_metadata_and_prefilter_cache_caps_independently() {
382 let mut config = MitoConfig::default();
383
384 config.adjust_buffer_and_cache_size(ReadableSize::gb(1));
385 assert_eq!(ReadableSize::mb(128), config.sst_meta_cache_size);
386 assert_eq!(ReadableSize::mb(32), config.prefilter_result_cache_size);
387
388 config.adjust_buffer_and_cache_size(ReadableSize::gb(64));
389 assert_eq!(ReadableSize::mb(512), config.sst_meta_cache_size);
390 assert_eq!(ReadableSize::mb(128), config.prefilter_result_cache_size);
391 }
392}
393
394#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Default)]
396#[serde(rename_all = "snake_case")]
397pub enum IndexBuildMode {
398 #[default]
400 Sync,
401 Async,
403}
404
405#[serde_as]
406#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
407#[serde(default)]
408pub struct IndexConfig {
409 pub aux_path: String,
419
420 pub staging_size: ReadableSize,
422 #[serde(with = "humantime_serde")]
426 pub staging_ttl: Option<Duration>,
427
428 pub build_mode: IndexBuildMode,
430
431 pub write_buffer_size: ReadableSize,
433
434 pub metadata_cache_size: ReadableSize,
436 pub content_cache_size: ReadableSize,
438 pub content_cache_page_size: ReadableSize,
440 pub result_cache_size: ReadableSize,
442}
443
444impl Default for IndexConfig {
445 fn default() -> Self {
446 Self {
447 aux_path: String::new(),
448 staging_size: ReadableSize::gb(2),
449 staging_ttl: Some(Duration::from_secs(7 * 24 * 60 * 60)),
450 build_mode: IndexBuildMode::default(),
451 write_buffer_size: ReadableSize::mb(8),
452 metadata_cache_size: ReadableSize::mb(64),
453 content_cache_size: ReadableSize::mb(128),
454 content_cache_page_size: ReadableSize::kb(64),
455 result_cache_size: ReadableSize::mb(128),
456 }
457 }
458}
459
460impl IndexConfig {
461 pub fn sanitize(
462 &mut self,
463 data_home: &str,
464 inverted_index: &InvertedIndexConfig,
465 ) -> Result<()> {
466 #[allow(deprecated)]
467 if self.aux_path.is_empty() && !inverted_index.intermediate_path.is_empty() {
468 self.aux_path.clone_from(&inverted_index.intermediate_path);
469 warn!(
470 "`inverted_index.intermediate_path` is deprecated, use
471 `index.aux_path` instead. Set `index.aux_path` to {}",
472 &inverted_index.intermediate_path
473 )
474 }
475 if self.aux_path.is_empty() {
476 let path = Path::new(data_home).join("index_intermediate");
477 self.aux_path = path.as_os_str().to_string_lossy().to_string();
478 }
479
480 if self.write_buffer_size < MULTIPART_UPLOAD_MINIMUM_SIZE {
481 self.write_buffer_size = MULTIPART_UPLOAD_MINIMUM_SIZE;
482 warn!(
483 "Sanitize index write buffer size to {}",
484 self.write_buffer_size
485 );
486 }
487
488 if self.staging_ttl.map(|ttl| ttl.is_zero()).unwrap_or(false) {
489 self.staging_ttl = None;
490 }
491
492 Ok(())
493 }
494
495 pub fn adjust_buffer_and_cache_size(&mut self, sys_memory: ReadableSize) {
496 let cache_size = cmp::min(sys_memory / MEM_CACHE_SIZE_FACTOR, ReadableSize::mb(128));
497 self.result_cache_size = cmp::min(self.result_cache_size, cache_size);
498 self.content_cache_size = cmp::min(self.content_cache_size, cache_size);
499
500 let metadata_cache_size = cmp::min(
501 sys_memory / INDEX_METADATA_CACHE_SIZE_FACTOR,
502 ReadableSize::mb(64),
503 );
504 self.metadata_cache_size = cmp::min(self.metadata_cache_size, metadata_cache_size);
505 }
506}
507
508#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Default)]
510#[serde(rename_all = "snake_case")]
511pub enum Mode {
512 #[default]
514 Auto,
515 Disable,
517}
518
519impl Mode {
520 pub fn disabled(&self) -> bool {
522 matches!(self, Mode::Disable)
523 }
524
525 pub fn auto(&self) -> bool {
527 matches!(self, Mode::Auto)
528 }
529}
530
531#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
533#[serde(rename_all = "snake_case")]
534pub enum MemoryThreshold {
535 #[default]
537 Auto,
538 Unlimited,
540 #[serde(untagged)]
542 Size(ReadableSize),
543}
544
545#[serde_as]
547#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
548#[serde(default)]
549pub struct InvertedIndexConfig {
550 pub create_on_flush: Mode,
552 pub create_on_compaction: Mode,
554 pub apply_on_query: Mode,
556
557 pub mem_threshold_on_create: MemoryThreshold,
559
560 #[deprecated = "use [IndexConfig::aux_path] instead"]
561 #[serde(skip_serializing)]
562 pub intermediate_path: String,
563
564 #[deprecated = "use [IndexConfig::write_buffer_size] instead"]
565 #[serde(skip_serializing)]
566 pub write_buffer_size: ReadableSize,
567}
568
569impl Default for InvertedIndexConfig {
570 #[allow(deprecated)]
571 fn default() -> Self {
572 Self {
573 create_on_flush: Mode::Auto,
574 create_on_compaction: Mode::Auto,
575 apply_on_query: Mode::Auto,
576 mem_threshold_on_create: MemoryThreshold::Auto,
577 write_buffer_size: ReadableSize::mb(8),
578 intermediate_path: String::new(),
579 }
580 }
581}
582
583impl InvertedIndexConfig {
584 pub fn mem_threshold_on_create(&self) -> Option<usize> {
585 match self.mem_threshold_on_create {
586 MemoryThreshold::Auto => {
587 if let Some(sys_memory) = get_total_memory_readable() {
588 Some((sys_memory / INDEX_CREATE_MEM_THRESHOLD_FACTOR).as_bytes() as usize)
589 } else {
590 Some(ReadableSize::mb(64).as_bytes() as usize)
591 }
592 }
593 MemoryThreshold::Unlimited => None,
594 MemoryThreshold::Size(size) => Some(size.as_bytes() as usize),
595 }
596 }
597}
598
599#[serde_as]
601#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
602#[serde(default)]
603pub struct FulltextIndexConfig {
604 pub create_on_flush: Mode,
606 pub create_on_compaction: Mode,
608 pub apply_on_query: Mode,
610 pub mem_threshold_on_create: MemoryThreshold,
612 pub compress: bool,
614}
615
616impl Default for FulltextIndexConfig {
617 fn default() -> Self {
618 Self {
619 create_on_flush: Mode::Auto,
620 create_on_compaction: Mode::Auto,
621 apply_on_query: Mode::Auto,
622 mem_threshold_on_create: MemoryThreshold::Auto,
623 compress: true,
624 }
625 }
626}
627
628impl FulltextIndexConfig {
629 pub fn mem_threshold_on_create(&self) -> usize {
630 match self.mem_threshold_on_create {
631 MemoryThreshold::Auto => {
632 if let Some(sys_memory) = get_total_memory_readable() {
633 (sys_memory / INDEX_CREATE_MEM_THRESHOLD_FACTOR).as_bytes() as _
634 } else {
635 ReadableSize::mb(64).as_bytes() as _
636 }
637 }
638 MemoryThreshold::Unlimited => usize::MAX,
639 MemoryThreshold::Size(size) => size.as_bytes() as _,
640 }
641 }
642}
643
644#[serde_as]
646#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
647#[serde(default)]
648pub struct BloomFilterConfig {
649 pub create_on_flush: Mode,
651 pub create_on_compaction: Mode,
653 pub apply_on_query: Mode,
655 pub mem_threshold_on_create: MemoryThreshold,
657}
658
659impl Default for BloomFilterConfig {
660 fn default() -> Self {
661 Self {
662 create_on_flush: Mode::Auto,
663 create_on_compaction: Mode::Auto,
664 apply_on_query: Mode::Auto,
665 mem_threshold_on_create: MemoryThreshold::Auto,
666 }
667 }
668}
669
670impl BloomFilterConfig {
671 pub fn mem_threshold_on_create(&self) -> Option<usize> {
672 match self.mem_threshold_on_create {
673 MemoryThreshold::Auto => {
674 if let Some(sys_memory) = get_total_memory_readable() {
675 Some((sys_memory / INDEX_CREATE_MEM_THRESHOLD_FACTOR).as_bytes() as usize)
676 } else {
677 Some(ReadableSize::mb(64).as_bytes() as usize)
678 }
679 }
680 MemoryThreshold::Unlimited => None,
681 MemoryThreshold::Size(size) => Some(size.as_bytes() as usize),
682 }
683 }
684}
685
686#[cfg(feature = "vector_index")]
688#[serde_as]
689#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
690#[serde(default)]
691pub struct VectorIndexConfig {
692 pub create_on_flush: Mode,
694 pub create_on_compaction: Mode,
696 pub apply_on_query: Mode,
698 pub mem_threshold_on_create: MemoryThreshold,
700}
701
702#[cfg(feature = "vector_index")]
703impl Default for VectorIndexConfig {
704 fn default() -> Self {
705 Self {
706 create_on_flush: Mode::Auto,
707 create_on_compaction: Mode::Auto,
708 apply_on_query: Mode::Auto,
709 mem_threshold_on_create: MemoryThreshold::Auto,
710 }
711 }
712}
713
714#[cfg(feature = "vector_index")]
715impl VectorIndexConfig {
716 pub fn mem_threshold_on_create(&self) -> Option<usize> {
717 match self.mem_threshold_on_create {
718 MemoryThreshold::Auto => {
719 if let Some(sys_memory) = get_total_memory_readable() {
720 Some((sys_memory / INDEX_CREATE_MEM_THRESHOLD_FACTOR).as_bytes() as usize)
721 } else {
722 Some(ReadableSize::mb(64).as_bytes() as usize)
723 }
724 }
725 MemoryThreshold::Unlimited => None,
726 MemoryThreshold::Size(size) => Some(size.as_bytes() as usize),
727 }
728 }
729}
730
731fn divide_num_cpus(divisor: usize) -> usize {
733 debug_assert!(divisor > 0);
734 let cores = get_total_cpu_cores();
735 debug_assert!(cores > 0);
736
737 cores.div_ceil(divisor)
738}