mito2/
compaction.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
15mod buckets;
16pub mod compactor;
17mod json2;
18mod last_non_null;
19pub mod memory_manager;
20#[cfg(feature = "enterprise")]
21pub mod multi_region;
22mod overlap;
23pub mod picker;
24mod reader;
25pub mod run;
26mod scheduler;
27mod task;
28#[cfg(test)]
29mod test_util;
30mod twcs;
31mod unit;
32mod window;
33
34use std::any::Any;
35use std::collections::HashMap;
36
37use common_meta::key::SchemaMetadataManagerRef;
38use common_telemetry::{debug, error};
39use common_time::TimeToLive;
40use common_time::range::TimestampRange;
41pub(crate) use json2::{
42    Json2RewritePlans, collect_json2_rewrite_plans, rewrite_json2_batch, rewrite_json2_schema,
43};
44pub use scheduler::CompactionRequest;
45pub(crate) use scheduler::{
46    CompactionCoordinator, CompactionExecution, CompactionPickFinished, CompactionScheduler,
47    CompactionState, CompactionStateRef, CompactionTransition,
48};
49use serde::{Deserialize, Serialize};
50use snafu::ResultExt;
51use store_api::mito_engine_options::{TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_TRIGGER_FILE_NUM};
52use store_api::storage::RegionId;
53
54use crate::error::{GetSchemaMetadataSnafu, Result, TimeoutSnafu};
55use crate::sst::file::{FileHandle, FileMeta, Level};
56
57/// Finds compaction options and TTL together with a single metadata fetch to reduce RTT.
58async fn find_dynamic_options(
59    region_id: RegionId,
60    region_options: &crate::region::options::RegionOptions,
61    schema_metadata_manager: &SchemaMetadataManagerRef,
62) -> Result<(crate::region::options::CompactionOptions, TimeToLive)> {
63    let table_id = region_id.table_id();
64    if let (true, Some(ttl)) = (region_options.compaction_override, region_options.ttl) {
65        debug!(
66            "Use region options directly for table {}: compaction={:?}, ttl={:?}",
67            table_id, region_options.compaction, region_options.ttl
68        );
69        return Ok((region_options.compaction.clone(), ttl));
70    }
71
72    let db_options = tokio::time::timeout(
73        crate::config::FETCH_OPTION_TIMEOUT,
74        schema_metadata_manager.get_schema_options_by_table_id(table_id),
75    )
76    .await
77    .context(TimeoutSnafu)?
78    .context(GetSchemaMetadataSnafu)?;
79
80    let ttl = if let Some(ttl) = region_options.ttl {
81        debug!(
82            "Use region TTL directly for table {}: ttl={:?}",
83            table_id, region_options.ttl
84        );
85        ttl
86    } else {
87        db_options
88            .as_ref()
89            .and_then(|options| options.ttl)
90            .unwrap_or_default()
91            .into()
92    };
93
94    let compaction = if !region_options.compaction_override {
95        if let Some(schema_opts) = db_options {
96            let mut map: HashMap<String, String> = schema_opts
97                .extra_options
98                .iter()
99                .filter_map(|(k, v)| {
100                    if k.starts_with("compaction.") {
101                        Some((k.clone(), v.clone()))
102                    } else {
103                        None
104                    }
105                })
106                .collect();
107            // Historical metadata may contain both aliases; prefer the canonical key.
108            if map.contains_key(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM) {
109                map.remove(TWCS_TRIGGER_FILE_NUM);
110            }
111            if map.is_empty() {
112                region_options.compaction.clone()
113            } else {
114                crate::region::options::RegionOptions::try_from_options(region_id, &map)
115                    .map(|o| o.compaction)
116                    .unwrap_or_else(|e| {
117                        error!(e; "Failed to create RegionOptions from map");
118                        region_options.compaction.clone()
119                    })
120            }
121        } else {
122            debug!(
123                "DB options is None for table {}, use region compaction: compaction={:?}",
124                table_id, region_options.compaction
125            );
126            region_options.compaction.clone()
127        }
128    } else {
129        debug!(
130            "No schema options for table {}, use region compaction: compaction={:?}",
131            table_id, region_options.compaction
132        );
133        region_options.compaction.clone()
134    };
135
136    debug!(
137        "Resolved dynamic options for table {}: compaction={:?}, ttl={:?}",
138        table_id, compaction, ttl
139    );
140    Ok((compaction, ttl))
141}
142
143#[derive(Debug, Clone)]
144pub struct CompactionOutput {
145    /// Compaction output file level.
146    pub output_level: Level,
147    /// Compaction input files.
148    pub inputs: Vec<FileHandle>,
149    /// Whether to remove deletion markers.
150    pub filter_deleted: bool,
151    /// Compaction output time range. Only windowed compaction specifies output time range.
152    pub output_time_range: Option<TimestampRange>,
153}
154
155/// SerializedCompactionOutput is a serialized version of [CompactionOutput] by replacing [FileHandle] with [FileMeta].
156#[derive(Debug, Clone, Serialize, Deserialize)]
157pub struct SerializedCompactionOutput {
158    output_level: Level,
159    inputs: Vec<FileMeta>,
160    filter_deleted: bool,
161    output_time_range: Option<TimestampRange>,
162}
163
164/// Returns the printable reason carried by a caught panic payload.
165pub(crate) fn format_panic_reason(payload: &(dyn Any + Send + 'static)) -> String {
166    if let Some(message) = payload.downcast_ref::<&str>() {
167        message.to_string()
168    } else if let Some(message) = payload.downcast_ref::<String>() {
169        message.clone()
170    } else {
171        "unknown panic".to_string()
172    }
173}
mito2/compaction.rs

mito2/
compaction.rs
