rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-08-18 12:08:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mito2): limit compaction picker threads (#8704)

Browse Source 
* fix(mito2): limit compaction picker threads

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* refactor(mito2): extract TWCS input picking

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* refactor(mito2): make compaction picker async

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* refactor(mito2): remove redundant build_output test helper

After making the compaction picker async and extracting TWCS input
picking, the test-only build_output helper is just a thin wrapper around
build_output_with_time_range. Drop it and call the full method directly
from tests.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* docs: correct compact runtime config wording

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* test(runtime): harden compact blocking limit check

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

---------

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
This commit is contained in:
Lei, HUANG
2026-07-31 23:15:55 +08:00
committed by Lei, HUANG
parent 39aa8ab974
commit 8eeff8b441
11 changed files with 403 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/config.md
+8 -4
View File
@@ -24,7 +24,8 @@
| `enable_telemetry` | Bool | `true` | Enable telemetry to collect anonymous usage data. Enabled by default. |
| `runtime` | -- | -- | The runtime options. |
| `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
| `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute the runtime for global write operations. |
| `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute compact operations. |
| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 1). |
| `http` | -- | -- | The HTTP server options. |
| `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.<br/>When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the<br/>`prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value. |
@@ -251,7 +252,8 @@
| `write_bytes_exhausted_policy` | String | Unset | Policy when write bytes quota is exhausted.<br/>Options: "wait" (default, 10s timeout), "wait(<duration>)" (e.g., "wait(30s)"), "fail" |
| `runtime` | -- | -- | The runtime options. |
| `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
| `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute the runtime for global write operations. |
| `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute compact operations. |
| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 1). |
| `http` | -- | -- | The HTTP server options. |
| `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.<br/>When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the<br/>`prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value. |
@@ -391,7 +393,8 @@
| `enable_telemetry` | Bool | `true` | Whether to enable greptimedb telemetry. Enabled by default. |
| `runtime` | -- | -- | The runtime options. |
| `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
| `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute the runtime for global write operations. |
| `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute compact operations. |
| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 1). |
| `backend_tls` | -- | -- | TLS configuration for kv store backend (applicable for etcd, PostgreSQL, and MySQL backends)<br/>When using etcd, PostgreSQL, or MySQL as metadata store, you can configure TLS here<br/><br/>Note: if TLS is configured in both this section and the `store_addrs` connection string, the<br/>settings here will override the TLS settings in `store_addrs`. |
| `backend_tls.mode` | String | `prefer` | TLS mode, refer to https://www.postgresql.org/docs/current/libpq-ssl.html<br/>- "disable" - No TLS<br/>- "prefer" (default) - Try TLS, fallback to plain<br/>- "require" - Require TLS<br/>- "verify_ca" - Require TLS and verify CA<br/>- "verify_full" - Require TLS and verify hostname |
| `backend_tls.cert_path` | String | `""` | Path to client certificate file (for client authentication)<br/>Like "/path/to/client.crt" |
@@ -500,7 +503,8 @@
| `grpc.tls.watch` | Bool | `false` | Watch for Certificate and key file change and auto reload.<br/>For now, gRPC tls config does not support auto reload. |
| `runtime` | -- | -- | The runtime options. |
| `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
| `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute the runtime for global write operations. |
| `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute compact operations. |
| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 1). |
| `runtime.query_rt_size` | Integer | `7` | The number of threads to execute datanode query operations.<br/>Defaults to max(num_cpus - 1, 1). |
| `runtime.ingest_rt_size` | Integer | `8` | The number of threads to execute datanode ingestion operations. |
| `meta_client` | -- | -- | The metasrv client options. |
config/datanode.example.toml
+4 -1
View File
@@ -81,8 +81,11 @@ watch = false
#+ [runtime]
## The number of threads to execute the runtime for global read operations.
#+ global_rt_size = 8
## The number of threads to execute the runtime for global write operations.
## The number of threads to execute compact operations.
#+ compact_rt_size = 4
## The maximum number of blocking threads for compact operations.
## Defaults to max(num_cpus / 2, 1).
#+ compact_rt_max_blocking_threads = 4
## The number of threads to execute datanode query operations.
## Defaults to max(num_cpus - 1, 1).
#+ query_rt_size = 7
config/frontend.example.toml
+4 -1
View File
@@ -41,8 +41,11 @@ default_column_prefix = "greptime"
#+ [runtime]
## The number of threads to execute the runtime for global read operations.
#+ global_rt_size = 8
## The number of threads to execute the runtime for global write operations.
## The number of threads to execute compact operations.
#+ compact_rt_size = 4
## The maximum number of blocking threads for compact operations.
## Defaults to max(num_cpus / 2, 1).
#+ compact_rt_max_blocking_threads = 4
## The HTTP server options.
[http]
config/metasrv.example.toml
+4 -1
View File
@@ -89,8 +89,11 @@ node_max_idle_time = "24hours"
#+ [runtime]
## The number of threads to execute the runtime for global read operations.
#+ global_rt_size = 8
## The number of threads to execute the runtime for global write operations.
## The number of threads to execute compact operations.
#+ compact_rt_size = 4
## The maximum number of blocking threads for compact operations.
## Defaults to max(num_cpus / 2, 1).
#+ compact_rt_max_blocking_threads = 4
## TLS configuration for kv store backend (applicable for etcd, PostgreSQL, and MySQL backends)
## When using etcd, PostgreSQL, or MySQL as metadata store, you can configure TLS here
config/standalone.example.toml
+4 -1
View File
@@ -54,8 +54,11 @@ max_concurrent_queries = 0
#+ [runtime]
## The number of threads to execute the runtime for global read operations.
#+ global_rt_size = 8
## The number of threads to execute the runtime for global write operations.
## The number of threads to execute compact operations.
#+ compact_rt_size = 4
## The maximum number of blocking threads for compact operations.
## Defaults to max(num_cpus / 2, 1).
#+ compact_rt_max_blocking_threads = 4
## The HTTP server options.
[http]
src/cmd/tests/load_config_test.rs
+21
View File
@@ -20,6 +20,7 @@ use common_base::memory_limit::MemoryLimit;
use common_base::readable_size::ReadableSize;
use common_config::{Configurable, DEFAULT_DATA_HOME, ENV_VAR_SEP};
use common_options::datanode::{ClientOptions, DatanodeClientOptions};
use common_runtime::global::RuntimeOptions;
use common_telemetry::logging::{DEFAULT_LOGGING_DIR, DEFAULT_OTLP_HTTP_ENDPOINT, LoggingOptions};
use common_test_util::temp_dir::create_named_temp_file;
use common_wal::config::DatanodeWalConfig;
@@ -46,6 +47,7 @@ fn test_load_datanode_runtime_options_from_runtime_section() {
[runtime]
global_rt_size = 8
compact_rt_size = 4
compact_rt_max_blocking_threads = 6
ingest_rt_size = 8
query_rt_size = 7
"#;
@@ -54,10 +56,29 @@ fn test_load_datanode_runtime_options_from_runtime_section() {
assert_eq!(8, options.runtime.global_rt_size);
assert_eq!(4, options.runtime.compact_rt_size);
assert_eq!(6, options.runtime.compact_rt_max_blocking_threads);
assert_eq!(8, options.runtime.ingest_rt_size);
assert_eq!(7, options.runtime.query_rt_size);
}
#[test]
fn test_load_runtime_options_without_max_blocking_threads() {
let toml = r#"
[runtime]
global_rt_size = 8
compact_rt_size = 4
ingest_rt_size = 8
query_rt_size = 7
"#;
let options: GreptimeOptions<DatanodeOptions> = toml::from_str(toml).unwrap();
assert_eq!(
RuntimeOptions::default().compact_rt_max_blocking_threads,
options.runtime.compact_rt_max_blocking_threads
);
}
#[allow(deprecated)]
#[test]
fn test_load_datanode_example_config() {
src/common/runtime/src/global.rs
+80 -3
View File
@@ -36,6 +36,8 @@ pub struct RuntimeOptions {
pub global_rt_size: usize,
/// The number of threads to execute the runtime for compact operations.
pub compact_rt_size: usize,
/// The maximum number of blocking threads for compact operations.
pub compact_rt_max_blocking_threads: usize,
/// The number of threads to execute datanode query operations.
pub query_rt_size: usize,
/// The number of threads to execute datanode ingestion operations.
@@ -48,6 +50,7 @@ impl Default for RuntimeOptions {
Self {
global_rt_size: cpus,
compact_rt_size: usize::max(cpus / 2, 1),
compact_rt_max_blocking_threads: usize::max(cpus / 2, 1),
query_rt_size: usize::max(cpus.saturating_sub(1), 1),
ingest_rt_size: cpus,
}
@@ -66,6 +69,25 @@ pub fn create_runtime(runtime_name: &str, thread_name: &str, worker_threads: usi
.expect("Fail to create runtime")
}
fn create_compact_runtime(
runtime_name: &str,
thread_name: &str,
worker_threads: usize,
max_blocking_threads: usize,
) -> Runtime {
let max_blocking_threads = max_blocking_threads.max(1);
info!(
"Creating compact runtime with runtime_name: {runtime_name}, thread_name: {thread_name}, work_threads: {worker_threads}, max_blocking_threads: {max_blocking_threads}."
);
Builder::default()
.runtime_name(runtime_name)
.thread_name(thread_name)
.worker_threads(worker_threads)
.max_blocking_threads(max_blocking_threads)
.build()
.expect("Fail to create runtime")
}
struct GlobalRuntimes {
global_runtime: Runtime,
compact_runtime: Runtime,
@@ -122,8 +144,16 @@ impl GlobalRuntimes {
Self {
global_runtime,
compact_runtime: compact
.unwrap_or_else(|| create_runtime("compact", "compact-worker", COMPACT_WORKERS)),
compact_runtime: compact.unwrap_or_else(|| {
let max_blocking_threads =
RuntimeOptions::default().compact_rt_max_blocking_threads;
create_compact_runtime(
"compact",
"compact-worker",
COMPACT_WORKERS,
max_blocking_threads,
)
}),
hb_runtime: heartbeat
.unwrap_or_else(|| create_runtime("heartbeat", "hb-worker", HB_WORKERS)),
query_runtime,
@@ -172,10 +202,11 @@ pub fn init_global_runtimes(options: &RuntimeOptions) {
"global-worker",
options.global_rt_size,
));
c.compact_runtime = Some(create_runtime(
c.compact_runtime = Some(create_compact_runtime(
"compact",
"compact-worker",
options.compact_rt_size,
options.compact_rt_max_blocking_threads,
));
c.hb_runtime = Some(create_runtime("heartbeat", "hb-worker", HB_WORKERS));
});
@@ -246,6 +277,9 @@ define_global_runtime_spawn!(ingest);
#[cfg(test)]
mod tests {
use std::sync::mpsc;
use std::time::Duration;
use tokio_test::assert_ok;
use super::*;
@@ -257,6 +291,10 @@ mod tests {
assert_eq!(cpus, options.global_rt_size);
assert_eq!(usize::max(cpus / 2, 1), options.compact_rt_size);
assert_eq!(
usize::max(cpus / 2, 1),
options.compact_rt_max_blocking_threads
);
assert_eq!(usize::max(cpus.saturating_sub(1), 1), options.query_rt_size);
assert_eq!(cpus, options.ingest_rt_size);
}
@@ -276,6 +314,45 @@ mod tests {
assert_eq!("test-global", runtimes.ingest_runtime.name());
}
#[test]
fn test_create_compact_runtime_with_zero_max_blocking_threads() {
let runtime = create_compact_runtime("test-compact", "test-compact-worker", 1, 0);
let handle = runtime.spawn_blocking(|| 1 + 1);
assert_eq!(2, runtime.block_on(handle).unwrap());
}
#[test]
fn test_compact_runtime_limits_blocking_threads() {
let runtime = create_compact_runtime("test-compact", "test-compact-worker", 1, 1);
let (first_started_tx, first_started_rx) = mpsc::channel();
let (release_first_tx, release_first_rx) = mpsc::channel();
let first = runtime.spawn_blocking(move || {
first_started_tx.send(()).unwrap();
release_first_rx.recv().unwrap();
});
first_started_rx
.recv_timeout(Duration::from_secs(5))
.unwrap();
let (second_started_tx, second_started_rx) = mpsc::channel();
let second = runtime.spawn_blocking(move || second_started_tx.send(()).unwrap());
assert!(
second_started_rx
.recv_timeout(Duration::from_secs(1))
.is_err()
);
release_first_tx.send(()).unwrap();
second_started_rx
.recv_timeout(Duration::from_secs(5))
.unwrap();
runtime.block_on(async {
first.await.unwrap();
second.await.unwrap();
});
}
#[test]
fn test_datanode_runtime_spawn_block_on() {
let handle = spawn_query(async { 1 + 1 });
src/mito2/src/compaction/picker.rs
+3 -1
View File
@@ -24,6 +24,7 @@ use crate::compaction::compactor::CompactionRegion;
use crate::compaction::twcs::TwcsPicker;
use crate::compaction::window::WindowedCompactionPicker;
use crate::compaction::{CompactionOutput, SerializedCompactionOutput};
use crate::error::Result;
use crate::region::options::CompactionOptions;
use crate::sst::file::{FileHandle, FileMeta};
use crate::sst::file_purger::FilePurger;
@@ -36,9 +37,10 @@ pub(crate) trait CompactionTask: Debug + Send + Sync + 'static {
/// Picker picks input SST files for compaction.
/// Different compaction strategy may implement different pickers.
#[async_trait::async_trait]
pub trait Picker: Debug + Send + Sync + 'static {
/// Picks input SST files for compaction.
fn pick(&self, compaction_region: &CompactionRegion) -> Option<PickerOutput>;
async fn pick(&self, compaction_region: &CompactionRegion) -> Result<Option<PickerOutput>>;
}
/// PickerOutput is the output of a [`Picker`].
src/mito2/src/compaction/scheduler/planning.rs
+5 -13
View File
@@ -39,9 +39,7 @@ use crate::compaction::scheduler::state::{
use crate::compaction::task::CompactionTaskImpl;
use crate::compaction::{CompactionOutput, find_dynamic_options};
use crate::config::MitoConfig;
use crate::error::{
CompactRegionSnafu, Error, JoinSnafu, RemoteCompactionSnafu, Result, UnexpectedSnafu,
};
use crate::error::{CompactRegionSnafu, Error, RemoteCompactionSnafu, Result, UnexpectedSnafu};
use crate::metrics::{
COMPACTION_MEMORY_REJECTED, COMPACTION_STAGE_ELAPSED, INFLIGHT_COMPACTION_COUNT,
};
@@ -254,16 +252,10 @@ impl CompactionScheduler {
};
listener.on_compaction_pick_begin(region_id).await;
let picker_region = compaction_region.clone();
let picker_output = match common_runtime::spawn_blocking_compact(move || {
let _pick_timer = COMPACTION_STAGE_ELAPSED
.with_label_values(&["pick"])
.start_timer();
picker.pick(&picker_region)
})
.await
.context(JoinSnafu)
{
let _pick_timer = COMPACTION_STAGE_ELAPSED
.with_label_values(&["pick"])
.start_timer();
let picker_output = match picker.pick(&compaction_region).await {
Ok(output) => output,
Err(err) => return CompactionPlanningResult::Error(Arc::new(err)),
};
src/mito2/src/compaction/twcs.rs
+254 -209
View File
@@ -16,6 +16,7 @@ use std::collections::hash_map::Entry;
use std::collections::{BTreeMap, HashMap, HashSet};
use std::fmt::Debug;
use std::num::NonZeroU64;
use std::sync::Arc;
use common_base::readable_size::ReadableSize;
use common_telemetry::{debug, info};
@@ -23,7 +24,7 @@ use common_time::Timestamp;
use common_time::range::TimestampRange;
use common_time::timestamp::TimeUnit;
use common_time::timestamp_millis::BucketAligned;
use rayon::prelude::*;
use snafu::ResultExt;
use store_api::storage::RegionId;
use crate::compaction::CompactionOutput;
@@ -34,6 +35,7 @@ use crate::compaction::run::{
FileGroup, Item, Ranged, find_sorted_runs, find_sorted_runs_by_time_range,
merge_primary_key_ranges, merge_seq_files, primary_key_ranges_overlap, reduce_runs,
};
use crate::error::{JoinSnafu, Result};
use crate::sst::file::{FileHandle, Level, overlaps};
use crate::sst::version::LevelMeta;
@@ -44,7 +46,7 @@ const DEFAULT_MAX_INPUT_FILE_NUM: usize = 32;
/// `TwcsPicker` picks files of which the max timestamp are in the same time window as compaction
/// candidates.
#[derive(Debug)]
#[derive(Clone, Debug)]
pub struct TwcsPicker {
/// Minimum file num to trigger a compaction.
pub trigger_file_num: usize,
@@ -61,114 +63,13 @@ pub struct TwcsPicker {
}
impl TwcsPicker {
/// Builds compaction output from files.
#[cfg(test)]
fn build_output(
async fn build_output_with_time_range(
&self,
region_id: RegionId,
time_windows: &mut BTreeMap<i64, Window>,
active_window: Option<i64>,
) -> Vec<CompactionOutput> {
self.build_output_with_time_range(region_id, time_windows, active_window, None)
}
fn build_output_with_time_range(
&self,
region_id: RegionId,
time_windows: &mut BTreeMap<i64, Window>,
time_windows: BTreeMap<i64, Window>,
active_window: Option<i64>,
time_window_size: Option<i64>,
) -> Vec<CompactionOutput> {
let find_inputs = |files: &Window,
windows: &BTreeMap<i64, Window>|
-> (Vec<FileGroup>, bool) {
let window = &files.time_window;
let mut files_to_merge: Vec<_> = files.files().cloned().collect();
// Filter out large files in append mode - they won't benefit from compaction
if self.append_mode
&& let Some(max_size) = self.max_output_file_size
{
let (kept_files, ignored_files) = files_to_merge
.into_iter()
.partition(|fg| fg.size() <= max_size as usize);
files_to_merge = kept_files;
if !ignored_files.is_empty() {
info!(
"Skipped {} large files in append mode for region {}, window {}, max_size: {}",
ignored_files.len(),
region_id,
window,
max_size
);
}
}
let sorted_runs = if files_to_merge.len() < 1024 {
find_sorted_runs(&mut files_to_merge)
} else {
find_sorted_runs_by_time_range(&mut files_to_merge)
};
let found_runs = sorted_runs.len();
// We only remove deletion markers if we found less than 2 runs and not in append mode.
// because after compaction there will be no overlapping files.
let filter_deleted =
found_runs <= 2 && !self.append_mode && !window_has_overlap(files, windows);
if found_runs == 0 {
return (vec![], filter_deleted);
}
let mut inputs = if found_runs > 1 {
reduce_runs(sorted_runs)
} else {
let run = sorted_runs.last().unwrap();
if run.items().len() < self.trigger_file_num {
return (vec![], filter_deleted);
}
// no overlapping files, try merge small files
merge_seq_files(run.items(), self.max_output_file_size)
};
// Limits the number of input files.
let total_input_files: usize = inputs.iter().map(|fg| fg.num_files()).sum();
if total_input_files > DEFAULT_MAX_INPUT_FILE_NUM {
// Sorts file groups by size first.
inputs.sort_unstable_by_key(|fg| fg.size());
let mut num_picked_files = 0;
inputs = inputs
.into_iter()
.take_while(|fg| {
let current_group_file_num = fg.num_files();
if current_group_file_num + num_picked_files <= DEFAULT_MAX_INPUT_FILE_NUM {
num_picked_files += current_group_file_num;
true
} else {
false
}
})
.collect::<Vec<_>>();
info!(
"Compaction for region {} enforces max input file num limit: {}, current total: {}, input: {:?}",
region_id, DEFAULT_MAX_INPUT_FILE_NUM, total_input_files, inputs
);
}
if inputs.len() > 1 {
// If we have more than one group to compact.
log_pick_result(
region_id,
*window,
active_window,
found_runs,
files.files.len(),
self.max_output_file_size,
filter_deleted,
&inputs,
);
}
(inputs, filter_deleted)
};
) -> Result<Vec<CompactionOutput>> {
let mut output = vec![];
let windows = time_windows
.values()
@@ -184,14 +85,27 @@ impl TwcsPicker {
})
})
})
.map(|window| window.time_window)
.collect::<Vec<_>>();
let time_windows = Arc::new(time_windows);
let chunk_size = self.max_background_tasks.unwrap_or(windows.len()).max(1);
'chunks: for chunk in windows.chunks(chunk_size) {
for (inputs, filter_deleted) in chunk
.par_iter() // parallelly calculate the inputs
.map(|window| find_inputs(window, time_windows))
.collect::<Vec<_>>()
{
let mut handles = Vec::with_capacity(chunk.len());
for window in chunk {
let picker = self.clone();
let time_windows = time_windows.clone();
let window = *window;
handles.push(common_runtime::spawn_blocking_compact(move || {
time_windows.get(&window).map(|window| {
picker.find_inputs(region_id, active_window, window, &time_windows)
})
}));
tokio::task::yield_now().await;
}
for result in futures::future::join_all(handles).await {
let Some((inputs, filter_deleted)) = result.context(JoinSnafu)? else {
continue;
};
if inputs.is_empty() {
continue;
}
@@ -214,7 +128,101 @@ impl TwcsPicker {
}
}
}
output
Ok(output)
}
fn find_inputs(
&self,
region_id: RegionId,
active_window: Option<i64>,
files: &Window,
windows: &BTreeMap<i64, Window>,
) -> (Vec<FileGroup>, bool) {
let window = &files.time_window;
let mut files_to_merge: Vec<_> = files.files().cloned().collect();
// Filter out large files in append mode - they won't benefit from compaction
if self.append_mode
&& let Some(max_size) = self.max_output_file_size
{
let (kept_files, ignored_files) = files_to_merge
.into_iter()
.partition(|fg| fg.size() <= max_size as usize);
files_to_merge = kept_files;
if !ignored_files.is_empty() {
info!(
"Skipped {} large files in append mode for region {}, window {}, max_size: {}",
ignored_files.len(),
region_id,
window,
max_size
);
}
}
let sorted_runs = if files_to_merge.len() < 1024 {
find_sorted_runs(&mut files_to_merge)
} else {
find_sorted_runs_by_time_range(&mut files_to_merge)
};
let found_runs = sorted_runs.len();
// We only remove deletion markers if we found less than 2 runs and not in append mode.
// because after compaction there will be no overlapping files.
let filter_deleted =
found_runs <= 2 && !self.append_mode && !window_has_overlap(files, windows);
if found_runs == 0 {
return (vec![], filter_deleted);
}
let mut inputs = if found_runs > 1 {
reduce_runs(sorted_runs)
} else {
let run = sorted_runs.last().unwrap();
if run.items().len() < self.trigger_file_num {
return (vec![], filter_deleted);
}
// no overlapping files, try merge small files
merge_seq_files(run.items(), self.max_output_file_size)
};
// Limits the number of input files.
let total_input_files: usize = inputs.iter().map(|fg| fg.num_files()).sum();
if total_input_files > DEFAULT_MAX_INPUT_FILE_NUM {
// Sorts file groups by size first.
inputs.sort_unstable_by_key(|fg| fg.size());
let mut num_picked_files = 0;
inputs = inputs
.into_iter()
.take_while(|fg| {
let current_group_file_num = fg.num_files();
if current_group_file_num + num_picked_files <= DEFAULT_MAX_INPUT_FILE_NUM {
num_picked_files += current_group_file_num;
true
} else {
false
}
})
.collect::<Vec<_>>();
info!(
"Compaction for region {} enforces max input file num limit: {}, current total: {}, input: {:?}",
region_id, DEFAULT_MAX_INPUT_FILE_NUM, total_input_files, inputs
);
}
if inputs.len() > 1 {
// If we have more than one group to compact.
log_pick_result(
region_id,
*window,
active_window,
found_runs,
files.files.len(),
self.max_output_file_size,
filter_deleted,
&inputs,
);
}
(inputs, filter_deleted)
}
}
@@ -264,67 +272,77 @@ fn log_pick_result(
);
}
#[async_trait::async_trait]
impl Picker for TwcsPicker {
fn pick(&self, compaction_region: &CompactionRegion) -> Option<PickerOutput> {
async fn pick(&self, compaction_region: &CompactionRegion) -> Result<Option<PickerOutput>> {
let region_id = compaction_region.region_id;
let levels = compaction_region.current_version.ssts.levels();
let expired_ssts =
get_expired_ssts(levels, compaction_region.ttl, Timestamp::current_millis());
if !expired_ssts.is_empty() {
info!("Expired SSTs in region {}: {:?}", region_id, expired_ssts);
}
let expired_file_ids = expired_ssts
.iter()
.map(|file| file.file_id())
.collect::<HashSet<_>>();
let compaction_time_window = compaction_region
.current_version
.compaction_time_window
.map(|window| window.as_secs() as i64);
let time_window_size = compaction_time_window
.or(self.time_window_seconds)
.unwrap_or_else(|| {
let inferred = infer_time_bucket(levels[0].files());
info!(
"Compaction window for region {} is not present, inferring from files: {:?}",
region_id, inferred
let picker = self.clone();
let compaction_region = compaction_region.clone();
let (expired_ssts, time_window_size, active_window, windows) =
common_runtime::spawn_blocking_compact(move || {
let levels = compaction_region.current_version.ssts.levels();
let expired_ssts = get_expired_ssts(
levels,
compaction_region.ttl,
Timestamp::current_millis(),
);
inferred
});
if !expired_ssts.is_empty() {
info!("Expired SSTs in region {}: {:?}", region_id, expired_ssts);
}
let expired_file_ids = expired_ssts
.iter()
.map(|file| file.file_id())
.collect::<HashSet<_>>();
// Find active window from files in level 0.
let active_window = find_latest_window_in_seconds(levels[0].files(), time_window_size);
// Assign files to windows
let mut windows = assign_to_windows(
levels
.iter()
.flat_map(LevelMeta::files)
.filter(|file| !expired_file_ids.contains(&file.file_id())),
time_window_size,
);
let outputs = self.build_output_with_time_range(
region_id,
&mut windows,
active_window,
Some(time_window_size),
);
let compaction_time_window = compaction_region
.current_version
.compaction_time_window
.map(|window| window.as_secs() as i64);
let time_window_size = compaction_time_window
.or(picker.time_window_seconds)
.unwrap_or_else(|| {
let inferred = infer_time_bucket(levels[0].files());
info!(
"Compaction window for region {} is not present, inferring from files: {:?}",
region_id, inferred
);
inferred
});
let active_window =
find_latest_window_in_seconds(levels[0].files(), time_window_size);
let windows = assign_to_windows(
levels
.iter()
.flat_map(LevelMeta::files)
.filter(|file| !expired_file_ids.contains(&file.file_id())),
time_window_size,
);
(expired_ssts, time_window_size, active_window, windows)
})
.await
.context(JoinSnafu)?;
let outputs = self
.build_output_with_time_range(region_id, windows, active_window, Some(time_window_size))
.await?;
if outputs.is_empty() && expired_ssts.is_empty() {
return None;
return Ok(None);
}
let max_file_size = self.max_output_file_size.map(|v| v as usize);
Some(PickerOutput {
Ok(Some(PickerOutput {
outputs,
expired_ssts,
time_window_size,
max_file_size,
})
}))
}
}
#[derive(Clone)]
struct Window {
start: Timestamp,
end: Timestamp,
@@ -558,7 +576,7 @@ mod tests {
};
let compaction_region = compaction_region_with_expired_sst().await;
let output = picker.pick(&compaction_region).unwrap();
let output = picker.pick(&compaction_region).await.unwrap().unwrap();
assert!(output.outputs.is_empty());
assert!(!output.expired_ssts.is_empty());
@@ -936,14 +954,14 @@ mod tests {
}
impl CompactionPickerTestCase {
fn check(&self) {
async fn check(&self) {
let file_id_to_idx = self
.input_files
.iter()
.enumerate()
.map(|(idx, file)| (file.file_id(), idx))
.collect::<HashMap<_, _>>();
let mut windows = assign_to_windows(self.input_files.iter(), self.window_size);
let windows = assign_to_windows(self.input_files.iter(), self.window_size);
let active_window =
find_latest_window_in_seconds(self.input_files.iter(), self.window_size);
let output = TwcsPicker {
@@ -954,7 +972,9 @@ mod tests {
max_background_tasks: None,
time_range: None,
}
.build_output(RegionId::from_u64(0), &mut windows, active_window);
.build_output_with_time_range(RegionId::from_u64(0), windows, active_window, None)
.await
.unwrap();
let output = output
.iter()
@@ -985,8 +1005,8 @@ mod tests {
output_level: Level,
}
#[test]
fn test_build_twcs_output() {
#[tokio::test]
async fn test_build_twcs_output() {
let file_ids = (0..4).map(|_| FileId::random()).collect::<Vec<_>>();
// Case 1: 2 runs found in each time window.
@@ -1010,7 +1030,8 @@ mod tests {
},
],
}
.check();
.check()
.await;
// Case 2:
// -2000........-3
@@ -1040,7 +1061,8 @@ mod tests {
},
],
}
.check();
.check()
.await;
// Case 3:
// A compaction may split output into several files that have overlapping time ranges and same sequence,
@@ -1061,11 +1083,12 @@ mod tests {
output_level: 1,
}],
}
.check();
.check()
.await;
}
#[test]
fn test_build_output_skips_pk_disjoint_files() {
#[tokio::test]
async fn test_build_output_skips_pk_disjoint_files() {
let files = [
new_file_handle_with_size_sequence_and_primary_key_range(
FileId::random(),
@@ -1086,7 +1109,7 @@ mod tests {
pk_range(b"x", b"z"),
),
];
let mut windows = assign_to_windows(files.iter(), 3);
let windows = assign_to_windows(files.iter(), 3);
let active_window = find_latest_window_in_seconds(files.iter(), 3);
let output = TwcsPicker {
trigger_file_num: 4,
@@ -1096,7 +1119,9 @@ mod tests {
max_background_tasks: None,
time_range: None,
}
.build_output(RegionId::from_u64(0), &mut windows, active_window);
.build_output_with_time_range(RegionId::from_u64(0), windows, active_window, None)
.await
.unwrap();
assert!(output.is_empty());
}
@@ -1141,8 +1166,8 @@ mod tests {
}
}
#[test]
fn test_build_output_multiple_windows_with_zero_runs() {
#[tokio::test]
async fn test_build_output_multiple_windows_with_zero_runs() {
let file_ids = (0..6).map(|_| FileId::random()).collect::<Vec<_>>();
let files = [
@@ -1156,7 +1181,7 @@ mod tests {
new_file_handle_with_sequence(file_ids[5], 3000, 3999, 0, 6),
];
let mut windows = assign_to_windows(files.iter(), 3);
let windows = assign_to_windows(files.iter(), 3);
// Create picker with trigger_file_num of 4 so single files won't form runs in first window
let picker = TwcsPicker {
@@ -1169,7 +1194,10 @@ mod tests {
};
let active_window = find_latest_window_in_seconds(files.iter(), 3);
let output = picker.build_output(RegionId::from_u64(123), &mut windows, active_window);
let output = picker
.build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
.await
.unwrap();
assert!(
!output.is_empty(),
@@ -1190,8 +1218,8 @@ mod tests {
);
}
#[test]
fn test_build_output_single_window_zero_runs() {
#[tokio::test]
async fn test_build_output_single_window_zero_runs() {
let file_ids = (0..2).map(|_| FileId::random()).collect::<Vec<_>>();
let large_file_1 = new_file_handle_with_size_and_sequence(file_ids[0], 0, 999, 0, 1, 2000); // 2000 bytes
@@ -1199,7 +1227,7 @@ mod tests {
let files = [large_file_1, large_file_2];
let mut windows = assign_to_windows(files.iter(), 3);
let windows = assign_to_windows(files.iter(), 3);
let picker = TwcsPicker {
trigger_file_num: 2,
@@ -1211,7 +1239,10 @@ mod tests {
};
let active_window = find_latest_window_in_seconds(files.iter(), 3);
let output = picker.build_output(RegionId::from_u64(456), &mut windows, active_window);
let output = picker
.build_output_with_time_range(RegionId::from_u64(456), windows, active_window, None)
.await
.unwrap();
// Should return empty output (no compaction needed)
assert!(
@@ -1220,8 +1251,8 @@ mod tests {
);
}
#[test]
fn test_max_background_tasks_truncation() {
#[tokio::test]
async fn test_max_background_tasks_truncation() {
let file_ids = (0..10).map(|_| FileId::random()).collect::<Vec<_>>();
let max_background_tasks = 3;
@@ -1242,7 +1273,7 @@ mod tests {
new_file_handle_with_sequence(file_ids[9], 6000, 6999, 0, 10),
];
let mut windows = assign_to_windows(files.iter(), 3);
let windows = assign_to_windows(files.iter(), 3);
let picker = TwcsPicker {
trigger_file_num: 4,
@@ -1254,7 +1285,10 @@ mod tests {
};
let active_window = find_latest_window_in_seconds(files.iter(), 3);
let output = picker.build_output(RegionId::from_u64(123), &mut windows, active_window);
let output = picker
.build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
.await
.unwrap();
// Should have at most max_background_tasks outputs
assert!(
@@ -1274,12 +1308,16 @@ mod tests {
time_range: None,
};
let mut windows_no_limit = assign_to_windows(files.iter(), 3);
let output_no_limit = picker_no_limit.build_output(
RegionId::from_u64(123),
&mut windows_no_limit,
active_window,
);
let windows_no_limit = assign_to_windows(files.iter(), 3);
let output_no_limit = picker_no_limit
.build_output_with_time_range(
RegionId::from_u64(123),
windows_no_limit,
active_window,
None,
)
.await
.unwrap();
// Without limit, should have more outputs (if there are enough windows)
if output_no_limit.len() > max_background_tasks {
@@ -1290,8 +1328,8 @@ mod tests {
}
}
#[test]
fn test_max_background_tasks_no_truncation_when_under_limit() {
#[tokio::test]
async fn test_max_background_tasks_no_truncation_when_under_limit() {
let file_ids = (0..4).map(|_| FileId::random()).collect::<Vec<_>>();
let max_background_tasks = 10; // Larger than expected outputs
@@ -1303,7 +1341,7 @@ mod tests {
new_file_handle_with_sequence(file_ids[3], 0, 999, 0, 4),
];
let mut windows = assign_to_windows(files.iter(), 3);
let windows = assign_to_windows(files.iter(), 3);
let picker = TwcsPicker {
trigger_file_num: 4,
@@ -1315,7 +1353,10 @@ mod tests {
};
let active_window = find_latest_window_in_seconds(files.iter(), 3);
let output = picker.build_output(RegionId::from_u64(123), &mut windows, active_window);
let output = picker
.build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
.await
.unwrap();
// Should have all outputs since we're under the limit
assert!(
@@ -1326,8 +1367,8 @@ mod tests {
assert!(!output.is_empty(), "Should have at least one output");
}
#[test]
fn test_pick_multiple_runs() {
#[tokio::test]
async fn test_pick_multiple_runs() {
common_telemetry::init_default_ut_logging();
let num_files = 8;
@@ -1349,7 +1390,7 @@ mod tests {
})
.collect();
let mut windows = assign_to_windows(files.iter(), 3);
let windows = assign_to_windows(files.iter(), 3);
let picker = TwcsPicker {
trigger_file_num: 4,
@@ -1361,14 +1402,17 @@ mod tests {
};
let active_window = find_latest_window_in_seconds(files.iter(), 3);
let output = picker.build_output(RegionId::from_u64(123), &mut windows, active_window);
let output = picker
.build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
.await
.unwrap();
assert_eq!(1, output.len());
assert_eq!(output[0].inputs.len(), 2);
}
#[test]
fn test_limit_max_input_files() {
#[tokio::test]
async fn test_limit_max_input_files() {
common_telemetry::init_default_ut_logging();
let num_files = 50;
@@ -1390,7 +1434,7 @@ mod tests {
})
.collect();
let mut windows = assign_to_windows(files.iter(), 3);
let windows = assign_to_windows(files.iter(), 3);
let picker = TwcsPicker {
trigger_file_num: 4,
@@ -1402,7 +1446,10 @@ mod tests {
};
let active_window = find_latest_window_in_seconds(files.iter(), 3);
let output = picker.build_output(RegionId::from_u64(123), &mut windows, active_window);
let output = picker
.build_output_with_time_range(RegionId::from_u64(123), windows, active_window, None)
.await
.unwrap();
assert_eq!(1, output.len());
assert_eq!(output[0].inputs.len(), 32);
@@ -1433,8 +1480,8 @@ mod tests {
assert!(!time_window_intersects_range(0, 4, &overflowing_range));
}
#[test]
fn test_time_range_filter_precedes_background_task_limit() {
#[tokio::test]
async fn test_time_range_filter_precedes_background_task_limit() {
let early_file_ids = [FileId::random(), FileId::random()];
let selected_file_ids = [FileId::random(), FileId::random()];
let files = [
@@ -1443,7 +1490,7 @@ mod tests {
new_file_handle_with_sequence(selected_file_ids[0], 7_000, 7_999, 0, 3),
new_file_handle_with_sequence(selected_file_ids[1], 7_000, 7_999, 0, 4),
];
let mut windows = assign_to_windows(files.iter(), 3);
let windows = assign_to_windows(files.iter(), 3);
let picker = TwcsPicker {
trigger_file_num: 2,
time_window_seconds: Some(3),
@@ -1456,12 +1503,10 @@ mod tests {
),
};
let output = picker.build_output_with_time_range(
RegionId::from_u64(123),
&mut windows,
Some(9),
Some(3),
);
let output = picker
.build_output_with_time_range(RegionId::from_u64(123), windows, Some(9), Some(3))
.await
.unwrap();
assert_eq!(1, output.len());
assert_eq!(
src/mito2/src/compaction/window.rs
+16 -9
View File
@@ -20,18 +20,20 @@ use common_time::Timestamp;
use common_time::range::TimestampRange;
use common_time::timestamp::TimeUnit;
use common_time::timestamp_millis::BucketAligned;
use snafu::ResultExt;
use store_api::storage::RegionId;
use crate::compaction::CompactionOutput;
use crate::compaction::buckets::infer_time_bucket;
use crate::compaction::compactor::{CompactionRegion, CompactionVersion};
use crate::compaction::picker::{Picker, PickerOutput, get_expired_ssts};
use crate::error::{JoinSnafu, Result};
use crate::sst::file::FileHandle;
/// Compaction picker that splits the time range of all involved files to windows, and merges
/// the data segments intersects with those windows of files together so that the output files
/// never overlaps.
#[derive(Debug)]
#[derive(Clone, Debug)]
pub struct WindowedCompactionPicker {
compaction_time_window_seconds: Option<i64>,
time_range: Option<TimestampRange>,
@@ -115,20 +117,25 @@ impl WindowedCompactionPicker {
}
}
#[async_trait::async_trait]
impl Picker for WindowedCompactionPicker {
fn pick(&self, compaction_region: &CompactionRegion) -> Option<PickerOutput> {
let (outputs, expired_ssts, time_window) = self.pick_inner(
compaction_region.current_version.metadata.region_id,
&compaction_region.current_version,
Timestamp::current_millis(),
);
async fn pick(&self, compaction_region: &CompactionRegion) -> Result<Option<PickerOutput>> {
let picker = self.clone();
let region_id = compaction_region.current_version.metadata.region_id;
let current_version = compaction_region.current_version.clone();
let (outputs, expired_ssts, time_window) =
common_runtime::spawn_blocking_compact(move || {
picker.pick_inner(region_id, &current_version, Timestamp::current_millis())
})
.await
.context(JoinSnafu)?;
Some(PickerOutput {
Ok(Some(PickerOutput {
outputs,
expired_ssts,
time_window_size: time_window,
max_file_size: None, // todo (hl): we may need to support `max_file_size` parameter in manual compaction.
})
}))
}
}