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

mito2/
compaction.rs
