Skip to main content

mito2/
config.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Configurations.
16
17use 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);
35const DEFAULT_SERIES_INDEX_MAINTENANCE_INTERVAL: Duration = Duration::from_secs(5 * 60);
36/// Default maximum number of SST files to scan concurrently.
37pub(crate) const DEFAULT_MAX_CONCURRENT_SCAN_FILES: usize = 384;
38
39// Use `1/GLOBAL_WRITE_BUFFER_SIZE_FACTOR` of OS memory as global write buffer size in default mode
40const GLOBAL_WRITE_BUFFER_SIZE_FACTOR: u64 = 8;
41/// Use `1/SST_META_CACHE_SIZE_FACTOR` of OS memory size as SST meta cache size in default mode
42const SST_META_CACHE_SIZE_FACTOR: u64 = 8;
43/// Use `1/PREFILTER_RESULT_CACHE_SIZE_FACTOR` of OS memory size as prefilter result cache size in default mode
44const PREFILTER_RESULT_CACHE_SIZE_FACTOR: u64 = 32;
45/// Use `1/INDEX_METADATA_CACHE_SIZE_FACTOR` of OS memory size as index metadata cache size in default mode
46const INDEX_METADATA_CACHE_SIZE_FACTOR: u64 = 32;
47/// Use `1/MEM_CACHE_SIZE_FACTOR` of OS memory size as mem cache size in default mode
48const MEM_CACHE_SIZE_FACTOR: u64 = 16;
49/// Use `1/PAGE_CACHE_SIZE_FACTOR` of OS memory size as page cache size in default mode
50const PAGE_CACHE_SIZE_FACTOR: u64 = 8;
51/// Use `1/INDEX_CREATE_MEM_THRESHOLD_FACTOR` of OS memory size as mem threshold for creating index
52const INDEX_CREATE_MEM_THRESHOLD_FACTOR: u64 = 16;
53
54/// Memory accounted against the shared scan limit.
55#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
56#[serde(rename_all = "snake_case")]
57pub enum ScanMemoryMode {
58    /// Count projected Parquet columns' uncompressed row-group sizes.
59    #[default]
60    Basic,
61    /// Also count cumulative streamed batches.
62    Extended,
63}
64
65/// Fetch option timeout
66pub(crate) const FETCH_OPTION_TIMEOUT: Duration = Duration::from_secs(3);
67
68/// Configuration for [MitoEngine](crate::engine::MitoEngine).
69/// Before using the config, call [`MitoConfig::sanitize()`] to validate and normalize it.
70#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
71#[serde(default)]
72pub struct MitoConfig {
73    // Worker configs:
74    /// Number of region workers (default: 1/2 of cpu cores).
75    /// Sets to 0 to use the default value.
76    pub num_workers: usize,
77    /// Request channel size of each worker (default 128).
78    pub worker_channel_size: usize,
79    /// Max batch size for a worker to handle requests (default 64).
80    pub worker_request_batch_size: usize,
81
82    // Manifest configs:
83    /// Number of meta action updated to trigger a new checkpoint
84    /// for the manifest (default 10).
85    pub manifest_checkpoint_distance: u64,
86    /// Number of removed files to keep in manifest's `removed_files` field before also
87    /// remove them from `removed_files`. Mostly for debugging purpose.
88    /// If set to 0, it will only use `keep_removed_file_ttl` to decide when to remove files
89    /// from `removed_files` field.
90    pub experimental_manifest_keep_removed_file_count: usize,
91    /// How long to keep removed files in the `removed_files` field of manifest
92    /// after they are removed from manifest.
93    /// files will only be removed from `removed_files` field
94    /// if both `keep_removed_file_count` and `keep_removed_file_ttl` is reached.
95    #[serde(with = "humantime_serde")]
96    pub experimental_manifest_keep_removed_file_ttl: Duration,
97    /// Whether to compress manifest and checkpoint file by gzip (default false).
98    pub compress_manifest: bool,
99
100    // Background job configs:
101    /// Max number of running background index build jobs (default: 1/8 of cpu cores).
102    pub max_background_index_builds: usize,
103    /// Under development; do not enable. Whether to enable series indexes (default false).
104    /// Indexes are stored on the local filesystem under `{data_home}/series_index`.
105    pub experimental_enable_series_index: bool,
106    /// Approximate series and range index size limit in open regions (default: 5 GiB).
107    /// Workers share a periodically refreshed estimate and skip maintenance when full.
108    /// In-flight reconciliation can exceed the limit; closed-region files are not counted.
109    pub experimental_series_index_max_size: ReadableSize,
110    /// Whether to build and query range indexes when series indexes are enabled (default false).
111    /// Obsolete range-index metadata and files are still cleaned up when disabled.
112    pub experimental_enable_range_index: bool,
113    /// Interval between series-index maintenance runs (default 5 min). Zero uses the default.
114    #[serde(with = "humantime_serde")]
115    pub experimental_series_index_maintenance_interval: Duration,
116    /// Under development; do not enable. Requested minimum bucket width for series indexes.
117    /// It is rounded up to an exact multiple of each region's compaction time window.
118    #[serde(with = "humantime_serde")]
119    pub experimental_series_index_bucket_width: Duration,
120    /// Max number of running background flush jobs (default: 1/2 of cpu cores).
121    pub max_background_flushes: usize,
122    /// Max number of running background compaction jobs (default: 1/4 of cpu cores).
123    pub max_background_compactions: usize,
124    /// Max number of running background purge jobs (default: number of cpu cores).
125    pub max_background_purges: usize,
126    /// Memory budget for compaction tasks.
127    /// Supports absolute size (e.g., "2GiB", "512MB") or percentage of system memory (e.g., "50%").
128    /// Setting it to 0 or "unlimited" disables the limit.
129    pub experimental_compaction_memory_limit: MemoryLimit,
130    /// Behavior when compaction cannot acquire memory from the budget.
131    pub experimental_compaction_on_exhausted: OnExhaustedPolicy,
132
133    // Flush configs:
134    /// Interval to auto flush a region if it has not flushed yet (default 10 min).
135    #[serde(with = "humantime_serde")]
136    pub auto_flush_interval: Duration,
137    /// Global write buffer size threshold to trigger flush.
138    pub global_write_buffer_size: ReadableSize,
139    /// Global write buffer size threshold to reject write requests.
140    pub global_write_buffer_reject_size: ReadableSize,
141    /// Default write buffer size for each region. Regions stall at this size and
142    /// reject writes at twice this size. Setting it to 0 disables both limits
143    /// unless the table specifies `write_buffer_size`.
144    pub default_region_write_buffer_size: ReadableSize,
145
146    // Cache configs:
147    /// Cache size for SST metadata. Setting it to 0 to disable the cache.
148    pub sst_meta_cache_size: ReadableSize,
149    /// Cache size for vectors and arrow arrays. Setting it to 0 to disable the cache.
150    pub vector_cache_size: ReadableSize,
151    /// Cache size for pages of SST row groups. Setting it to 0 to disable the cache.
152    pub page_cache_size: ReadableSize,
153    /// Cache size for time series selector (e.g. `last_value()`). Setting it to 0 to disable the cache.
154    pub selector_result_cache_size: ReadableSize,
155    /// Cache size for flat range scan results. Setting it to 0 to disable the cache.
156    pub range_result_cache_size: ReadableSize,
157    /// Cache size for prefilter results. Setting it to 0 to disable the cache.
158    pub prefilter_result_cache_size: ReadableSize,
159    /// Whether to enable the write cache.
160    pub enable_write_cache: bool,
161    /// File system path for write cache dir's root, defaults to `{data_home}`.
162    pub write_cache_path: String,
163    /// Capacity for write cache.
164    pub write_cache_size: ReadableSize,
165    /// TTL for write cache.
166    #[serde(with = "humantime_serde")]
167    pub write_cache_ttl: Option<Duration>,
168    /// Preload index (puffin) files into cache on region open (default: true).
169    pub preload_index_cache: bool,
170    /// Percentage of write cache capacity allocated for index (puffin) files (default: 20).
171    /// The remaining capacity is used for data (parquet) files.
172    /// Must be between 0 and 100 (exclusive).
173    pub index_cache_percent: u8,
174    /// Enable background downloading of files to the local cache when accessed during queries (default: true).
175    /// When enabled, files will be asynchronously downloaded to improve performance for subsequent reads.
176    pub enable_refill_cache_on_read: bool,
177    /// Capacity for manifest cache (default: 256MB).
178    pub manifest_cache_size: ReadableSize,
179
180    // Other configs:
181    /// Buffer size for SST writing.
182    // TODO(#9240): Support per-object-store write buffer sizes for backend-specific tuning.
183    pub sst_write_buffer_size: ReadableSize,
184    /// Maximum number of SST files to scan concurrently when memory accounting
185    /// does not cover the scan or the limit is disabled (default 384).
186    pub max_concurrent_scan_files: usize,
187    /// Whether to allow stale entries read during replay.
188    pub allow_stale_entries: bool,
189    /// Memory limit for table scans across all queries (default: 25% of system memory).
190    /// Setting it to 0 or "unlimited" disables the limit.
191    /// Supports absolute size (e.g., "2GB") or percentage of system memory (e.g., "50%").
192    pub scan_memory_limit: MemoryLimit,
193    /// Scan memory accounting: basic counts projected Parquet row groups;
194    /// extended also counts cumulative streamed batches. Limited reservations round up to KB.
195    /// Internal series merges use the query runtime pool in both modes.
196    pub scan_memory_mode: ScanMemoryMode,
197    /// Scan memory exhaustion always fails immediately (default: `fail`).
198    /// Legacy `wait` and `wait(<duration>)` values are accepted but deprecated:
199    /// sanitization warns and replaces them with [`OnExhaustedPolicy::Fail`].
200    pub scan_memory_on_exhausted: OnExhaustedPolicy,
201
202    /// Index configs.
203    pub index: IndexConfig,
204    /// Inverted index configs.
205    pub inverted_index: InvertedIndexConfig,
206    /// Full-text index configs.
207    pub fulltext_index: FulltextIndexConfig,
208    /// Bloom filter index configs.
209    pub bloom_filter_index: BloomFilterConfig,
210
211    /// Minimum time interval between two compactions.
212    /// To align with the old behavior, the default value is 0 (no restrictions).
213    #[serde(with = "humantime_serde")]
214    pub min_compaction_interval: Duration,
215    /// Whether to schedule compaction after applying a region edit.
216    pub schedule_compaction_after_edit: bool,
217
218    /// Whether to enable flat format as the default SST format.
219    /// When enabled, forces using BulkMemtable and BulkMemtableBuilder.
220    pub default_flat_format: bool,
221
222    /// Whether to enable the experimental two-phase mode for eligible metric series scans.
223    pub experimental_series_scan_v2: bool,
224
225    pub gc: GcConfig,
226}
227
228impl Default for MitoConfig {
229    fn default() -> Self {
230        let mut mito_config = MitoConfig {
231            num_workers: divide_num_cpus(2),
232            worker_channel_size: 128,
233            worker_request_batch_size: 64,
234            manifest_checkpoint_distance: 10,
235            experimental_manifest_keep_removed_file_count: 256,
236            experimental_manifest_keep_removed_file_ttl: Duration::from_secs(60 * 60),
237            compress_manifest: false,
238            max_background_index_builds: divide_num_cpus(8),
239            experimental_enable_series_index: false,
240            experimental_series_index_max_size: ReadableSize::gb(5),
241            experimental_enable_range_index: false,
242            experimental_series_index_maintenance_interval:
243                DEFAULT_SERIES_INDEX_MAINTENANCE_INTERVAL,
244            experimental_series_index_bucket_width: Duration::from_secs(5 * 24 * 60 * 60),
245            max_background_flushes: divide_num_cpus(2),
246            max_background_compactions: divide_num_cpus(4),
247            max_background_purges: get_total_cpu_cores(),
248            experimental_compaction_memory_limit: MemoryLimit::Unlimited,
249            experimental_compaction_on_exhausted: OnExhaustedPolicy::default(),
250            auto_flush_interval: Duration::from_secs(10 * 60),
251            global_write_buffer_size: ReadableSize::gb(1),
252            global_write_buffer_reject_size: ReadableSize::gb(2),
253            default_region_write_buffer_size: ReadableSize::mb(0),
254            sst_meta_cache_size: ReadableSize::mb(128),
255            vector_cache_size: ReadableSize::mb(512),
256            page_cache_size: ReadableSize::mb(512),
257            selector_result_cache_size: ReadableSize::mb(512),
258            range_result_cache_size: ReadableSize::mb(512),
259            prefilter_result_cache_size: ReadableSize::mb(128),
260            enable_write_cache: false,
261            write_cache_path: String::new(),
262            write_cache_size: ReadableSize::gb(5),
263            write_cache_ttl: None,
264            preload_index_cache: true,
265            index_cache_percent: DEFAULT_INDEX_CACHE_PERCENT,
266            enable_refill_cache_on_read: true,
267            manifest_cache_size: ReadableSize::mb(256),
268            sst_write_buffer_size: DEFAULT_WRITE_BUFFER_SIZE,
269            max_concurrent_scan_files: DEFAULT_MAX_CONCURRENT_SCAN_FILES,
270            allow_stale_entries: false,
271            scan_memory_limit: MemoryLimit::Percentage(25),
272            scan_memory_mode: ScanMemoryMode::Basic,
273            scan_memory_on_exhausted: OnExhaustedPolicy::Fail,
274            index: IndexConfig::default(),
275            inverted_index: InvertedIndexConfig::default(),
276            fulltext_index: FulltextIndexConfig::default(),
277            bloom_filter_index: BloomFilterConfig::default(),
278            min_compaction_interval: Duration::from_secs(0),
279            schedule_compaction_after_edit: true,
280            default_flat_format: true,
281            // FIXME(#9435): Keep v2 opt-in while long-range scan memory usage is investigated.
282            experimental_series_scan_v2: false,
283            gc: GcConfig::default(),
284        };
285
286        // Adjust buffer and cache size according to system memory if we can.
287        if let Some(sys_memory) = get_total_memory_readable() {
288            mito_config.adjust_buffer_and_cache_size(sys_memory);
289        }
290
291        mito_config
292    }
293}
294
295impl MitoConfig {
296    /// Sanitize incorrect configurations.
297    ///
298    /// Returns an error if there is a configuration that unable to sanitize.
299    pub fn sanitize(&mut self, data_home: &str) -> Result<()> {
300        if matches!(
301            self.scan_memory_on_exhausted,
302            OnExhaustedPolicy::Wait { .. }
303        ) {
304            warn!(
305                "scan_memory_on_exhausted = {:?} is deprecated and is treated as fail; \
306                 scan memory exhaustion now fails immediately. Set scan_memory_on_exhausted = \"fail\"",
307                self.scan_memory_on_exhausted
308            );
309            self.scan_memory_on_exhausted = OnExhaustedPolicy::Fail;
310        }
311
312        if self.experimental_enable_series_index {
313            snafu::ensure!(
314                self.experimental_series_index_max_size.as_bytes() >= 1024,
315                crate::error::InvalidConfigSnafu {
316                    reason: "experimental_series_index_max_size must be at least 1KiB"
317                }
318            );
319        }
320        // Use default value if `num_workers` is 0.
321        if self.num_workers == 0 {
322            self.num_workers = divide_num_cpus(2);
323        }
324
325        // Sanitize channel size.
326        if self.worker_channel_size == 0 {
327            warn!("Sanitize channel size 0 to 1");
328            self.worker_channel_size = 1;
329        }
330
331        if self.max_background_flushes == 0 {
332            warn!(
333                "Sanitize max background flushes 0 to {}",
334                divide_num_cpus(2)
335            );
336            self.max_background_flushes = divide_num_cpus(2);
337        }
338        if self.max_background_compactions == 0 {
339            warn!(
340                "Sanitize max background compactions 0 to {}",
341                divide_num_cpus(4)
342            );
343            self.max_background_compactions = divide_num_cpus(4);
344        }
345        if self.max_background_purges == 0 {
346            let cpu_cores = get_total_cpu_cores();
347            warn!("Sanitize max background purges 0 to {}", cpu_cores);
348            self.max_background_purges = cpu_cores;
349        }
350
351        if self
352            .experimental_series_index_maintenance_interval
353            .is_zero()
354        {
355            warn!("Sanitize series-index maintenance interval 0 to 5 minutes");
356            self.experimental_series_index_maintenance_interval =
357                DEFAULT_SERIES_INDEX_MAINTENANCE_INTERVAL;
358        }
359
360        if self.global_write_buffer_reject_size <= self.global_write_buffer_size {
361            self.global_write_buffer_reject_size = self.global_write_buffer_size * 2;
362            warn!(
363                "Sanitize global write buffer reject size to {}",
364                self.global_write_buffer_reject_size
365            );
366        }
367
368        if self.sst_write_buffer_size < MULTIPART_UPLOAD_MINIMUM_SIZE {
369            self.sst_write_buffer_size = MULTIPART_UPLOAD_MINIMUM_SIZE;
370            warn!(
371                "Sanitize sst write buffer size to {}",
372                self.sst_write_buffer_size
373            );
374        }
375
376        // Sets write cache path if it is empty.
377        if self.write_cache_path.trim().is_empty() {
378            self.write_cache_path = data_home.to_string();
379        }
380
381        // Validate index_cache_percent is within valid range (0, 100)
382        if self.index_cache_percent == 0 || self.index_cache_percent >= 100 {
383            warn!(
384                "Invalid index_cache_percent {}, resetting to default {}",
385                self.index_cache_percent, DEFAULT_INDEX_CACHE_PERCENT
386            );
387            self.index_cache_percent = DEFAULT_INDEX_CACHE_PERCENT;
388        }
389
390        self.index.sanitize(data_home, &self.inverted_index)?;
391
392        Ok(())
393    }
394
395    fn adjust_buffer_and_cache_size(&mut self, sys_memory: ReadableSize) {
396        // shouldn't be greater than 1G in default mode.
397        let global_write_buffer_size = cmp::min(
398            sys_memory / GLOBAL_WRITE_BUFFER_SIZE_FACTOR,
399            ReadableSize::gb(1),
400        );
401        // Use 2x of global write buffer size as global write buffer reject size.
402        let global_write_buffer_reject_size = global_write_buffer_size * 2;
403        // Page-index-bearing SST metadata can be much larger than footers alone.
404        // Keep the auto-sized default bounded, but allow a larger warm working set.
405        let sst_meta_cache_size = cmp::min(
406            sys_memory / SST_META_CACHE_SIZE_FACTOR,
407            ReadableSize::mb(512),
408        );
409        let prefilter_result_cache_size = cmp::min(
410            sys_memory / PREFILTER_RESULT_CACHE_SIZE_FACTOR,
411            ReadableSize::mb(128),
412        );
413        // shouldn't be greater than 512MB in default mode.
414        let mem_cache_size = cmp::min(sys_memory / MEM_CACHE_SIZE_FACTOR, ReadableSize::mb(512));
415        let page_cache_size = sys_memory / PAGE_CACHE_SIZE_FACTOR;
416
417        self.global_write_buffer_size = global_write_buffer_size;
418        self.global_write_buffer_reject_size = global_write_buffer_reject_size;
419        self.sst_meta_cache_size = sst_meta_cache_size;
420        self.vector_cache_size = mem_cache_size;
421        self.page_cache_size = page_cache_size;
422        self.selector_result_cache_size = mem_cache_size;
423        self.range_result_cache_size = mem_cache_size;
424        // Use a smaller cache size because prefilter result usually should be small.
425        self.prefilter_result_cache_size = prefilter_result_cache_size;
426
427        self.index.adjust_buffer_and_cache_size(sys_memory);
428    }
429
430    /// Enable write cache.
431    #[cfg(test)]
432    pub fn enable_write_cache(
433        mut self,
434        path: String,
435        size: ReadableSize,
436        ttl: Option<Duration>,
437    ) -> Self {
438        self.enable_write_cache = true;
439        self.write_cache_path = path;
440        self.write_cache_size = size;
441        self.write_cache_ttl = ttl;
442        self
443    }
444}
445
446#[cfg(test)]
447mod tests {
448    use super::*;
449
450    #[test]
451    fn test_scan_memory_policy_upgrade() {
452        for policy in [
453            None,
454            Some("fail"),
455            Some("skip"),
456            Some("wait"),
457            Some("wait(30s)"),
458        ] {
459            let input = policy
460                .map(|policy| format!("scan_memory_on_exhausted = {policy:?}"))
461                .unwrap_or_default();
462            let mut config: MitoConfig = toml::from_str(&input).unwrap();
463            let compaction_policy = config.experimental_compaction_on_exhausted;
464            config.sanitize("/data").unwrap();
465            assert_eq!(OnExhaustedPolicy::Fail, config.scan_memory_on_exhausted);
466            assert_eq!(
467                compaction_policy,
468                config.experimental_compaction_on_exhausted
469            );
470            let serialized = toml::to_string(&config).unwrap();
471            assert!(serialized.contains("scan_memory_on_exhausted = \"fail\""));
472            config.sanitize("/data").unwrap();
473            assert_eq!(serialized, toml::to_string(&config).unwrap());
474        }
475        for policy in ["unknown", "wait(invalid)", "wait()"] {
476            assert!(
477                toml::from_str::<MitoConfig>(&format!("scan_memory_on_exhausted = {policy:?}"))
478                    .is_err()
479            );
480        }
481    }
482
483    #[test]
484    fn test_default_auto_flush_interval() {
485        assert_eq!(
486            Duration::from_secs(10 * 60),
487            MitoConfig::default().auto_flush_interval
488        );
489    }
490
491    #[test]
492    fn test_adjust_sst_metadata_and_prefilter_cache_caps_independently() {
493        let mut config = MitoConfig::default();
494
495        config.adjust_buffer_and_cache_size(ReadableSize::gb(1));
496        assert_eq!(ReadableSize::mb(128), config.sst_meta_cache_size);
497        assert_eq!(ReadableSize::mb(32), config.prefilter_result_cache_size);
498
499        config.adjust_buffer_and_cache_size(ReadableSize::gb(64));
500        assert_eq!(ReadableSize::mb(512), config.sst_meta_cache_size);
501        assert_eq!(ReadableSize::mb(128), config.prefilter_result_cache_size);
502    }
503
504    #[test]
505    fn test_series_index_config() {
506        assert!(!MitoConfig::default().experimental_enable_series_index);
507        assert!(!MitoConfig::default().experimental_enable_range_index);
508        assert_eq!(
509            MitoConfig::default().experimental_series_index_bucket_width,
510            Duration::from_secs(5 * 24 * 60 * 60)
511        );
512        let mut config: MitoConfig = toml::from_str(
513            "experimental_enable_series_index = true
514             experimental_enable_range_index = false
515             experimental_series_index_max_size = '64MiB'
516             experimental_series_index_maintenance_interval = '30s'
517             experimental_series_index_bucket_width = '2days'",
518        )
519        .unwrap();
520        config.sanitize("/data").unwrap();
521        assert!(config.experimental_enable_series_index);
522        assert_eq!(
523            ReadableSize::mb(64),
524            config.experimental_series_index_max_size
525        );
526        assert!(!config.experimental_enable_range_index);
527        assert_eq!(
528            config.experimental_series_index_maintenance_interval,
529            Duration::from_secs(30)
530        );
531        assert_eq!(
532            config.experimental_series_index_bucket_width,
533            Duration::from_secs(2 * 24 * 60 * 60)
534        );
535        let restored: MitoConfig = toml::from_str(&toml::to_string(&config).unwrap()).unwrap();
536        assert_eq!(config, restored);
537        config.experimental_series_index_max_size = ReadableSize(1023);
538        assert!(config.sanitize("/data").is_err());
539        config.experimental_series_index_max_size = ReadableSize(1024);
540        config.sanitize("/data").unwrap();
541
542        let mut config: MitoConfig =
543            toml::from_str("experimental_series_index_maintenance_interval = '0s'").unwrap();
544        config.sanitize("/data").unwrap();
545        assert_eq!(
546            config.experimental_series_index_maintenance_interval,
547            MitoConfig::default().experimental_series_index_maintenance_interval
548        );
549    }
550
551    #[test]
552    fn test_range_index_config() {
553        let config: MitoConfig = toml::from_str("experimental_enable_series_index = true").unwrap();
554        assert!(!config.experimental_enable_range_index);
555        for enabled in [false, true] {
556            let config: MitoConfig =
557                toml::from_str(&format!("experimental_enable_range_index = {enabled}")).unwrap();
558            assert_eq!(enabled, config.experimental_enable_range_index);
559            let restored: MitoConfig = toml::from_str(&toml::to_string(&config).unwrap()).unwrap();
560            assert_eq!(config, restored);
561        }
562    }
563}
564
565/// Index build mode.
566#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Default)]
567#[serde(rename_all = "snake_case")]
568pub enum IndexBuildMode {
569    /// Build index synchronously.
570    #[default]
571    Sync,
572    /// Build index asynchronously.
573    Async,
574}
575
576#[serde_as]
577#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
578#[serde(default)]
579pub struct IndexConfig {
580    /// Auxiliary directory path for the index in filesystem, used to
581    /// store intermediate files for creating the index and staging files
582    /// for searching the index, defaults to `{data_home}/index_intermediate`.
583    ///
584    /// This path contains two subdirectories:
585    /// - `__intm`: for storing intermediate files used during creating index.
586    /// - `staging`: for storing staging files used during searching index.
587    ///
588    /// The default name for this directory is `index_intermediate` for backward compatibility.
589    pub aux_path: String,
590
591    /// The max capacity of the staging directory.
592    pub staging_size: ReadableSize,
593    /// The TTL of the staging directory.
594    /// Defaults to 7 days.
595    /// Setting it to "0s" to disable TTL.
596    #[serde(with = "humantime_serde")]
597    pub staging_ttl: Option<Duration>,
598
599    /// Index Build Mode
600    pub build_mode: IndexBuildMode,
601
602    /// Write buffer size for creating the index.
603    pub write_buffer_size: ReadableSize,
604
605    /// Cache size for metadata of puffin files. Setting it to 0 to disable the cache.
606    pub metadata_cache_size: ReadableSize,
607    /// Cache size for inverted index content. Setting it to 0 to disable the cache.
608    pub content_cache_size: ReadableSize,
609    /// Page size for inverted index content.
610    pub content_cache_page_size: ReadableSize,
611    /// Cache size for index result. Setting it to 0 to disable the cache.
612    pub result_cache_size: ReadableSize,
613}
614
615impl Default for IndexConfig {
616    fn default() -> Self {
617        Self {
618            aux_path: String::new(),
619            staging_size: ReadableSize::gb(2),
620            staging_ttl: Some(Duration::from_secs(7 * 24 * 60 * 60)),
621            build_mode: IndexBuildMode::default(),
622            write_buffer_size: ReadableSize::mb(8),
623            metadata_cache_size: ReadableSize::mb(64),
624            content_cache_size: ReadableSize::mb(128),
625            content_cache_page_size: ReadableSize::kb(64),
626            result_cache_size: ReadableSize::mb(128),
627        }
628    }
629}
630
631impl IndexConfig {
632    pub fn sanitize(
633        &mut self,
634        data_home: &str,
635        inverted_index: &InvertedIndexConfig,
636    ) -> Result<()> {
637        #[allow(deprecated)]
638        if self.aux_path.is_empty() && !inverted_index.intermediate_path.is_empty() {
639            self.aux_path.clone_from(&inverted_index.intermediate_path);
640            warn!(
641                "`inverted_index.intermediate_path` is deprecated, use
642                 `index.aux_path` instead. Set `index.aux_path` to {}",
643                &inverted_index.intermediate_path
644            )
645        }
646        if self.aux_path.is_empty() {
647            let path = Path::new(data_home).join("index_intermediate");
648            self.aux_path = path.as_os_str().to_string_lossy().to_string();
649        }
650
651        if self.write_buffer_size < MULTIPART_UPLOAD_MINIMUM_SIZE {
652            self.write_buffer_size = MULTIPART_UPLOAD_MINIMUM_SIZE;
653            warn!(
654                "Sanitize index write buffer size to {}",
655                self.write_buffer_size
656            );
657        }
658
659        if self.staging_ttl.map(|ttl| ttl.is_zero()).unwrap_or(false) {
660            self.staging_ttl = None;
661        }
662
663        Ok(())
664    }
665
666    pub fn adjust_buffer_and_cache_size(&mut self, sys_memory: ReadableSize) {
667        let cache_size = cmp::min(sys_memory / MEM_CACHE_SIZE_FACTOR, ReadableSize::mb(128));
668        self.result_cache_size = cmp::min(self.result_cache_size, cache_size);
669        self.content_cache_size = cmp::min(self.content_cache_size, cache_size);
670
671        let metadata_cache_size = cmp::min(
672            sys_memory / INDEX_METADATA_CACHE_SIZE_FACTOR,
673            ReadableSize::mb(64),
674        );
675        self.metadata_cache_size = cmp::min(self.metadata_cache_size, metadata_cache_size);
676    }
677}
678
679/// Operational mode for certain actions.
680#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Default)]
681#[serde(rename_all = "snake_case")]
682pub enum Mode {
683    /// The action is performed automatically based on internal criteria.
684    #[default]
685    Auto,
686    /// The action is explicitly disabled.
687    Disable,
688}
689
690impl Mode {
691    /// Whether the action is disabled.
692    pub fn disabled(&self) -> bool {
693        matches!(self, Mode::Disable)
694    }
695
696    /// Whether the action is automatic.
697    pub fn auto(&self) -> bool {
698        matches!(self, Mode::Auto)
699    }
700}
701
702/// Memory threshold for performing certain actions.
703#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
704#[serde(rename_all = "snake_case")]
705pub enum MemoryThreshold {
706    /// Automatically determine the threshold based on internal criteria.
707    #[default]
708    Auto,
709    /// Unlimited memory.
710    Unlimited,
711    /// Fixed memory threshold.
712    #[serde(untagged)]
713    Size(ReadableSize),
714}
715
716/// Configuration options for the inverted index.
717#[serde_as]
718#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
719#[serde(default)]
720pub struct InvertedIndexConfig {
721    /// Whether to create the index on flush: automatically or never.
722    pub create_on_flush: Mode,
723    /// Whether to create the index on compaction: automatically or never.
724    pub create_on_compaction: Mode,
725    /// Whether to apply the index on query: automatically or never.
726    pub apply_on_query: Mode,
727
728    /// Memory threshold for performing an external sort during index creation.
729    pub mem_threshold_on_create: MemoryThreshold,
730
731    #[deprecated = "use [IndexConfig::aux_path] instead"]
732    #[serde(skip_serializing)]
733    pub intermediate_path: String,
734
735    #[deprecated = "use [IndexConfig::write_buffer_size] instead"]
736    #[serde(skip_serializing)]
737    pub write_buffer_size: ReadableSize,
738}
739
740impl Default for InvertedIndexConfig {
741    #[allow(deprecated)]
742    fn default() -> Self {
743        Self {
744            create_on_flush: Mode::Auto,
745            create_on_compaction: Mode::Auto,
746            apply_on_query: Mode::Auto,
747            mem_threshold_on_create: MemoryThreshold::Auto,
748            write_buffer_size: ReadableSize::mb(8),
749            intermediate_path: String::new(),
750        }
751    }
752}
753
754impl InvertedIndexConfig {
755    pub fn mem_threshold_on_create(&self) -> Option<usize> {
756        match self.mem_threshold_on_create {
757            MemoryThreshold::Auto => {
758                if let Some(sys_memory) = get_total_memory_readable() {
759                    Some((sys_memory / INDEX_CREATE_MEM_THRESHOLD_FACTOR).as_bytes() as usize)
760                } else {
761                    Some(ReadableSize::mb(64).as_bytes() as usize)
762                }
763            }
764            MemoryThreshold::Unlimited => None,
765            MemoryThreshold::Size(size) => Some(size.as_bytes() as usize),
766        }
767    }
768}
769
770/// Configuration options for the full-text index.
771#[serde_as]
772#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
773#[serde(default)]
774pub struct FulltextIndexConfig {
775    /// Whether to create the index on flush: automatically or never.
776    pub create_on_flush: Mode,
777    /// Whether to create the index on compaction: automatically or never.
778    pub create_on_compaction: Mode,
779    /// Whether to apply the index on query: automatically or never.
780    pub apply_on_query: Mode,
781    /// Memory threshold for creating the index.
782    pub mem_threshold_on_create: MemoryThreshold,
783    /// Whether to compress the index data.
784    pub compress: bool,
785}
786
787impl Default for FulltextIndexConfig {
788    fn default() -> Self {
789        Self {
790            create_on_flush: Mode::Auto,
791            create_on_compaction: Mode::Auto,
792            apply_on_query: Mode::Auto,
793            mem_threshold_on_create: MemoryThreshold::Auto,
794            compress: true,
795        }
796    }
797}
798
799impl FulltextIndexConfig {
800    pub fn mem_threshold_on_create(&self) -> usize {
801        match self.mem_threshold_on_create {
802            MemoryThreshold::Auto => {
803                if let Some(sys_memory) = get_total_memory_readable() {
804                    (sys_memory / INDEX_CREATE_MEM_THRESHOLD_FACTOR).as_bytes() as _
805                } else {
806                    ReadableSize::mb(64).as_bytes() as _
807                }
808            }
809            MemoryThreshold::Unlimited => usize::MAX,
810            MemoryThreshold::Size(size) => size.as_bytes() as _,
811        }
812    }
813}
814
815/// Configuration options for the bloom filter.
816#[serde_as]
817#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
818#[serde(default)]
819pub struct BloomFilterConfig {
820    /// Whether to create the index on flush: automatically or never.
821    pub create_on_flush: Mode,
822    /// Whether to create the index on compaction: automatically or never.
823    pub create_on_compaction: Mode,
824    /// Whether to apply the index on query: automatically or never.
825    pub apply_on_query: Mode,
826    /// Memory threshold for creating the index.
827    pub mem_threshold_on_create: MemoryThreshold,
828}
829
830impl Default for BloomFilterConfig {
831    fn default() -> Self {
832        Self {
833            create_on_flush: Mode::Auto,
834            create_on_compaction: Mode::Auto,
835            apply_on_query: Mode::Auto,
836            mem_threshold_on_create: MemoryThreshold::Auto,
837        }
838    }
839}
840
841impl BloomFilterConfig {
842    pub fn mem_threshold_on_create(&self) -> Option<usize> {
843        match self.mem_threshold_on_create {
844            MemoryThreshold::Auto => {
845                if let Some(sys_memory) = get_total_memory_readable() {
846                    Some((sys_memory / INDEX_CREATE_MEM_THRESHOLD_FACTOR).as_bytes() as usize)
847                } else {
848                    Some(ReadableSize::mb(64).as_bytes() as usize)
849                }
850            }
851            MemoryThreshold::Unlimited => None,
852            MemoryThreshold::Size(size) => Some(size.as_bytes() as usize),
853        }
854    }
855}
856
857/// Divide cpu num by a non-zero `divisor` and returns at least 1.
858fn divide_num_cpus(divisor: usize) -> usize {
859    debug_assert!(divisor > 0);
860    let cores = get_total_cpu_cores();
861    debug_assert!(cores > 0);
862
863    cores.div_ceil(divisor)
864}