risingwave_meta/hummock/compaction/
compaction_config.rs1use risingwave_common::config::CompactionConfig as CompactionConfigOpt;
16use risingwave_common::config::default::compaction_config;
17use risingwave_pb::hummock::CompactionConfig;
18use risingwave_pb::hummock::compaction_config::CompactionMode;
19
20pub struct CompactionConfigBuilder {
21 config: CompactionConfig,
22}
23
24impl CompactionConfigBuilder {
25 pub fn new() -> Self {
26 Self {
27 config: CompactionConfig {
28 max_bytes_for_level_base: compaction_config::max_bytes_for_level_base(),
29 max_bytes_for_level_multiplier: compaction_config::max_bytes_for_level_multiplier(),
30 max_level: compaction_config::max_level() as u64,
31 max_compaction_bytes: compaction_config::max_compaction_bytes(),
32 sub_level_max_compaction_bytes: compaction_config::sub_level_max_compaction_bytes(),
33 level0_tier_compact_file_number: compaction_config::level0_tier_compact_file_number(
34 ),
35 target_file_size_base: compaction_config::target_file_size_base(),
36 compaction_mode: CompactionMode::Range as i32,
37 compression_algorithm: vec![
41 "None".to_owned(),
42 "None".to_owned(),
43 "None".to_owned(),
44 "Lz4".to_owned(),
45 "Lz4".to_owned(),
46 "Zstd".to_owned(),
47 "Zstd".to_owned(),
48 ],
49 compaction_filter_mask: compaction_config::compaction_filter_mask(),
50 max_sub_compaction: compaction_config::max_sub_compaction(),
51 max_space_reclaim_bytes: compaction_config::max_space_reclaim_bytes(),
52 split_by_state_table: false,
53 split_weight_by_vnode: 0,
54 level0_stop_write_threshold_sub_level_number:
55 compaction_config::level0_stop_write_threshold_sub_level_number(),
56 level0_max_compact_file_number: compaction_config::level0_max_compact_file_number(),
61 level0_sub_level_compact_level_count:
62 compaction_config::level0_sub_level_compact_level_count(),
63 level0_overlapping_sub_level_compact_level_count:
64 compaction_config::level0_overlapping_sub_level_compact_level_count(),
65 tombstone_reclaim_ratio: compaction_config::tombstone_reclaim_ratio(),
66 enable_emergency_picker: compaction_config::enable_emergency_picker(),
67 max_l0_compact_level_count: Some(compaction_config::max_l0_compact_level_count()),
68 sst_allowed_trivial_move_min_size: Some(
69 compaction_config::sst_allowed_trivial_move_min_size(),
70 ),
71 disable_auto_group_scheduling: Some(
72 compaction_config::disable_auto_group_scheduling(),
73 ),
74 max_overlapping_level_size: Some(compaction_config::max_overlapping_level_size()),
75 sst_allowed_trivial_move_max_count: Some(
76 compaction_config::sst_allowed_trivial_move_max_count(),
77 ),
78 emergency_level0_sst_file_count: Some(
79 compaction_config::emergency_level0_sst_file_count(),
80 ),
81 emergency_level0_sub_level_partition: Some(
82 compaction_config::emergency_level0_sub_level_partition(),
83 ),
84 level0_stop_write_threshold_max_sst_count: Some(
85 compaction_config::level0_stop_write_threshold_max_sst_count(),
86 ),
87 level0_stop_write_threshold_max_size: Some(
88 compaction_config::level0_stop_write_threshold_max_size(),
89 ),
90 },
91 }
92 }
93
94 pub fn with_config(config: CompactionConfig) -> Self {
95 Self { config }
96 }
97
98 pub fn with_opt(opt: &CompactionConfigOpt) -> Self {
99 Self::new()
100 .max_bytes_for_level_base(opt.max_bytes_for_level_base)
101 .max_bytes_for_level_multiplier(opt.max_bytes_for_level_multiplier)
102 .max_compaction_bytes(opt.max_compaction_bytes)
103 .sub_level_max_compaction_bytes(opt.sub_level_max_compaction_bytes)
104 .level0_tier_compact_file_number(opt.level0_tier_compact_file_number)
105 .target_file_size_base(opt.target_file_size_base)
106 .compaction_filter_mask(opt.compaction_filter_mask)
107 .max_sub_compaction(opt.max_sub_compaction)
108 .level0_stop_write_threshold_sub_level_number(
109 opt.level0_stop_write_threshold_sub_level_number,
110 )
111 .level0_sub_level_compact_level_count(opt.level0_sub_level_compact_level_count)
112 .level0_overlapping_sub_level_compact_level_count(
113 opt.level0_overlapping_sub_level_compact_level_count,
114 )
115 .max_space_reclaim_bytes(opt.max_space_reclaim_bytes)
116 .level0_max_compact_file_number(opt.level0_max_compact_file_number)
117 .tombstone_reclaim_ratio(opt.tombstone_reclaim_ratio)
118 .max_level(opt.max_level as u64)
119 .sst_allowed_trivial_move_min_size(Some(opt.sst_allowed_trivial_move_min_size))
120 .sst_allowed_trivial_move_max_count(Some(opt.sst_allowed_trivial_move_max_count))
121 .max_overlapping_level_size(Some(opt.max_overlapping_level_size))
122 .emergency_level0_sst_file_count(Some(opt.emergency_level0_sst_file_count))
123 .emergency_level0_sub_level_partition(Some(opt.emergency_level0_sub_level_partition))
124 .level0_stop_write_threshold_max_sst_count(Some(
125 opt.level0_stop_write_threshold_max_sst_count,
126 ))
127 .level0_stop_write_threshold_max_size(Some(opt.level0_stop_write_threshold_max_size))
128 }
129
130 pub fn build(self) -> CompactionConfig {
131 if let Err(reason) = validate_compaction_config(&self.config) {
132 tracing::warn!("Bad compaction config: {}", reason);
133 }
134 self.config
135 }
136}
137
138pub fn validate_compaction_config(config: &CompactionConfig) -> Result<(), String> {
141 let sub_level_number_threshold_min = 1;
142 if config.level0_stop_write_threshold_sub_level_number < sub_level_number_threshold_min {
143 return Err(format!(
144 "{} is too small for level0_stop_write_threshold_sub_level_number, expect >= {}",
145 config.level0_stop_write_threshold_sub_level_number, sub_level_number_threshold_min
146 ));
147 }
148 Ok(())
149}
150
151impl Default for CompactionConfigBuilder {
152 fn default() -> Self {
153 Self::new()
154 }
155}
156
157macro_rules! builder_field {
158 ($( $name:ident: $type:ty ),* ,) => {
159 impl CompactionConfigBuilder {
160 $(
161 pub fn $name(mut self, v:$type) -> Self {
162 self.config.$name = v;
163 self
164 }
165 )*
166 }
167 }
168}
169
170builder_field! {
171 max_bytes_for_level_base: u64,
172 max_bytes_for_level_multiplier: u64,
173 max_level: u64,
174 max_compaction_bytes: u64,
175 sub_level_max_compaction_bytes: u64,
176 level0_tier_compact_file_number: u64,
177 compaction_mode: i32,
178 compression_algorithm: Vec<String>,
179 compaction_filter_mask: u32,
180 target_file_size_base: u64,
181 max_sub_compaction: u32,
182 max_space_reclaim_bytes: u64,
183 level0_stop_write_threshold_sub_level_number: u64,
184 level0_max_compact_file_number: u64,
185 level0_sub_level_compact_level_count: u32,
186 level0_overlapping_sub_level_compact_level_count: u32,
187 tombstone_reclaim_ratio: u32,
188 sst_allowed_trivial_move_min_size: Option<u64>,
189 sst_allowed_trivial_move_max_count: Option<u32>,
190 disable_auto_group_scheduling: Option<bool>,
191 max_overlapping_level_size: Option<u64>,
192 emergency_level0_sst_file_count: Option<u32>,
193 emergency_level0_sub_level_partition: Option<u32>,
194 level0_stop_write_threshold_max_sst_count: Option<u32>,
195 level0_stop_write_threshold_max_size: Option<u64>,
196}