1use std::num::NonZero;
16use std::sync::Arc;
17use std::time::Duration;
18
19use api::v1::region::compact_request;
20use common_meta::key::SchemaMetadataManagerRef;
21use common_telemetry::{info, warn};
22use common_time::TimeToLive;
23use either::Either;
24use itertools::Itertools;
25use object_store::manager::ObjectStoreManagerRef;
26use partition::expr::PartitionExpr;
27use serde::{Deserialize, Serialize};
28use snafu::{OptionExt, ResultExt};
29use store_api::metadata::RegionMetadataRef;
30use store_api::region_request::PathType;
31use store_api::storage::RegionId;
32
33use crate::access_layer::{
34 AccessLayer, AccessLayerRef, Metrics, OperationType, SstWriteRequest, WriteType,
35};
36use crate::cache::{CacheManager, CacheManagerRef};
37use crate::compaction::picker::{PickerOutput, new_picker};
38use crate::compaction::{CompactionSstReaderBuilder, find_ttl};
39use crate::config::MitoConfig;
40use crate::error::{
41 EmptyRegionDirSnafu, InvalidPartitionExprSnafu, JoinSnafu, ObjectStoreNotFoundSnafu, Result,
42};
43use crate::manifest::action::{RegionEdit, RegionMetaAction, RegionMetaActionList};
44use crate::manifest::manager::{RegionManifestManager, RegionManifestOptions, RemoveFileOptions};
45use crate::manifest::storage::manifest_compress_type;
46use crate::metrics;
47use crate::read::{FlatSource, Source};
48use crate::region::opener::new_manifest_dir;
49use crate::region::options::RegionOptions;
50use crate::region::version::VersionRef;
51use crate::region::{ManifestContext, RegionLeaderState, RegionRoleState};
52use crate::schedule::scheduler::LocalScheduler;
53use crate::sst::FormatType;
54use crate::sst::file::FileMeta;
55use crate::sst::file_purger::LocalFilePurger;
56use crate::sst::index::intermediate::IntermediateManager;
57use crate::sst::index::puffin_manager::PuffinManagerFactory;
58use crate::sst::location::region_dir_from_table_dir;
59use crate::sst::parquet::WriteOptions;
60use crate::sst::version::{SstVersion, SstVersionRef};
61
62#[derive(Clone)]
64pub struct CompactionVersion {
65 pub(crate) metadata: RegionMetadataRef,
70 pub(crate) options: RegionOptions,
72 pub(crate) ssts: SstVersionRef,
74 pub(crate) compaction_time_window: Option<Duration>,
76}
77
78impl From<VersionRef> for CompactionVersion {
79 fn from(value: VersionRef) -> Self {
80 Self {
81 metadata: value.metadata.clone(),
82 options: value.options.clone(),
83 ssts: value.ssts.clone(),
84 compaction_time_window: value.compaction_time_window,
85 }
86 }
87}
88
89#[derive(Clone)]
92pub struct CompactionRegion {
93 pub region_id: RegionId,
94 pub region_options: RegionOptions,
95
96 pub(crate) engine_config: Arc<MitoConfig>,
97 pub(crate) region_metadata: RegionMetadataRef,
98 pub(crate) cache_manager: CacheManagerRef,
99 pub access_layer: AccessLayerRef,
101 pub(crate) manifest_ctx: Arc<ManifestContext>,
102 pub(crate) current_version: CompactionVersion,
103 pub(crate) file_purger: Option<Arc<LocalFilePurger>>,
104 pub(crate) ttl: Option<TimeToLive>,
105
106 pub max_parallelism: usize,
111}
112
113#[derive(Debug, Clone)]
115pub struct OpenCompactionRegionRequest {
116 pub region_id: RegionId,
117 pub table_dir: String,
118 pub path_type: PathType,
119 pub region_options: RegionOptions,
120 pub max_parallelism: usize,
121}
122
123pub async fn open_compaction_region(
126 req: &OpenCompactionRegionRequest,
127 mito_config: &MitoConfig,
128 object_store_manager: ObjectStoreManagerRef,
129 ttl_provider: Either<TimeToLive, SchemaMetadataManagerRef>,
130) -> Result<CompactionRegion> {
131 let object_store = {
132 let name = &req.region_options.storage;
133 if let Some(name) = name {
134 object_store_manager
135 .find(name)
136 .with_context(|| ObjectStoreNotFoundSnafu {
137 object_store: name.clone(),
138 })?
139 } else {
140 object_store_manager.default_object_store()
141 }
142 };
143
144 let access_layer = {
145 let puffin_manager_factory = PuffinManagerFactory::new(
146 &mito_config.index.aux_path,
147 mito_config.index.staging_size.as_bytes(),
148 Some(mito_config.index.write_buffer_size.as_bytes() as _),
149 mito_config.index.staging_ttl,
150 )
151 .await?;
152 let intermediate_manager =
153 IntermediateManager::init_fs(mito_config.index.aux_path.clone()).await?;
154
155 Arc::new(AccessLayer::new(
156 &req.table_dir,
157 req.path_type,
158 object_store.clone(),
159 puffin_manager_factory,
160 intermediate_manager,
161 ))
162 };
163
164 let manifest_manager = {
165 let region_manifest_options = RegionManifestOptions {
166 manifest_dir: new_manifest_dir(®ion_dir_from_table_dir(
167 &req.table_dir,
168 req.region_id,
169 req.path_type,
170 )),
171 object_store: object_store.clone(),
172 compress_type: manifest_compress_type(mito_config.compress_manifest),
173 checkpoint_distance: mito_config.manifest_checkpoint_distance,
174 remove_file_options: RemoveFileOptions {
175 keep_count: mito_config.experimental_manifest_keep_removed_file_count,
176 keep_ttl: mito_config.experimental_manifest_keep_removed_file_ttl,
177 },
178 };
179
180 RegionManifestManager::open(
181 region_manifest_options,
182 Default::default(),
183 Default::default(),
184 )
185 .await?
186 .context(EmptyRegionDirSnafu {
187 region_id: req.region_id,
188 region_dir: ®ion_dir_from_table_dir(&req.table_dir, req.region_id, req.path_type),
189 })?
190 };
191
192 let manifest = manifest_manager.manifest();
193 let region_metadata = manifest.metadata.clone();
194 let manifest_ctx = Arc::new(ManifestContext::new(
195 manifest_manager,
196 RegionRoleState::Leader(RegionLeaderState::Writable),
197 ));
198
199 let file_purger = {
200 let purge_scheduler = Arc::new(LocalScheduler::new(mito_config.max_background_purges));
201 Arc::new(LocalFilePurger::new(
202 purge_scheduler.clone(),
203 access_layer.clone(),
204 None,
205 ))
206 };
207
208 let current_version = {
209 let mut ssts = SstVersion::new();
210 ssts.add_files(file_purger.clone(), manifest.files.values().cloned());
211 CompactionVersion {
212 metadata: region_metadata.clone(),
213 options: req.region_options.clone(),
214 ssts: Arc::new(ssts),
215 compaction_time_window: manifest.compaction_time_window,
216 }
217 };
218
219 let ttl = match ttl_provider {
220 Either::Left(ttl) => ttl,
222 Either::Right(schema_metadata_manager) => find_ttl(
224 req.region_id.table_id(),
225 current_version.options.ttl,
226 &schema_metadata_manager,
227 )
228 .await
229 .unwrap_or_else(|e| {
230 warn!(e; "Failed to get ttl for region: {}", region_metadata.region_id);
231 TimeToLive::default()
232 }),
233 };
234
235 Ok(CompactionRegion {
236 region_id: req.region_id,
237 region_options: req.region_options.clone(),
238 engine_config: Arc::new(mito_config.clone()),
239 region_metadata: region_metadata.clone(),
240 cache_manager: Arc::new(CacheManager::default()),
241 access_layer,
242 manifest_ctx,
243 current_version,
244 file_purger: Some(file_purger),
245 ttl: Some(ttl),
246 max_parallelism: req.max_parallelism,
247 })
248}
249
250impl CompactionRegion {
251 pub fn file_purger(&self) -> Option<Arc<LocalFilePurger>> {
253 self.file_purger.clone()
254 }
255
256 pub async fn stop_purger_scheduler(&self) -> Result<()> {
258 if let Some(file_purger) = &self.file_purger {
259 file_purger.stop_scheduler().await
260 } else {
261 Ok(())
262 }
263 }
264}
265
266#[derive(Default, Clone, Debug, Serialize, Deserialize)]
268pub struct MergeOutput {
269 pub files_to_add: Vec<FileMeta>,
270 pub files_to_remove: Vec<FileMeta>,
271 pub compaction_time_window: Option<i64>,
272}
273
274impl MergeOutput {
275 pub fn is_empty(&self) -> bool {
276 self.files_to_add.is_empty() && self.files_to_remove.is_empty()
277 }
278
279 pub fn input_file_size(&self) -> u64 {
280 self.files_to_remove.iter().map(|f| f.file_size).sum()
281 }
282
283 pub fn output_file_size(&self) -> u64 {
284 self.files_to_add.iter().map(|f| f.file_size).sum()
285 }
286}
287
288#[async_trait::async_trait]
290pub trait Compactor: Send + Sync + 'static {
291 async fn merge_ssts(
293 &self,
294 compaction_region: &CompactionRegion,
295 picker_output: PickerOutput,
296 ) -> Result<MergeOutput>;
297
298 async fn update_manifest(
300 &self,
301 compaction_region: &CompactionRegion,
302 merge_output: MergeOutput,
303 ) -> Result<RegionEdit>;
304
305 async fn compact(
307 &self,
308 compaction_region: &CompactionRegion,
309 compact_request_options: compact_request::Options,
310 ) -> Result<()>;
311}
312
313pub struct DefaultCompactor;
315
316#[async_trait::async_trait]
317impl Compactor for DefaultCompactor {
318 async fn merge_ssts(
319 &self,
320 compaction_region: &CompactionRegion,
321 mut picker_output: PickerOutput,
322 ) -> Result<MergeOutput> {
323 let mut futs = Vec::with_capacity(picker_output.outputs.len());
324 let mut compacted_inputs =
325 Vec::with_capacity(picker_output.outputs.iter().map(|o| o.inputs.len()).sum());
326 let internal_parallelism = compaction_region.max_parallelism.max(1);
327
328 for output in picker_output.outputs.drain(..) {
329 compacted_inputs.extend(output.inputs.iter().map(|f| f.meta_ref().clone()));
330 let write_opts = WriteOptions {
331 write_buffer_size: compaction_region.engine_config.sst_write_buffer_size,
332 max_file_size: picker_output.max_file_size,
333 ..Default::default()
334 };
335
336 let region_metadata = compaction_region.region_metadata.clone();
337 let sst_layer = compaction_region.access_layer.clone();
338 let region_id = compaction_region.region_id;
339 let cache_manager = compaction_region.cache_manager.clone();
340 let storage = compaction_region.region_options.storage.clone();
341 let index_options = compaction_region
342 .current_version
343 .options
344 .index_options
345 .clone();
346 let append_mode = compaction_region.current_version.options.append_mode;
347 let merge_mode = compaction_region.current_version.options.merge_mode();
348 let flat_format = compaction_region
349 .region_options
350 .sst_format
351 .map(|format| format == FormatType::Flat)
352 .unwrap_or(
353 compaction_region
354 .engine_config
355 .default_experimental_flat_format,
356 );
357 let index_config = compaction_region.engine_config.index.clone();
358 let inverted_index_config = compaction_region.engine_config.inverted_index.clone();
359 let fulltext_index_config = compaction_region.engine_config.fulltext_index.clone();
360 let bloom_filter_index_config =
361 compaction_region.engine_config.bloom_filter_index.clone();
362 let max_sequence = output
363 .inputs
364 .iter()
365 .map(|f| f.meta_ref().sequence)
366 .max()
367 .flatten();
368 let region_metadata_for_filemeta = region_metadata.clone();
369 futs.push(async move {
370 let input_file_names = output
371 .inputs
372 .iter()
373 .map(|f| f.file_id().to_string())
374 .join(",");
375 let builder = CompactionSstReaderBuilder {
376 metadata: region_metadata.clone(),
377 sst_layer: sst_layer.clone(),
378 cache: cache_manager.clone(),
379 inputs: &output.inputs,
380 append_mode,
381 filter_deleted: output.filter_deleted,
382 time_range: output.output_time_range,
383 merge_mode,
384 };
385 let source = if flat_format {
386 let reader = builder.build_flat_sst_reader().await?;
387 either::Right(FlatSource::Stream(reader))
388 } else {
389 let reader = builder.build_sst_reader().await?;
390 either::Left(Source::Reader(reader))
391 };
392 let mut metrics = Metrics::new(WriteType::Compaction);
393 let sst_infos = sst_layer
394 .write_sst(
395 SstWriteRequest {
396 op_type: OperationType::Compact,
397 metadata: region_metadata,
398 source,
399 cache_manager,
400 storage,
401 max_sequence: max_sequence.map(NonZero::get),
402 index_options,
403 index_config,
404 inverted_index_config,
405 fulltext_index_config,
406 bloom_filter_index_config,
407 },
408 &write_opts,
409 &mut metrics,
410 )
411 .await?;
412 let partition_expr = match ®ion_metadata_for_filemeta.partition_expr {
414 None => None,
415 Some(json_str) if json_str.is_empty() => None,
416 Some(json_str) => {
417 PartitionExpr::from_json_str(json_str).with_context(|_| {
418 InvalidPartitionExprSnafu {
419 expr: json_str.clone(),
420 }
421 })?
422 }
423 };
424
425 let output_files = sst_infos
426 .into_iter()
427 .map(|sst_info| FileMeta {
428 region_id,
429 file_id: sst_info.file_id,
430 time_range: sst_info.time_range,
431 level: output.output_level,
432 file_size: sst_info.file_size,
433 available_indexes: sst_info.index_metadata.build_available_indexes(),
434 index_file_size: sst_info.index_metadata.file_size,
435 num_rows: sst_info.num_rows as u64,
436 num_row_groups: sst_info.num_row_groups,
437 sequence: max_sequence,
438 partition_expr: partition_expr.clone(),
439 num_series: sst_info.num_series,
440 })
441 .collect::<Vec<_>>();
442 let output_file_names =
443 output_files.iter().map(|f| f.file_id.to_string()).join(",");
444 info!(
445 "Region {} compaction inputs: [{}], outputs: [{}], flat_format: {}, metrics: {:?}",
446 region_id, input_file_names, output_file_names, flat_format, metrics
447 );
448 metrics.observe();
449 Ok(output_files)
450 });
451 }
452 let mut output_files = Vec::with_capacity(futs.len());
453 while !futs.is_empty() {
454 let mut task_chunk = Vec::with_capacity(internal_parallelism);
455 for _ in 0..internal_parallelism {
456 if let Some(task) = futs.pop() {
457 task_chunk.push(common_runtime::spawn_compact(task));
458 }
459 }
460 let metas = futures::future::try_join_all(task_chunk)
461 .await
462 .context(JoinSnafu)?
463 .into_iter()
464 .collect::<Result<Vec<Vec<_>>>>()?;
465 output_files.extend(metas.into_iter().flatten());
466 }
467
468 let mut inputs: Vec<_> = compacted_inputs.into_iter().collect();
469 inputs.extend(
470 picker_output
471 .expired_ssts
472 .iter()
473 .map(|f| f.meta_ref().clone()),
474 );
475
476 Ok(MergeOutput {
477 files_to_add: output_files,
478 files_to_remove: inputs,
479 compaction_time_window: Some(picker_output.time_window_size),
480 })
481 }
482
483 async fn update_manifest(
484 &self,
485 compaction_region: &CompactionRegion,
486 merge_output: MergeOutput,
487 ) -> Result<RegionEdit> {
488 let edit = RegionEdit {
490 files_to_add: merge_output.files_to_add,
491 files_to_remove: merge_output.files_to_remove,
492 timestamp_ms: Some(chrono::Utc::now().timestamp_millis()),
494 compaction_time_window: merge_output
495 .compaction_time_window
496 .map(|seconds| Duration::from_secs(seconds as u64)),
497 flushed_entry_id: None,
498 flushed_sequence: None,
499 committed_sequence: None,
500 };
501
502 let action_list = RegionMetaActionList::with_action(RegionMetaAction::Edit(edit.clone()));
503 compaction_region
505 .manifest_ctx
506 .update_manifest(RegionLeaderState::Writable, action_list)
507 .await?;
508
509 Ok(edit)
510 }
511
512 async fn compact(
515 &self,
516 compaction_region: &CompactionRegion,
517 compact_request_options: compact_request::Options,
518 ) -> Result<()> {
519 let picker_output = {
520 let picker_output = new_picker(
521 &compact_request_options,
522 &compaction_region.region_options.compaction,
523 compaction_region.region_options.append_mode,
524 )
525 .pick(compaction_region);
526
527 if let Some(picker_output) = picker_output {
528 picker_output
529 } else {
530 info!(
531 "No files to compact for region_id: {}",
532 compaction_region.region_id
533 );
534 return Ok(());
535 }
536 };
537
538 let merge_output = self.merge_ssts(compaction_region, picker_output).await?;
539 if merge_output.is_empty() {
540 info!(
541 "No files to compact for region_id: {}",
542 compaction_region.region_id
543 );
544 return Ok(());
545 }
546
547 metrics::COMPACTION_INPUT_BYTES.inc_by(merge_output.input_file_size() as f64);
548 metrics::COMPACTION_OUTPUT_BYTES.inc_by(merge_output.output_file_size() as f64);
549 self.update_manifest(compaction_region, merge_output)
550 .await?;
551
552 Ok(())
553 }
554}