diff --git a/config/config.md b/config/config.md index a05c08f770..90c346d09e 100644 --- a/config/config.md +++ b/config/config.md @@ -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.
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.
When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the
`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.
Options: "wait" (default, 10s timeout), "wait()" (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.
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.
When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the
`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.
Defaults to max(num_cpus / 2, 1). | | `backend_tls` | -- | -- | 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

Note: if TLS is configured in both this section and the `store_addrs` connection string, the
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
- "disable" - No TLS
- "prefer" (default) - Try TLS, fallback to plain
- "require" - Require TLS
- "verify_ca" - Require TLS and verify CA
- "verify_full" - Require TLS and verify hostname | | `backend_tls.cert_path` | String | `""` | Path to client certificate file (for client authentication)
Like "/path/to/client.crt" | @@ -500,7 +503,8 @@ | `grpc.tls.watch` | Bool | `false` | Watch for Certificate and key file change and auto reload.
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.
Defaults to max(num_cpus / 2, 1). | | `runtime.query_rt_size` | Integer | `7` | The number of threads to execute datanode query operations.
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. | diff --git a/config/datanode.example.toml b/config/datanode.example.toml index 9b5fc2e9a4..9b7b427254 100644 --- a/config/datanode.example.toml +++ b/config/datanode.example.toml @@ -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 diff --git a/config/frontend.example.toml b/config/frontend.example.toml index d9df1f0969..f495f3b19f 100644 --- a/config/frontend.example.toml +++ b/config/frontend.example.toml @@ -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] diff --git a/config/metasrv.example.toml b/config/metasrv.example.toml index c4b44ca38e..ff7f68037f 100644 --- a/config/metasrv.example.toml +++ b/config/metasrv.example.toml @@ -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 diff --git a/config/standalone.example.toml b/config/standalone.example.toml index fccfe28695..4abbf815c0 100644 --- a/config/standalone.example.toml +++ b/config/standalone.example.toml @@ -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] diff --git a/src/cmd/tests/load_config_test.rs b/src/cmd/tests/load_config_test.rs index d55a3f233b..bdbaf720b7 100644 --- a/src/cmd/tests/load_config_test.rs +++ b/src/cmd/tests/load_config_test.rs @@ -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 = 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() { diff --git a/src/common/runtime/src/global.rs b/src/common/runtime/src/global.rs index e3f9b93280..e28eaf61d1 100644 --- a/src/common/runtime/src/global.rs +++ b/src/common/runtime/src/global.rs @@ -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 }); diff --git a/src/mito2/src/compaction/picker.rs b/src/mito2/src/compaction/picker.rs index be82730507..a207a390a4 100644 --- a/src/mito2/src/compaction/picker.rs +++ b/src/mito2/src/compaction/picker.rs @@ -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; + async fn pick(&self, compaction_region: &CompactionRegion) -> Result>; } /// PickerOutput is the output of a [`Picker`]. diff --git a/src/mito2/src/compaction/scheduler/planning.rs b/src/mito2/src/compaction/scheduler/planning.rs index e9d31727bf..ad09fb2f69 100644 --- a/src/mito2/src/compaction/scheduler/planning.rs +++ b/src/mito2/src/compaction/scheduler/planning.rs @@ -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)), }; diff --git a/src/mito2/src/compaction/twcs.rs b/src/mito2/src/compaction/twcs.rs index d41b9091ad..abe6a81cd9 100644 --- a/src/mito2/src/compaction/twcs.rs +++ b/src/mito2/src/compaction/twcs.rs @@ -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, - active_window: Option, - ) -> Vec { - 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, + time_windows: BTreeMap, active_window: Option, time_window_size: Option, - ) -> Vec { - let find_inputs = |files: &Window, - windows: &BTreeMap| - -> (Vec, 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::>(); - 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> { let mut output = vec![]; let windows = time_windows .values() @@ -184,14 +85,27 @@ impl TwcsPicker { }) }) }) + .map(|window| window.time_window) .collect::>(); + 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::>() - { + 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, + files: &Window, + windows: &BTreeMap, + ) -> (Vec, 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::>(); + 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 { + async fn pick(&self, compaction_region: &CompactionRegion) -> Result> { 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::>(); - - 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::>(); - // 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::>(); - 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::>(); // 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::>(); 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::>(); 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::>(); 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::>(); 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!( diff --git a/src/mito2/src/compaction/window.rs b/src/mito2/src/compaction/window.rs index dd3b3d56cd..422d600a57 100644 --- a/src/mito2/src/compaction/window.rs +++ b/src/mito2/src/compaction/window.rs @@ -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, time_range: Option, @@ -115,20 +117,25 @@ impl WindowedCompactionPicker { } } +#[async_trait::async_trait] impl Picker for WindowedCompactionPicker { - fn pick(&self, compaction_region: &CompactionRegion) -> Option { - 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> { + 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, ¤t_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. - }) + })) } }