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

mito2/
compaction.rs
