risingwave_meta/hummock/compaction/
compaction_config.rs

1// Copyright 2022 RisingWave Labs
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
15use risingwave_common::config::CompactionConfig as CompactionConfigOpt;
16use risingwave_common::config::meta::default::compaction_config;
17use risingwave_hummock_sdk::filter_utils::{
18    parse_sstable_filter_layout, parse_sstable_filter_type,
19};
20use risingwave_pb::hummock::CompactionConfig;
21use risingwave_pb::hummock::compaction_config::CompactionMode;
22
23pub struct CompactionConfigBuilder {
24    config: CompactionConfig,
25}
26
27impl CompactionConfigBuilder {
28    pub fn new() -> Self {
29        #[expect(deprecated)]
30        Self {
31            config: CompactionConfig {
32                max_bytes_for_level_base: compaction_config::max_bytes_for_level_base(),
33                max_bytes_for_level_multiplier: compaction_config::max_bytes_for_level_multiplier(),
34                max_level: compaction_config::max_level() as u64,
35                max_compaction_bytes: compaction_config::max_compaction_bytes(),
36                sub_level_max_compaction_bytes: compaction_config::sub_level_max_compaction_bytes(),
37                level0_tier_compact_file_number: compaction_config::level0_tier_compact_file_number(
38                ),
39                target_file_size_base: compaction_config::target_file_size_base(),
40                compaction_mode: CompactionMode::Range as i32,
41                compression_algorithm: compaction_config::compression_algorithm_vec(
42                    compaction_config::max_level(),
43                ),
44                sstable_filter_type: compaction_config::sstable_filter_type(),
45                sstable_filter_layout: compaction_config::sstable_filter_layout(),
46                compaction_filter_mask: compaction_config::compaction_filter_mask(),
47                max_sub_compaction: compaction_config::max_sub_compaction(),
48                max_space_reclaim_bytes: compaction_config::max_space_reclaim_bytes(),
49                split_by_state_table: false,
50                split_weight_by_vnode: 0,
51                level0_stop_write_threshold_sub_level_number:
52                    compaction_config::level0_stop_write_threshold_sub_level_number(),
53                // This configure variable shall be larger than level0_tier_compact_file_number, and
54                // it shall meet the following condition:
55                //    level0_max_compact_file_number * target_file_size_base >
56                // max_bytes_for_level_base
57                level0_max_compact_file_number: compaction_config::level0_max_compact_file_number(),
58                level0_sub_level_compact_level_count:
59                    compaction_config::level0_sub_level_compact_level_count(),
60                level0_overlapping_sub_level_compact_level_count:
61                    compaction_config::level0_overlapping_sub_level_compact_level_count(),
62                tombstone_reclaim_ratio: compaction_config::tombstone_reclaim_ratio(),
63                enable_emergency_picker: compaction_config::enable_emergency_picker(),
64                max_l0_compact_level_count: Some(compaction_config::max_l0_compact_level_count()),
65                sst_allowed_trivial_move_min_size: Some(
66                    compaction_config::sst_allowed_trivial_move_min_size(),
67                ),
68                disable_auto_group_scheduling: Some(
69                    compaction_config::disable_auto_group_scheduling(),
70                ),
71                max_overlapping_level_size: Some(compaction_config::max_overlapping_level_size()),
72                sst_allowed_trivial_move_max_count: Some(
73                    compaction_config::sst_allowed_trivial_move_max_count(),
74                ),
75                emergency_level0_sst_file_count: Some(
76                    compaction_config::emergency_level0_sst_file_count(),
77                ),
78                emergency_level0_sub_level_partition: Some(
79                    compaction_config::emergency_level0_sub_level_partition(),
80                ),
81                level0_stop_write_threshold_max_sst_count: Some(
82                    compaction_config::level0_stop_write_threshold_max_sst_count(),
83                ),
84                level0_stop_write_threshold_max_size: Some(
85                    compaction_config::level0_stop_write_threshold_max_size(),
86                ),
87                enable_optimize_l0_interval_selection: Some(
88                    compaction_config::enable_optimize_l0_interval_selection(),
89                ),
90                vnode_aligned_level_size_threshold: None,
91                max_kv_count_for_xor16: compaction_config::blocked_xor_filter_kv_count_threshold(),
92                max_vnode_key_range_bytes: compaction_config::max_vnode_key_range_bytes(),
93            },
94        }
95    }
96
97    pub fn with_config(config: CompactionConfig) -> Self {
98        Self { config }
99    }
100
101    pub fn with_opt(opt: &CompactionConfigOpt) -> Self {
102        Self::new()
103            .max_bytes_for_level_base(opt.max_bytes_for_level_base)
104            .max_bytes_for_level_multiplier(opt.max_bytes_for_level_multiplier)
105            .max_compaction_bytes(opt.max_compaction_bytes)
106            .sub_level_max_compaction_bytes(opt.sub_level_max_compaction_bytes)
107            .level0_tier_compact_file_number(opt.level0_tier_compact_file_number)
108            .target_file_size_base(opt.target_file_size_base)
109            .compaction_filter_mask(opt.compaction_filter_mask)
110            .max_sub_compaction(opt.max_sub_compaction)
111            .level0_stop_write_threshold_sub_level_number(
112                opt.level0_stop_write_threshold_sub_level_number,
113            )
114            .level0_sub_level_compact_level_count(opt.level0_sub_level_compact_level_count)
115            .level0_overlapping_sub_level_compact_level_count(
116                opt.level0_overlapping_sub_level_compact_level_count,
117            )
118            .max_space_reclaim_bytes(opt.max_space_reclaim_bytes)
119            .level0_max_compact_file_number(opt.level0_max_compact_file_number)
120            .tombstone_reclaim_ratio(opt.tombstone_reclaim_ratio)
121            .enable_emergency_picker(opt.enable_emergency_picker)
122            .max_level(opt.max_level as u64)
123            .sst_allowed_trivial_move_min_size(Some(opt.sst_allowed_trivial_move_min_size))
124            .sst_allowed_trivial_move_max_count(Some(opt.sst_allowed_trivial_move_max_count))
125            .max_l0_compact_level_count(Some(opt.max_l0_compact_level_count))
126            .disable_auto_group_scheduling(Some(opt.disable_auto_group_scheduling))
127            .max_overlapping_level_size(Some(opt.max_overlapping_level_size))
128            .emergency_level0_sst_file_count(Some(opt.emergency_level0_sst_file_count))
129            .emergency_level0_sub_level_partition(Some(opt.emergency_level0_sub_level_partition))
130            .level0_stop_write_threshold_max_sst_count(Some(
131                opt.level0_stop_write_threshold_max_sst_count,
132            ))
133            .level0_stop_write_threshold_max_size(Some(opt.level0_stop_write_threshold_max_size))
134            .enable_optimize_l0_interval_selection(Some(opt.enable_optimize_l0_interval_selection))
135            .max_kv_count_for_xor16(opt.blocked_xor_filter_kv_count_threshold)
136            .max_vnode_key_range_bytes(opt.max_vnode_key_range_bytes)
137            .sstable_filter_type(opt.sstable_filter_type.clone())
138            .sstable_filter_layout(opt.sstable_filter_layout.clone())
139    }
140
141    pub fn build(self) -> CompactionConfig {
142        let mut config = self.config;
143        if let Err(reason) = validate_compaction_config(&config) {
144            // Avoid crashing later in compaction task planning due to invalid per-level filter
145            // configs. Fallback to legacy behavior (xor16 + auto) and keep other fields unchanged.
146            tracing::warn!(
147                "Bad compaction config: {}. Falling back to default sstable filter type/layout.",
148                reason
149            );
150            config.sstable_filter_type.clear();
151            config.sstable_filter_layout.clear();
152        }
153        config
154    }
155}
156
157/// Returns Ok if `config` is valid,
158/// or the reason why it's invalid.
159pub fn validate_compaction_config(config: &CompactionConfig) -> Result<(), String> {
160    let sub_level_number_threshold_min = 1;
161    if config.level0_stop_write_threshold_sub_level_number < sub_level_number_threshold_min {
162        return Err(format!(
163            "{} is too small for level0_stop_write_threshold_sub_level_number, expect >= {}",
164            config.level0_stop_write_threshold_sub_level_number, sub_level_number_threshold_min
165        ));
166    }
167    if !config.sstable_filter_type.is_empty() {
168        if config.sstable_filter_type.len() < config.max_level as usize + 1 {
169            return Err(format!(
170                "sstable_filter_type must provide at least {} entries for max_level {}",
171                config.max_level + 1,
172                config.max_level
173            ));
174        }
175        for filter_type in &config.sstable_filter_type {
176            parse_sstable_filter_type(filter_type)?;
177        }
178    }
179
180    if !config.sstable_filter_layout.is_empty() {
181        if config.sstable_filter_layout.len() < config.max_level as usize + 1 {
182            return Err(format!(
183                "sstable_filter_layout must provide at least {} entries for max_level {}",
184                config.max_level + 1,
185                config.max_level
186            ));
187        }
188        for layout in &config.sstable_filter_layout {
189            parse_sstable_filter_layout(layout)?;
190        }
191    }
192    Ok(())
193}
194
195impl Default for CompactionConfigBuilder {
196    fn default() -> Self {
197        Self::new()
198    }
199}
200
201macro_rules! builder_field {
202    ($( $name:ident: $type:ty ),* ,) => {
203        impl CompactionConfigBuilder {
204            $(
205                pub fn $name(mut self, v:$type) -> Self {
206                    self.config.$name = v;
207                    self
208                }
209            )*
210        }
211    }
212}
213
214builder_field! {
215    max_bytes_for_level_base: u64,
216    max_bytes_for_level_multiplier: u64,
217    max_level: u64,
218    max_compaction_bytes: u64,
219    sub_level_max_compaction_bytes: u64,
220    level0_tier_compact_file_number: u64,
221    compaction_mode: i32,
222    compression_algorithm: Vec<String>,
223    sstable_filter_type: Vec<String>,
224    sstable_filter_layout: Vec<String>,
225    compaction_filter_mask: u32,
226    target_file_size_base: u64,
227    max_sub_compaction: u32,
228    max_space_reclaim_bytes: u64,
229    level0_stop_write_threshold_sub_level_number: u64,
230    level0_max_compact_file_number: u64,
231    level0_sub_level_compact_level_count: u32,
232    level0_overlapping_sub_level_compact_level_count: u32,
233    tombstone_reclaim_ratio: u32,
234    enable_emergency_picker: bool,
235    sst_allowed_trivial_move_min_size: Option<u64>,
236    sst_allowed_trivial_move_max_count: Option<u32>,
237    max_l0_compact_level_count: Option<u32>,
238    disable_auto_group_scheduling: Option<bool>,
239    max_overlapping_level_size: Option<u64>,
240    emergency_level0_sst_file_count: Option<u32>,
241    emergency_level0_sub_level_partition: Option<u32>,
242    level0_stop_write_threshold_max_sst_count: Option<u32>,
243    level0_stop_write_threshold_max_size: Option<u64>,
244    enable_optimize_l0_interval_selection: Option<bool>,
245    max_kv_count_for_xor16: Option<u64>,
246    max_vnode_key_range_bytes: Option<u64>,
247}
248
249#[cfg(test)]
250mod tests {
251    use risingwave_common::config::RwConfig;
252
253    use super::{CompactionConfigBuilder, validate_compaction_config};
254
255    #[test]
256    fn test_compaction_config_from_toml() {
257        let config: RwConfig = toml::from_str(
258            r#"
259            [meta.compaction_config]
260            level0_max_compact_file_number = 20
261            max_l0_compact_level_count = 10
262            max_compaction_bytes = 1073741824
263            "#,
264        )
265        .unwrap();
266        let config = CompactionConfigBuilder::with_opt(&config.meta.compaction_config).build();
267
268        assert_eq!(config.level0_max_compact_file_number, 20);
269        assert_eq!(config.max_l0_compact_level_count, Some(10));
270        assert_eq!(config.max_compaction_bytes, 1_073_741_824);
271    }
272
273    #[test]
274    fn test_compaction_config_flags_from_toml() {
275        let config: RwConfig = toml::from_str(
276            r#"
277            [meta.compaction_config]
278            enable_emergency_picker = false
279            disable_auto_group_scheduling = true
280            "#,
281        )
282        .unwrap();
283        let config = CompactionConfigBuilder::with_opt(&config.meta.compaction_config).build();
284
285        assert!(!config.enable_emergency_picker);
286        assert_eq!(config.disable_auto_group_scheduling, Some(true));
287    }
288
289    #[test]
290    fn test_compaction_config_defaults_from_toml() {
291        let config: RwConfig = toml::from_str("").unwrap();
292        let config = CompactionConfigBuilder::with_opt(&config.meta.compaction_config).build();
293
294        assert_eq!(config, CompactionConfigBuilder::new().build());
295    }
296
297    #[test]
298    fn test_validate_compaction_config_accepts_missing_filter_config() {
299        let mut config = CompactionConfigBuilder::new().build();
300        config.sstable_filter_type.clear();
301        config.sstable_filter_layout.clear();
302        assert!(validate_compaction_config(&config).is_ok());
303    }
304
305    #[test]
306    fn test_validate_compaction_config_rejects_invalid_filter_config() {
307        let mut config = CompactionConfigBuilder::new().build();
308        config.sstable_filter_type = vec!["xor16".to_owned()];
309        assert!(validate_compaction_config(&config).is_err());
310
311        let mut config = CompactionConfigBuilder::new().build();
312        config.sstable_filter_layout = vec!["auto".to_owned()];
313        assert!(validate_compaction_config(&config).is_err());
314
315        let mut config = CompactionConfigBuilder::new().build();
316        config.sstable_filter_layout = vec!["auto".to_owned(); config.max_level as usize + 1];
317        config.sstable_filter_layout[0] = "unknown".to_owned();
318        assert!(validate_compaction_config(&config).is_err());
319    }
320
321    #[test]
322    fn test_validate_compaction_config_accepts_none_filter_type() {
323        let mut config = CompactionConfigBuilder::new().build();
324        config.sstable_filter_type = vec!["none".to_owned(); config.max_level as usize + 1];
325        assert!(validate_compaction_config(&config).is_ok());
326    }
327}
risingwave_meta/hummock/compaction/compaction_config.rs

risingwave_meta/hummock/compaction/
compaction_config.rs
