diff --git a/src/common/meta/src/ddl/alter_database.rs b/src/common/meta/src/ddl/alter_database.rs index ac24ff77e1..5b22183608 100644 --- a/src/common/meta/src/ddl/alter_database.rs +++ b/src/common/meta/src/ddl/alter_database.rs @@ -20,6 +20,7 @@ use common_procedure::{ use common_telemetry::tracing::info; use serde::{Deserialize, Serialize}; use snafu::{ResultExt, ensure}; +use store_api::mito_engine_options::{TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_TRIGGER_FILE_NUM}; use strum::AsRefStr; use crate::cache_invalidator::Context; @@ -39,6 +40,14 @@ pub struct AlterDatabaseProcedure { pub data: AlterDatabaseData, } +fn twcs_trigger_alias(key: &str) -> Option<&'static str> { + match key { + TWCS_TRIGGER_FILE_NUM => Some(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM), + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM => Some(TWCS_TRIGGER_FILE_NUM), + _ => None, + } +} + fn build_new_schema_value( mut value: SchemaNameValue, alter_kind: &AlterDatabaseKind, @@ -51,7 +60,20 @@ fn build_new_schema_value( value.ttl = Some(*ttl); } SetDatabaseOption::Other(key, val) => { - value.extra_options.insert(key.clone(), val.clone()); + // Keep the legacy key so older versions can read it after a downgrade. + // Persisting both aliases would deserialize as a duplicate field. + let persisted_key = if twcs_trigger_alias(key).is_some() { + value.extra_options.remove(TWCS_TRIGGER_FILE_NUM); + value + .extra_options + .remove(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM); + TWCS_TRIGGER_FILE_NUM + } else { + key + }; + value + .extra_options + .insert(persisted_key.to_string(), val.clone()); } } } @@ -62,6 +84,9 @@ fn build_new_schema_value( UnsetDatabaseOption::Ttl => value.ttl = None, UnsetDatabaseOption::Other(key) => { value.extra_options.remove(key); + if let Some(alias) = twcs_trigger_alias(key) { + value.extra_options.remove(alias); + } } } } @@ -235,6 +260,10 @@ impl AlterDatabaseData { mod tests { use std::time::Duration; + use store_api::mito_engine_options::{ + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_TRIGGER_FILE_NUM, + }; + use crate::ddl::alter_database::build_new_schema_value; use crate::key::schema_name::SchemaNameValue; use crate::rpc::ddl::{ @@ -297,4 +326,63 @@ mod tests { Some(&"1d".to_string()) ); } + + #[test] + fn test_set_twcs_trigger_persists_legacy_key() { + for key in [TWCS_TRIGGER_FILE_NUM, TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM] { + let mut current_schema_value = SchemaNameValue::default(); + current_schema_value.extra_options.insert( + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM.to_string(), + "8".to_string(), + ); + let set = AlterDatabaseKind::SetDatabaseOptions(SetDatabaseOptions(vec![ + SetDatabaseOption::Other(key.to_string(), "16".to_string()), + ])); + + let new_schema_value = build_new_schema_value(current_schema_value, &set).unwrap(); + + assert_eq!( + new_schema_value + .extra_options + .get(TWCS_TRIGGER_FILE_NUM) + .map(String::as_str), + Some("16") + ); + assert!( + !new_schema_value + .extra_options + .contains_key(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM) + ); + } + } + + #[test] + fn test_unset_twcs_trigger_removes_both_aliases() { + for key in [TWCS_TRIGGER_FILE_NUM, TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM] { + let mut current_schema_value = SchemaNameValue::default(); + current_schema_value + .extra_options + .insert(TWCS_TRIGGER_FILE_NUM.to_string(), "8".to_string()); + current_schema_value.extra_options.insert( + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM.to_string(), + "16".to_string(), + ); + let unset = AlterDatabaseKind::UnsetDatabaseOptions(UnsetDatabaseOptions(vec![ + UnsetDatabaseOption::Other(key.to_string()), + ])); + + let new_schema_value = build_new_schema_value(current_schema_value, &unset).unwrap(); + + assert!( + !new_schema_value + .extra_options + .contains_key(TWCS_TRIGGER_FILE_NUM) + ); + assert!( + !new_schema_value + .extra_options + .contains_key(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM) + ); + } + } } diff --git a/src/common/meta/src/error.rs b/src/common/meta/src/error.rs index aa27227c7c..3bd1475cb4 100644 --- a/src/common/meta/src/error.rs +++ b/src/common/meta/src/error.rs @@ -325,6 +325,22 @@ pub enum Error { location: Location, }, + #[snafu(display( + "Conflicting schema options: {}={} and {}={}", + first_key, + first_value, + second_key, + second_value + ))] + ConflictingSchemaOptions { + first_key: String, + first_value: String, + second_key: String, + second_value: String, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Corrupted table route data, err: {}", err_msg))] RouteInfoCorrupted { err_msg: String, @@ -1235,7 +1251,8 @@ impl ErrorExt for Error { | InvalidFileExtension { .. } | InvalidFileName { .. } | InvalidFlowRequestBody { .. } - | InvalidFilePath { .. } => StatusCode::InvalidArguments, + | InvalidFilePath { .. } + | ConflictingSchemaOptions { .. } => StatusCode::InvalidArguments, #[cfg(feature = "enterprise")] MissingInterval { .. } | NegativeDuration { .. } | TooLargeDuration { .. } => { diff --git a/src/common/meta/src/key/schema_name.rs b/src/common/meta/src/key/schema_name.rs index 5c0370f0a3..689f72ddb3 100644 --- a/src/common/meta/src/key/schema_name.rs +++ b/src/common/meta/src/key/schema_name.rs @@ -21,9 +21,14 @@ use futures::stream::BoxStream; use humantime_serde::re::humantime; use serde::{Deserialize, Serialize}; use snafu::{OptionExt, ResultExt, ensure}; +use store_api::mito_engine_options::{ + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_TRIGGER_FILE_NUM, normalize_twcs_trigger_options, +}; use crate::ensure_values; -use crate::error::{self, Error, InvalidMetadataSnafu, ParseOptionSnafu, Result}; +use crate::error::{ + self, ConflictingSchemaOptionsSnafu, Error, InvalidMetadataSnafu, ParseOptionSnafu, Result, +}; use crate::key::txn_helper::TxnOpGetResponseSet; use crate::key::{ DeserializedValueWithBytes, MetadataKey, SCHEMA_NAME_KEY_PATTERN, SCHEMA_NAME_KEY_PREFIX, @@ -82,6 +87,17 @@ impl TryFrom<&HashMap> for SchemaNameValue { type Error = Error; fn try_from(value: &HashMap) -> std::result::Result { + let mut value = value.clone(); + normalize_twcs_trigger_options(&mut value).map_err(|conflict| { + ConflictingSchemaOptionsSnafu { + first_key: TWCS_TRIGGER_FILE_NUM, + first_value: conflict.legacy_value, + second_key: TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, + second_value: conflict.canonical_value, + } + .build() + })?; + let ttl = value .get(OPT_KEY_TTL) .map(|ttl_str| { @@ -340,6 +356,12 @@ mod tests { use std::sync::Arc; use std::time::Duration; + use common_error::ext::ErrorExt; + use common_error::status_code::StatusCode; + use store_api::mito_engine_options::{ + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_TRIGGER_FILE_NUM, + }; + use super::*; use crate::kv_backend::memory::MemoryKvBackend; @@ -410,6 +432,48 @@ mod tests { assert!(err_empty.is_err()); } + #[test] + fn test_schema_value_normalizes_twcs_trigger_aliases() { + for options in [ + HashMap::from([( + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM.to_string(), + "4".to_string(), + )]), + HashMap::from([ + (TWCS_TRIGGER_FILE_NUM.to_string(), "4".to_string()), + ( + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM.to_string(), + "4".to_string(), + ), + ]), + ] { + let value = SchemaNameValue::try_from(&options).unwrap(); + assert_eq!( + BTreeMap::from([(TWCS_TRIGGER_FILE_NUM.to_string(), "4".to_string())]), + value.extra_options + ); + } + } + + #[test] + fn test_schema_value_rejects_conflicting_twcs_trigger_aliases() { + let options = HashMap::from([ + (TWCS_TRIGGER_FILE_NUM.to_string(), "4".to_string()), + ( + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM.to_string(), + "8".to_string(), + ), + ]); + + let error = SchemaNameValue::try_from(&options).unwrap_err(); + + assert_eq!(StatusCode::InvalidArguments, error.status_code()); + assert_eq!( + "Conflicting schema options: compaction.twcs.trigger_file_num=4 and compaction.twcs.active_window.trigger_file_num=8", + error.to_string() + ); + } + #[test] fn test_extra_options_compatibility() { // Test with extra_options only diff --git a/src/common/meta/src/rpc/ddl.rs b/src/common/meta/src/rpc/ddl.rs index 16ac82be19..9e88527f93 100644 --- a/src/common/meta/src/rpc/ddl.rs +++ b/src/common/meta/src/rpc/ddl.rs @@ -52,7 +52,7 @@ use serde::{Deserialize, Serialize}; use serde_with::{DefaultOnNull, serde_as}; use snafu::{OptionExt, ResultExt}; use table::metadata::{TableId, TableInfo}; -use table::requests::validate_database_option; +use table::requests::{validate_database_option, validate_database_option_value}; use table::table_name::TableName; use table::table_reference::TableReference; @@ -1222,7 +1222,9 @@ impl TryFrom for SetDatabaseOption { Ok(SetDatabaseOption::Ttl(ttl)) } _ => { - if validate_database_option(&key_lower) { + if validate_database_option(&key_lower) + && validate_database_option_value(&key_lower, Some(&value)).is_ok() + { Ok(SetDatabaseOption::Other(key_lower, value)) } else { InvalidSetDatabaseOptionSnafu { key, value }.fail() @@ -1901,6 +1903,43 @@ mod tests { ); } + #[test] + fn test_alter_database_rejects_invalid_trigger_values() { + let overflow = format!("{}0", usize::MAX); + for key in [ + "compaction.twcs.trigger_file_num", + "compaction.twcs.active_window.trigger_file_num", + "compaction.twcs.inactive_window.trigger_file_num", + "compaction.twcs.active_window.l1_merge_trigger", + "compaction.twcs.inactive_window.l1_merge_trigger", + ] { + for invalid in ["invalid", "-1", overflow.as_str()] { + let kind = PbAlterDatabaseKind::SetDatabaseOptions(api::v1::SetDatabaseOptions { + set_database_options: vec![PbOption { + key: key.to_string(), + value: invalid.to_string(), + }], + }); + let err = AlterDatabaseKind::try_from(kind).unwrap_err(); + assert!( + matches!(err, error::Error::InvalidSetDatabaseOption { .. }), + "{key}: {invalid}" + ); + } + for boundary in ["0", "1", "2"] { + let option = PbOption { + key: key.to_string(), + value: boundary.to_string(), + }; + assert_eq!( + SetDatabaseOption::try_from(option).is_ok(), + !key.ends_with("l1_merge_trigger") || boundary == "2", + "{key}: {boundary}" + ); + } + } + } + #[test] fn test_basic_ser_de_create_table_task() { let schema = SchemaBuilder::default().build().unwrap(); diff --git a/src/mito2/src/compaction.rs b/src/mito2/src/compaction.rs index cc98c3083e..e7f74b2264 100644 --- a/src/mito2/src/compaction.rs +++ b/src/mito2/src/compaction.rs @@ -41,6 +41,7 @@ pub(crate) use scheduler::{ }; use serde::{Deserialize, Serialize}; use snafu::ResultExt; +use store_api::mito_engine_options::{TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_TRIGGER_FILE_NUM}; use store_api::storage::RegionId; use crate::error::{GetSchemaMetadataSnafu, Result, TimeoutSnafu}; @@ -85,7 +86,7 @@ async fn find_dynamic_options( let compaction = if !region_options.compaction_override { if let Some(schema_opts) = db_options { - let map: HashMap = schema_opts + let mut map: HashMap = schema_opts .extra_options .iter() .filter_map(|(k, v)| { @@ -96,6 +97,10 @@ async fn find_dynamic_options( } }) .collect(); + // Historical metadata may contain both aliases; prefer the canonical key. + if map.contains_key(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM) { + map.remove(TWCS_TRIGGER_FILE_NUM); + } if map.is_empty() { region_options.compaction.clone() } else { diff --git a/src/mito2/src/compaction/picker.rs b/src/mito2/src/compaction/picker.rs index a207a390a4..693422955f 100644 --- a/src/mito2/src/compaction/picker.rs +++ b/src/mito2/src/compaction/picker.rs @@ -143,7 +143,10 @@ pub fn new_picker( } else { match compaction_options { CompactionOptions::Twcs(twcs_opts) => Arc::new(TwcsPicker { - trigger_file_num: twcs_opts.trigger_file_num, + trigger_file_num: twcs_opts.active_window_trigger_file_num, + active_window_l1_merge_trigger: twcs_opts.active_window_l1_merge_trigger, + inactive_window_trigger_file_num: twcs_opts.inactive_window_trigger_file_num, + inactive_window_l1_merge_trigger: twcs_opts.inactive_window_l1_merge_trigger, time_window_seconds: twcs_opts.time_window_seconds(), max_output_file_size: twcs_opts.max_output_file_size.map(|r| r.as_bytes()), append_mode, diff --git a/src/mito2/src/compaction/scheduler.rs b/src/mito2/src/compaction/scheduler.rs index 6aebb25eeb..ff92f8a41c 100644 --- a/src/mito2/src/compaction/scheduler.rs +++ b/src/mito2/src/compaction/scheduler.rs @@ -267,8 +267,8 @@ impl CompactionScheduler { /// Notifies waiters, schedules a pending manual or automatic follow-up, or /// removes the region status. A follow-up is scheduled whenever the cycle /// latched an automatic trigger, or when `made_progress` is true (the - /// execution removed more files than it added): successful compaction keeps - /// draining the region until the picker returns no plan. The returned + /// execution produced output or reduced the file count): successful compaction + /// keeps draining the region until the picker returns no plan. The returned /// transition reports a dispatched automatic follow-up or DDLs that are now /// safe to execute. /// @@ -617,8 +617,8 @@ impl CompactionScheduler { } // Keep draining when the cycle latched an automatic trigger, or when the - // execution reduced the file count. A no-progress rewrite stops here so a - // split-heavy output cannot loop forever; the next flush trigger resumes. + // execution produced output or reduced the file count. An empty edit stops + // here; the next flush trigger resumes. let should_continue = status.active.reset_automatic_followup() || made_progress; if should_continue && self.schedule_automatic_followup(region_id, manifest_ctx, schema_metadata_manager) diff --git a/src/mito2/src/compaction/scheduler/planning.rs b/src/mito2/src/compaction/scheduler/planning.rs index 0f1537df2b..facb1bb1d3 100644 --- a/src/mito2/src/compaction/scheduler/planning.rs +++ b/src/mito2/src/compaction/scheduler/planning.rs @@ -203,11 +203,14 @@ impl CompactionScheduler { ) }); + // Avoid queuing serial outputs from a stale snapshot. Repicking after each + // batch lets newly flushed L0 files outrank L1 work that has not started. + let max_picker_outputs = max_background_compactions.min(request.max_parallelism.max(1)); let picker = new_picker( &options, &dynamic_compaction_opts, request.current_version.options.append_mode, - Some(max_background_compactions), + Some(max_picker_outputs), time_range, ); let region_id = request.region_id(); diff --git a/src/mito2/src/compaction/scheduler_test.rs b/src/mito2/src/compaction/scheduler_test.rs index 9545d0eea2..5afcc9c37f 100644 --- a/src/mito2/src/compaction/scheduler_test.rs +++ b/src/mito2/src/compaction/scheduler_test.rs @@ -20,6 +20,9 @@ use api::v1::region::compact_request::Options; use common_datasource::compression::CompressionType; use common_meta::key::schema_name::SchemaNameValue; use common_time::{DatabaseTimeToLive, Timestamp}; +use store_api::mito_engine_options::{ + COMPACTION_TYPE, TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_TRIGGER_FILE_NUM, +}; use store_api::storage::FileId; use tokio::sync::{Barrier, mpsc, oneshot}; @@ -30,6 +33,7 @@ use crate::compaction::scheduler::state::{CompactingFiles, CompactionPhase}; use crate::compaction::scheduler::*; use crate::compaction::test_util::new_file_handle; use crate::compaction::{CompactionOutput, find_dynamic_options}; +use crate::config::MitoConfig; use crate::error::InvalidSchedulerStateSnafu; use crate::manifest::manager::{RegionManifestManager, RegionManifestOptions}; use crate::metrics::COMPACTION_MEMORY_REJECTED; @@ -207,6 +211,14 @@ async fn test_find_compaction_options_db_level() { schema_value .extra_options .insert("compaction.twcs.time_window".to_string(), "2h".to_string()); + schema_value.extra_options.insert( + "compaction.twcs.active_window.l1_merge_trigger".to_string(), + "12".to_string(), + ); + schema_value.extra_options.insert( + "compaction.twcs.inactive_window.l1_merge_trigger".to_string(), + "14".to_string(), + ); schema_metadata_manager .register_region_table_info( table_id, @@ -226,10 +238,50 @@ async fn test_find_compaction_options_db_level() { match opts { crate::region::options::CompactionOptions::Twcs(t) => { assert_eq!(t.time_window_seconds(), Some(2 * 3600)); + assert_eq!(t.active_window_l1_merge_trigger, 12); + assert_eq!(t.inactive_window_l1_merge_trigger, 14); } } } +#[tokio::test] +async fn test_find_compaction_options_db_level_prefers_canonical_trigger_alias() { + let builder = VersionControlBuilder::new(); + let (schema_metadata_manager, kv_backend) = mock_schema_metadata_manager(); + let region_id = builder.region_id(); + let table_id = region_id.table_id(); + let mut schema_value = SchemaNameValue::default(); + schema_value + .extra_options + .insert(COMPACTION_TYPE.to_string(), "twcs".to_string()); + schema_value + .extra_options + .insert(TWCS_TRIGGER_FILE_NUM.to_string(), "7".to_string()); + schema_value.extra_options.insert( + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM.to_string(), + "9".to_string(), + ); + schema_metadata_manager + .register_region_table_info( + table_id, + "t", + "c", + "s", + Some(schema_value), + kv_backend.clone(), + ) + .await; + + let version_control = Arc::new(builder.build()); + let region_opts = version_control.current().version.options.clone(); + let (opts, _) = find_dynamic_options(region_id, ®ion_opts, &schema_metadata_manager) + .await + .unwrap(); + + let crate::region::options::CompactionOptions::Twcs(twcs) = opts; + assert_eq!(twcs.active_window_trigger_file_num, 9); +} + #[tokio::test] async fn test_find_compaction_options_priority() { fn schema_value_with_twcs(time_window: &str) -> SchemaNameValue { @@ -969,15 +1021,15 @@ async fn test_execution_finished_drains_until_no_plan() { .map(|file| file.meta_ref().clone()) .collect(); - // 5 files for next compaction and removes old files. + // Replace the 5 old files with 5 non-empty output files. apply_edit( &version_control, &[(0, end), (20, end), (40, end), (60, end), (80, end)], &file_metas, purger.clone(), ); - // A completed execution that reduced the file count chains another pick even - // without an explicit trigger, because the layout is still compactable. + // A completed execution with non-empty output made semantic progress, so it + // chains another pick even without an explicit trigger. let transition = scheduler .on_compaction_finished( region_id, @@ -1073,9 +1125,8 @@ async fn test_execution_finished_without_progress_removes_status() { .await; assert_eq!(1, job_scheduler.num_jobs()); - // The execution rewrote files without reducing the file count (e.g. its output - // was split into more files than its input). Chaining would loop without making - // progress, so the lifecycle ends here; the next flush trigger resumes compaction. + // The execution produced an empty edit: no files were removed or added. The + // lifecycle ends here; the next flush trigger resumes compaction. let transition = scheduler .on_compaction_finished( region_id, @@ -1112,7 +1163,7 @@ async fn test_time_range_compaction_when_compaction_in_progress() { ) .await; - // 40 files to compact. The first task picks 32, leaving 8 for the pending request. + // 40 files to compact. The first task picks 16, leaving 24 for the pending request. let end = 1000 * 1000; for offset in 0..40 { builder.push_l0_file(offset * 10, end); @@ -1504,6 +1555,69 @@ async fn test_automatic_trigger_during_execution_clears_continuation_scope() { ); } +#[tokio::test] +async fn test_automatic_planning_picks_one_output_before_repicking() { + let env = SchedulerEnv::new().await; + let (tx, mut rx) = mpsc::channel(4); + let config = MitoConfig { + max_background_compactions: 4, + ..Default::default() + }; + let mut scheduler = env.mock_compaction_scheduler_with_config(tx, config); + + let mut builder = VersionControlBuilder::new(); + for window_start in [0, 2 * 3_600_000] { + for offset in [0, 10, 20, 30] { + builder.push_l0_file(window_start + offset, window_start + 1_000); + } + } + let region_id = builder.region_id(); + let version_control = Arc::new(builder.build()); + let manifest_ctx = env + .mock_manifest_context(version_control.current().version.metadata.clone()) + .await; + let (schema_metadata_manager, kv_backend) = mock_schema_metadata_manager(); + let mut schema_value = SchemaNameValue::default(); + schema_value + .extra_options + .insert("compaction.type".to_string(), "twcs".to_string()); + schema_value + .extra_options + .insert("compaction.twcs.time_window".to_string(), "1h".to_string()); + schema_metadata_manager + .register_region_table_info( + region_id.table_id(), + "t", + "c", + "s", + Some(schema_value), + kv_backend, + ) + .await; + + scheduler + .schedule_automatic_compaction( + compact_request::Options::Regular(Default::default()), + &version_control, + &env.access_layer, + &manifest_ctx, + schema_metadata_manager, + ) + .unwrap(); + let finished = recv_compaction_pick_finished(&mut rx).await; + let CompactionPlanningResult::Prepared(prepared) = finished.result else { + panic!("expected prepared compaction"); + }; + + assert_eq!(1, prepared.picker_output.outputs.len()); + assert!( + prepared.picker_output.outputs[0] + .inputs + .iter() + .all(|file| { file.time_range().0 >= Timestamp::new_millisecond(2 * 3_600_000) }) + ); +} + #[tokio::test] async fn test_compaction_bypass_in_staging_mode() { let env = SchedulerEnv::new().await; diff --git a/src/mito2/src/compaction/twcs.rs b/src/mito2/src/compaction/twcs.rs index 349d6259c4..2953ded896 100644 --- a/src/mito2/src/compaction/twcs.rs +++ b/src/mito2/src/compaction/twcs.rs @@ -40,11 +40,42 @@ use crate::sst::version::LevelMeta; const LEVEL_COMPACTED: Level = 1; +#[derive(Clone, Copy, Debug)] +enum PickPhase { + HasL0, + L1FileReduction, + L1OverlapOnly, +} + +impl PickPhase { + fn applies_to(self, window: &Window) -> bool { + match self { + Self::HasL0 => window.files().any(|file| file.level() == 0), + Self::L1FileReduction | Self::L1OverlapOnly => { + window.files().any(|file| file.level() != 0) + } + } + } +} + +const PICK_PHASES: [PickPhase; 3] = [ + PickPhase::HasL0, + PickPhase::L1FileReduction, + PickPhase::L1OverlapOnly, +]; + +struct WindowPickContext<'a> { + active_window: Option, + files: &'a Window, + windows: &'a BTreeMap, + phase: PickPhase, +} + /// A mixed L0/L1 compaction may rewrite at most this many L1 rows per L0 row. const MAX_L1_L0_ROW_RATIO: usize = 2; /// Default maximum number of input SST files in one compaction input. -const DEFAULT_MAX_INPUT_FILES: usize = 32; +const DEFAULT_MAX_INPUT_FILES: usize = 16; const MAX_INPUT_FILES_ENV: &str = "GREPTIME_TWCS_MAX_INPUT_FILES"; @@ -66,8 +97,14 @@ fn parse_max_input_files(env_value: Option<&str>) -> usize { /// candidates. #[derive(Clone, Debug)] pub struct TwcsPicker { - /// Minimum file num to trigger a compaction. + /// Minimum file num to trigger a compaction in the active window. pub trigger_file_num: usize, + /// Minimum L1 file num to allow a safety compaction in the active window. + pub active_window_l1_merge_trigger: usize, + /// Minimum file num to trigger a compaction in an inactive window. + pub inactive_window_trigger_file_num: usize, + /// Minimum L1 file num to trigger a compaction in an inactive window. + pub inactive_window_l1_merge_trigger: usize, /// Compaction time window in seconds. pub time_window_seconds: Option, /// Max allowed compaction output file size. The picker also uses it to predict @@ -109,59 +146,88 @@ impl TwcsPicker { .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) { - 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; + let mut selected_windows = HashSet::new(); + 'phases: for phase in PICK_PHASES { + for chunk in windows.chunks(chunk_size) { + let mut handles = Vec::with_capacity(chunk.len()); + for window in chunk { + if selected_windows.contains(window) { + continue; + } + if !time_windows + .get(window) + .is_some_and(|files| phase.applies_to(files)) + { + continue; + } + 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(|files| { + ( + window, + picker.find_inputs( + region_id, + WindowPickContext { + active_window, + files, + windows: &time_windows, + phase, + }, + ), + ) + }) + })); + tokio::task::yield_now().await; } + for result in futures::future::join_all(handles).await { + let Some((window, (inputs, filter_deleted))) = result.context(JoinSnafu)? + else { + continue; + }; + if inputs.is_empty() { + continue; + } - output.push(CompactionOutput { - output_level: LEVEL_COMPACTED, // always compact to l1 - inputs, - filter_deleted, - output_time_range: None, // we do not enforce output time range in twcs compactions. - }); + selected_windows.insert(window); + output.push(CompactionOutput { + output_level: LEVEL_COMPACTED, // always compact to l1 + inputs, + filter_deleted, + output_time_range: None, // we do not enforce output time range in twcs compactions. + }); - if let Some(max_background_tasks) = self.max_background_tasks - && output.len() >= max_background_tasks - { - debug!( - "Region ({:?}) compaction task size larger than max background tasks({}), remaining tasks discarded", - region_id, max_background_tasks - ); - break 'chunks; + if let Some(max_background_tasks) = self.max_background_tasks + && output.len() >= max_background_tasks + { + debug!( + "Region ({:?}) compaction task size larger than max background tasks({}), remaining tasks discarded", + region_id, max_background_tasks + ); + break 'phases; + } } } } + // The compactor pops outputs from the end, preserving phase priority and + // newer-window-first pop order within each phase. + output.reverse(); Ok(output) } fn find_inputs( &self, region_id: RegionId, - active_window: Option, - files: &Window, - windows: &BTreeMap, + context: WindowPickContext<'_>, ) -> (Vec, bool) { - if files.files.len() < self.trigger_file_num { - return (vec![], false); - } - + let WindowPickContext { + active_window, + files, + windows, + phase, + } = context; + let is_active_window = active_window == Some(files.time_window); let window = &files.time_window; let mut files_to_merge: Vec<_> = files.files().cloned().collect(); @@ -184,33 +250,45 @@ impl TwcsPicker { } } - let (mut l0_files, l1_files): (Vec<_>, Vec<_>) = files_to_merge + let (l0_files, l1_files): (Vec<_>, Vec<_>) = files_to_merge .into_iter() .partition(|file| file.level() == 0); let num_l0_files = l0_files.len(); let num_l1_files = l1_files.len(); - // Keep fresh L0 data and compacted L1 data in separate tasks whenever either - // level can trigger compaction on its own. This prevents each L0 batch from - // pulling the previous L1 output into another rewrite. - let (inputs, found_runs) = if num_l0_files >= self.trigger_file_num { - let l0_pick = - pick_candidate_files(l0_files, self.max_output_file_size, pick_count_first); - if l0_pick.0.is_empty() && num_l1_files >= self.trigger_file_num { - pick_candidate_files(l1_files, self.max_output_file_size, pick_count_first) - } else { - l0_pick + if !is_active_window + && files.files.len() < self.inactive_window_trigger_file_num + && num_l1_files < self.inactive_window_l1_merge_trigger + { + return (vec![], false); + } + let (inputs, found_runs) = if is_active_window { + match phase { + PickPhase::HasL0 if num_l0_files >= self.trigger_file_num => { + pick_candidate_files(l0_files, self.max_output_file_size, pick_count_first) + } + PickPhase::L1FileReduction | PickPhase::L1OverlapOnly + if num_l1_files >= self.active_window_l1_merge_trigger => + { + pick_l1_candidate_files(l1_files, self.max_output_file_size, phase) + } + _ => (vec![], 0), } - } else if num_l1_files >= self.trigger_file_num { - pick_candidate_files(l1_files, self.max_output_file_size, pick_count_first) } else { - l0_files.extend(l1_files); - let picker = if num_l0_files > 0 && num_l1_files > 0 { - pick_mixed_count_first - } else { - pick_count_first - }; - pick_candidate_files(l0_files, self.max_output_file_size, picker) + pick_inactive_window_files( + l0_files, + l1_files, + InactiveWindowPick { + l0_file_num: self.inactive_window_trigger_file_num, + l1_file_num: self.inactive_window_l1_merge_trigger, + phase, + }, + self.max_output_file_size, + ) }; + if inputs.is_empty() { + return (inputs, false); + } + let filter_deleted = !self.append_mode && !window_has_overlap(files, windows) && !selected_overlaps_unselected(&inputs, files); @@ -232,6 +310,74 @@ impl TwcsPicker { } } +/// Picks compaction inputs for an inactive window. +/// +/// The window no longer receives fresh writes (late arrivals aside), so it +/// should converge, but merging must stay within a bounded rewrite cost: +/// +/// 1. L0 candidates are selected before pure L1 candidates across all windows. +/// 2. Balanced L0 picks below the trigger continue converging without pulling +/// L1 files into the rewrite. +/// 3. A mixed merge may bypass the balance checks, but only when the total +/// rewrite fits in the output file budget, bounding write amplification. +/// 4. Last resort: converge L0 files among themselves regardless of balance. +/// The rewrite is bounded by the L0 bytes and leaves large compacted files +/// untouched. If nothing qualifies, the window is left as-is. +#[derive(Debug, Clone, Copy)] +struct InactiveWindowPick { + l0_file_num: usize, + l1_file_num: usize, + phase: PickPhase, +} + +fn pick_inactive_window_files( + l0_files: Vec, + l1_files: Vec, + pick: InactiveWindowPick, + max_output_file_size: Option, +) -> (Vec, usize) { + if !matches!(pick.phase, PickPhase::HasL0) { + return if l1_files.len() >= pick.l1_file_num { + pick_l1_candidate_files(l1_files, max_output_file_size, pick.phase) + } else { + (vec![], 0) + }; + } + + if l0_files.len() >= pick.l0_file_num { + let pick = pick_candidate_files(l0_files.clone(), max_output_file_size, pick_count_first); + if !pick.0.is_empty() { + return pick; + } + } + + let pick = pick_candidate_files(l0_files.clone(), max_output_file_size, pick_count_first); + if !pick.0.is_empty() { + return pick; + } + let mut all_files = l0_files.clone(); + all_files.extend(l1_files); + let pick = pick_candidate_files(all_files, max_output_file_size, pick_mixed_within_budget); + if !pick.0.is_empty() { + return pick; + } + + pick_candidate_files(l0_files, max_output_file_size, pick_unbalanced_count_first) +} + +fn pick_l1_candidate_files( + l1_files: Vec, + max_output_file_size: Option, + phase: PickPhase, +) -> (Vec, usize) { + let picker = match phase { + PickPhase::L1FileReduction => pick_l1_file_reduction, + PickPhase::L1OverlapOnly => pick_l1_overlap_only, + PickPhase::HasL0 => return (vec![], 0), + }; + pick_candidate_files(l1_files, max_output_file_size, picker) +} + fn pick_candidate_files( mut files: Vec, max_output_file_size: Option, @@ -358,6 +504,17 @@ impl Candidate { self.has_l0 && self.has_l1 } + /// Whether rewriting this candidate stays within the output file budget. + /// Inactive-window merges use this to bound write amplification when they + /// bypass the balance checks. An unset budget means no bound: the operator + /// removed the output size limit explicitly. + fn within_rewrite_budget(&self, max_output_file_size: Option) -> bool { + match max_output_file_size { + Some(limit) if limit > 0 => self.total_size <= limit as usize, + _ => true, + } + } + /// A candidate is worth compacting only if it makes progress on at least one /// axis: it reduces the physical file count, or it resolves at least one /// overlap (merging sorted runs and reducing read amplification). A pure @@ -402,8 +559,9 @@ impl CandidateScore { /// An interval is eligible when it /// /// - holds at least 2 files, -/// - is balanced: no single file dominates it (`largest <= sum of the others`), -/// - when mixing levels with known row counts, has at most twice as many L1 rows as L0 rows, +/// - passes the caller-supplied eligibility predicate (e.g. the byte- and +/// row-balance checks for regular picks, the rewrite budget for inactive +/// window fallbacks), /// - makes progress on at least one axis: it reduces the physical file count given /// the output split threshold `max_output_file_size`, or it resolves at least one /// overlap between sorted runs. @@ -413,18 +571,62 @@ fn pick_count_first( sorted_runs: Vec>, max_output_file_size: Option, ) -> Vec { - pick_count_first_where(sorted_runs, max_output_file_size, |_| true) + pick_count_first_where(sorted_runs, max_output_file_size, is_balanced_candidate) } +fn pick_l1_file_reduction( + sorted_runs: Vec>, + max_output_file_size: Option, +) -> Vec { + pick_count_first_where(sorted_runs, max_output_file_size, |candidate| { + is_balanced_candidate(candidate) && candidate.file_reduction(max_output_file_size) > 0 + }) +} + +fn pick_l1_overlap_only( + sorted_runs: Vec>, + max_output_file_size: Option, +) -> Vec { + pick_count_first_where(sorted_runs, max_output_file_size, |candidate| { + is_balanced_candidate(candidate) + && candidate.file_reduction(max_output_file_size) == 0 + && candidate.overlap_participants > 0 + }) +} + +#[cfg(test)] fn pick_mixed_count_first( sorted_runs: Vec>, max_output_file_size: Option, ) -> Vec { - pick_count_first_where( - sorted_runs, - max_output_file_size, - Candidate::has_mixed_levels, - ) + pick_count_first_where(sorted_runs, max_output_file_size, |candidate| { + candidate.has_mixed_levels() && is_balanced_candidate(candidate) + }) +} + +/// Mixed-level fallback for inactive windows: bypasses the balance checks but +/// bounds the total rewrite to the output file budget. +fn pick_mixed_within_budget( + sorted_runs: Vec>, + max_output_file_size: Option, +) -> Vec { + pick_count_first_where(sorted_runs, max_output_file_size, |candidate| { + candidate.has_mixed_levels() && candidate.within_rewrite_budget(max_output_file_size) + }) +} + +/// Last-resort pick for inactive windows with no balance requirement at all. +/// Callers must only pass single-level (L0) files so the rewrite stays bounded +/// by their bytes. +fn pick_unbalanced_count_first( + sorted_runs: Vec>, + max_output_file_size: Option, +) -> Vec { + pick_count_first_where(sorted_runs, max_output_file_size, |_| true) +} + +fn is_balanced_candidate(candidate: &Candidate) -> bool { + candidate.is_balanced() && candidate.has_balanced_level_rows() } fn pick_count_first_where( @@ -442,9 +644,7 @@ fn pick_count_first_where( for right in left..right_bound { candidate.absorb(&files[right], &files[left..right], &mut participations); if candidate.num_files < 2 - || !candidate.is_balanced() || !is_eligible(&candidate) - || !candidate.has_balanced_level_rows() || !candidate.makes_progress(max_output_file_size) { continue; @@ -606,8 +806,6 @@ impl Picker for TwcsPicker { inferred }); - let active_window = - find_latest_window_in_seconds(levels[0].files(), time_window_size); let windows = assign_to_windows( levels .iter() @@ -615,6 +813,21 @@ impl Picker for TwcsPicker { .filter(|file| !expired_file_ids.contains(&file.file_id())), time_window_size, ); + // Compute activity from the candidate files so expired or + // compacting files cannot identify a window absent from `windows`. + let active_window = find_active_window_by_sequence( + windows.values().flat_map(Window::files), + time_window_size, + ) + .or_else(|| { + find_latest_window_in_seconds( + windows + .values() + .flat_map(Window::files) + .filter(|file| file.level() == 0), + time_window_size, + ) + }); (expired_ssts, time_window_size, active_window, windows) }) @@ -629,7 +842,11 @@ impl Picker for TwcsPicker { return Ok(None); } - let max_file_size = self.max_output_file_size.map(|v| v as usize); + // The picker treats zero as unlimited, but the SST writer would split every batch. + let max_file_size = self + .max_output_file_size + .filter(|size| *size > 0) + .map(|size| size as usize); Ok(Some(PickerOutput { outputs, expired_ssts, @@ -784,6 +1001,31 @@ fn find_latest_window_in_seconds<'a>( .and_then(|ts| ts.value().align_to_ceil_by_bucket(time_window_size)) } +/// Finds the active window from the file with the highest sequence number. +/// +/// Flush and compaction outputs inherit the max input sequence, so the +/// max-sequence file always tracks the most recent write: the active window +/// survives the transient state where level 0 is empty right after its files +/// were compacted away. Returns `None` when no file carries a sequence. +/// +/// The window key follows the same convention as [`assign_to_windows`] +/// (truncate to seconds, then align up), since it is compared against the +/// window keys produced there. +fn find_active_window_by_sequence<'a>( + files: impl Iterator, + time_window_size: i64, +) -> Option { + files + .filter(|f| f.meta_ref().sequence.is_some()) + .max_by_key(|f| f.meta_ref().sequence) + .and_then(|f| { + f.time_range() + .1 + .convert_to(TimeUnit::Second) + .and_then(|ts| ts.value().align_to_ceil_by_bucket(time_window_size)) + }) +} + #[cfg(test)] mod tests { use std::collections::HashSet; @@ -818,27 +1060,21 @@ mod tests { #[test] fn test_invalid_max_input_files_env_falls_back_to_default() { for env_value in [None, Some(""), Some("invalid"), Some("0"), Some("1")] { - assert_eq!(32, parse_max_input_files(env_value)); + assert_eq!(16, parse_max_input_files(env_value)); } } - async fn compaction_region_with_expired_sst() -> CompactionRegion { + async fn compaction_region_with_ssts( + files: impl IntoIterator, + ttl: Duration, + ) -> CompactionRegion { let env = SchedulerEnv::new().await; let metadata = metadata_for_test(); let manifest_ctx = env.mock_manifest_context(metadata.clone()).await; let mut ssts = SstVersion::new(); ssts.add_files( Arc::new(crate::sst::file_purger::NoopFilePurger), - (1..=4).map(|sequence| FileMeta { - file_id: FileId::random(), - time_range: ( - Timestamp::new_millisecond(0), - Timestamp::new_millisecond(10), - ), - level: 0, - sequence: NonZeroU64::new(sequence), - ..Default::default() - }), + files.into_iter(), ); CompactionRegion { @@ -856,16 +1092,69 @@ mod tests { compaction_time_window: None, }, file_purger: None, - ttl: Some(Duration::from_millis(1).into()), + ttl: Some(ttl.into()), max_parallelism: 1, plugins: Plugins::new(), } } + async fn compaction_region_with_expired_sst() -> CompactionRegion { + compaction_region_with_ssts( + (1..=4).map(|sequence| FileMeta { + file_id: FileId::random(), + time_range: ( + Timestamp::new_millisecond(0), + Timestamp::new_millisecond(10), + ), + level: 0, + sequence: NonZeroU64::new(sequence), + ..Default::default() + }), + Duration::from_millis(1), + ) + .await + } + + #[tokio::test] + async fn test_pick_normalizes_zero_output_size_to_unlimited() { + let mut compaction_region = compaction_region_with_ssts( + (1..=4).map(|sequence| new_file(0, 10, sequence, 100).meta_ref().clone()), + Duration::from_secs(3600), + ) + .await; + compaction_region.ttl = None; + + for (max_output_file_size, expected_max_file_size) in + [(Some(0), None), (None, None), (Some(1024), Some(1024))] + { + let picker = TwcsPicker { + trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 4, + inactive_window_l1_merge_trigger: 8, + time_window_seconds: Some(3), + max_output_file_size, + append_mode: false, + max_background_tasks: None, + time_range: None, + }; + + let output = picker.pick(&compaction_region).await.unwrap().unwrap(); + + assert_eq!(output.outputs.len(), 1); + assert_eq!(output.outputs[0].inputs.len(), 4); + assert!(output.expired_ssts.is_empty()); + assert_eq!(output.max_file_size, expected_max_file_size); + } + } + #[tokio::test] async fn test_pick_expired_ssts_without_marking_compacting() { let picker = TwcsPicker { trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 4, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, @@ -881,6 +1170,176 @@ mod tests { assert!(output.expired_ssts.iter().all(|file| !file.compacting())); } + #[tokio::test] + async fn test_expired_sst_does_not_determine_active_window() { + let now = Timestamp::current_millis().value(); + let files = [ + FileMeta { + file_id: FileId::random(), + time_range: ( + Timestamp::new_millisecond(0), + Timestamp::new_millisecond(10), + ), + level: 0, + sequence: NonZeroU64::new(100), + ..Default::default() + }, + FileMeta { + file_id: FileId::random(), + time_range: ( + Timestamp::new_millisecond(now - 1000), + Timestamp::new_millisecond(now), + ), + level: 0, + sequence: NonZeroU64::new(1), + ..Default::default() + }, + FileMeta { + file_id: FileId::random(), + time_range: ( + Timestamp::new_millisecond(now - 1000), + Timestamp::new_millisecond(now), + ), + level: 0, + sequence: NonZeroU64::new(2), + ..Default::default() + }, + ]; + let compaction_region = compaction_region_with_ssts(files, Duration::from_secs(60)).await; + let picker = TwcsPicker { + trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, + time_window_seconds: Some(3), + max_output_file_size: None, + append_mode: false, + max_background_tasks: None, + time_range: None, + }; + + let output = picker.pick(&compaction_region).await.unwrap().unwrap(); + + assert_eq!(1, output.expired_ssts.len()); + assert!(output.outputs.is_empty()); + } + + #[tokio::test] + async fn test_active_window_survives_l0_consumption() { + // Simulates the transient state right after an L0 compaction: level 0 is + // empty and only L1 files remain. The active window must still be the + // window of the most recent write (tracked by the max-sequence file), so + // the window is NOT compacted under the inactive rules. + let env = SchedulerEnv::new().await; + let metadata = metadata_for_test(); + let manifest_ctx = env.mock_manifest_context(metadata.clone()).await; + let mut ssts = SstVersion::new(); + ssts.add_files( + Arc::new(crate::sst::file_purger::NoopFilePurger), + [100, 101].into_iter().map(|sequence| FileMeta { + file_id: FileId::random(), + time_range: ( + Timestamp::new_millisecond(0), + Timestamp::new_millisecond(10), + ), + level: 1, + sequence: NonZeroU64::new(sequence), + ..Default::default() + }), + ); + let compaction_region = CompactionRegion { + region_id: metadata.region_id, + region_options: RegionOptions::default(), + engine_config: Arc::new(MitoConfig::default()), + region_metadata: metadata.clone(), + cache_manager: Arc::new(CacheManager::default()), + access_layer: env.access_layer, + manifest_ctx, + current_version: CompactionVersion { + metadata, + options: RegionOptions::default(), + ssts: Arc::new(ssts), + compaction_time_window: None, + }, + file_purger: None, + ttl: None, + max_parallelism: 1, + plugins: Plugins::new(), + }; + let picker = TwcsPicker { + trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, + time_window_seconds: Some(3600), + max_output_file_size: None, + append_mode: false, + max_background_tasks: None, + time_range: None, + }; + + assert!(picker.pick(&compaction_region).await.unwrap().is_none()); + } + + #[test] + fn test_find_active_window_by_sequence() { + // The max-sequence file tracks the most recent write regardless of + // level: here a compacted L1 file ending at 2999ms (window key 2) is + // newer than the L0 file ending at 999ms (window key 0). Window keys + // follow the assign_to_windows convention (truncate to seconds). + let files = [ + new_file_handle_with_sequence(FileId::random(), 0, 999, 0, 1), + new_file_handle_with_sequence(FileId::random(), 2000, 2999, 1, 10), + ]; + assert_eq!(Some(2), find_active_window_by_sequence(files.iter(), 1)); + + // Backfill: the newest write lands in an older window, and that window + // is the active one. + let files = [ + new_file_handle_with_sequence(FileId::random(), 0, 999, 0, 10), + new_file_handle_with_sequence(FileId::random(), 2000, 2999, 0, 1), + ]; + assert_eq!(Some(0), find_active_window_by_sequence(files.iter(), 1)); + + // Files without a sequence are skipped; if none have one, the caller + // falls back to the L0-based rule. + let files = [new_file_handle_with_sequence( + FileId::random(), + 0, + 999, + 1, + 0, + )]; + assert_eq!(None, find_active_window_by_sequence(files.iter(), 1)); + assert!(find_active_window_by_sequence(Vec::::new().iter(), 1).is_none()); + } + + #[test] + fn test_active_window_falls_back_to_l0_rule_without_sequence() { + let active_window_of = |files: &[FileHandle]| { + find_active_window_by_sequence(files.iter(), 1).or_else(|| { + find_latest_window_in_seconds(files.iter().filter(|f| f.level() == 0), 1) + }) + }; + + // Legacy files without sequence: the L0-based rule still applies. + let files = [ + new_file_handle_with_sequence(FileId::random(), 0, 999, 1, 0), + new_file_handle_with_sequence(FileId::random(), 2000, 2999, 0, 0), + ]; + assert_eq!(Some(3), active_window_of(&files)); + + // No sequence and no L0: None, exactly as before. + let files = [new_file_handle_with_sequence( + FileId::random(), + 0, + 999, + 1, + 0, + )]; + assert_eq!(None, active_window_of(&files)); + } + #[test] fn test_get_latest_window_in_seconds() { assert_eq!( @@ -1261,9 +1720,14 @@ mod tests { .collect::>(); 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); + find_active_window_by_sequence(self.input_files.iter(), self.window_size).or_else( + || find_latest_window_in_seconds(self.input_files.iter(), self.window_size), + ); let output = TwcsPicker { trigger_file_num: 2, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 2, time_window_seconds: None, max_output_file_size: None, append_mode: false, @@ -1304,7 +1768,7 @@ mod tests { } #[tokio::test] - async fn test_build_twcs_output() { + async fn test_newer_windows_are_placed_last_for_execution() { let file_ids = (0..4).map(|_| FileId::random()).collect::>(); // Case 1: 2 runs found in each time window. @@ -1319,11 +1783,11 @@ mod tests { .to_vec(), expected_outputs: vec![ ExpectedOutput { - input_files: vec![2, 3], + input_files: vec![0, 1], output_level: 1, }, ExpectedOutput { - input_files: vec![0, 1], + input_files: vec![2, 3], output_level: 1, }, ], @@ -1350,11 +1814,11 @@ mod tests { .to_vec(), expected_outputs: vec![ ExpectedOutput { - input_files: vec![2, 3, 4], + input_files: vec![0, 1], output_level: 1, }, ExpectedOutput { - input_files: vec![0, 1], + input_files: vec![2, 3, 4], output_level: 1, }, ], @@ -1377,18 +1841,10 @@ mod tests { new_file_handle_with_sequence(file_ids[4], 11, 2990, 0, 3), ] .to_vec(), - expected_outputs: vec![ - ExpectedOutput { - input_files: vec![2, 3], - output_level: 1, - }, - ExpectedOutput { - // L1 reaches the trigger on its own, so compact it without - // rewriting the L0 tail. A chained pick can handle the tail. - input_files: vec![0, 1], - output_level: 1, - }, - ], + expected_outputs: vec![ExpectedOutput { + input_files: vec![2, 3], + output_level: 1, + }], } .check() .await; @@ -1420,6 +1876,9 @@ mod tests { let active_window = find_latest_window_in_seconds(files.iter(), 3); let output = TwcsPicker { trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 4, + inactive_window_l1_merge_trigger: 8, time_window_seconds: None, max_output_file_size: None, append_mode: false, @@ -1488,6 +1947,9 @@ mod tests { // Create picker with trigger_file_num of 4 so single files won't form runs in first window let picker = TwcsPicker { trigger_file_num: 4, // High enough to prevent runs in first window + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 4, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, @@ -1533,6 +1995,9 @@ mod tests { let picker = TwcsPicker { trigger_file_num: 2, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(3), max_output_file_size: Some(1000), append_mode: true, @@ -1564,6 +2029,9 @@ mod tests { let windows = assign_to_windows(files.iter(), 1); let picker = TwcsPicker { trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 4, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(1), max_output_file_size: Some(1_000), append_mode: true, @@ -1606,6 +2074,9 @@ mod tests { let picker = TwcsPicker { trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 4, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, @@ -1630,6 +2101,9 @@ mod tests { // Without max_background_tasks, should have more outputs let picker_no_limit = TwcsPicker { trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 4, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, @@ -1674,6 +2148,9 @@ mod tests { let picker = TwcsPicker { trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 4, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, @@ -1723,6 +2200,9 @@ mod tests { let picker = TwcsPicker { trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 4, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, @@ -1767,6 +2247,9 @@ mod tests { let picker = TwcsPicker { trigger_file_num: num_files, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: num_files, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, @@ -1788,8 +2271,8 @@ mod tests { async fn test_limit_max_input_files_keeps_deletion_markers() { common_telemetry::init_default_ut_logging(); - // One large file group spanning the whole window plus 32 small ones nested inside - // it. That is exactly 2 runs, so the window on its own allows filtering deletions. + // One large file group spanning the whole window plus enough small ones to fill the + // input limit. That is exactly 2 runs, so the window allows filtering deletions. let mut files = vec![new_file_handle_with_size_and_sequence( FileId::random(), 0, @@ -1798,7 +2281,7 @@ mod tests { 1, 1024 * 1024 * 1024, )]; - files.extend((0..32).map(|idx: i64| { + files.extend((0..DEFAULT_MAX_INPUT_FILES as i64).map(|idx| { new_file_handle_with_size_and_sequence( FileId::random(), (idx + 1) * 10_000, @@ -1813,6 +2296,9 @@ mod tests { let picker = TwcsPicker { trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 4, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(3600), max_output_file_size: None, append_mode: false, @@ -1829,7 +2315,7 @@ mod tests { assert_eq!(1, output.len()); // The input file num limit picks the smallest groups first, so the large group is // left behind while the small groups that overlap it are compacted. - assert_eq!(32, output[0].inputs.len()); + assert_eq!(DEFAULT_MAX_INPUT_FILES, output[0].inputs.len()); assert!( !output[0].filter_deleted, "deletion markers must be kept once the file num limit drops files they may mask" @@ -1840,9 +2326,9 @@ mod tests { async fn test_limit_max_input_files_still_filters_without_overlap() { common_telemetry::init_default_ut_logging(); - // 40 file groups with disjoint time ranges, i.e. a single run. Nothing the file num - // limit leaves behind can hold a row masked by a deletion marker we compact. - let files: Vec<_> = (0..40i64) + // More disjoint file groups than the input limit, i.e. a single run. Nothing left + // behind can hold a row masked by a deletion marker we compact. + let files: Vec<_> = (0..DEFAULT_MAX_INPUT_FILES as i64 + 8) .map(|idx| { new_file_handle_with_size_and_sequence( FileId::random(), @@ -1859,6 +2345,9 @@ mod tests { let picker = TwcsPicker { trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 4, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(3600), max_output_file_size: Some(1024 * 1024 * 1024), append_mode: false, @@ -1873,7 +2362,7 @@ mod tests { .unwrap(); assert_eq!(1, output.len()); - assert_eq!(32, output[0].inputs.len()); + assert_eq!(DEFAULT_MAX_INPUT_FILES, output[0].inputs.len()); assert!(output[0].filter_deleted); } @@ -1890,6 +2379,9 @@ mod tests { let windows = assign_to_windows(files.iter(), 3); let picker = TwcsPicker { trigger_file_num: 2, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, @@ -1951,6 +2443,9 @@ mod tests { let windows = assign_to_windows(files.iter(), 3); let picker = TwcsPicker { trigger_file_num: 2, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(3), max_output_file_size: None, append_mode: false, @@ -1977,6 +2472,132 @@ mod tests { ); } + #[tokio::test] + async fn test_inactive_window_uses_its_trigger_file_num() { + let files = [ + new_file_handle_with_size_and_sequence(FileId::random(), 0, 10, 0, 1, 10), + new_file_handle_with_size_and_sequence(FileId::random(), 20, 30, 0, 2, 10), + ]; + let windows = assign_to_windows(files.iter(), 100); + let picker = TwcsPicker { + trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, + time_window_seconds: Some(100), + max_output_file_size: None, + append_mode: false, + max_background_tasks: None, + time_range: None, + }; + + let output = picker + .build_output_with_time_range(RegionId::from_u64(1), windows, Some(100), None) + .await + .unwrap(); + + assert_eq!(1, output.len()); + assert_eq!(2, output[0].inputs.len()); + } + + #[test] + fn test_inactive_window_l1_trigger_is_independent_of_l0_trigger() { + let l0_files = (0..6) + .map(|idx| { + new_file_handle_with_size_and_sequence( + FileId::random(), + idx * 20, + idx * 20 + 10, + 0, + idx as u64 + 1, + 10, + ) + }) + .collect(); + let l1_files = (0..2) + .map(|idx| { + new_file_handle_with_size_and_sequence( + FileId::random(), + idx * 20, + idx * 20 + 10, + 1, + idx as u64 + 10, + 10, + ) + }) + .collect(); + + let (inputs, _) = pick_inactive_window_files( + l0_files, + l1_files, + InactiveWindowPick { + l0_file_num: 8, + l1_file_num: 2, + phase: PickPhase::L1FileReduction, + }, + None, + ); + + assert_eq!(2, inputs.len()); + assert!(inputs.iter().all(|file| file.level() == 1)); + } + + #[test] + fn test_inactive_window_does_not_fallback_to_l1_below_trigger() { + let l1_files = (0..2) + .map(|idx| { + new_file_handle_with_size_and_sequence( + FileId::random(), + idx * 20, + idx * 20 + 10, + 1, + idx as u64 + 1, + 10, + ) + }) + .collect(); + + let (inputs, _) = pick_inactive_window_files( + vec![], + l1_files, + InactiveWindowPick { + l0_file_num: 2, + l1_file_num: 8, + phase: PickPhase::L1FileReduction, + }, + None, + ); + + assert!(inputs.is_empty()); + } + + #[tokio::test] + async fn test_inactive_window_requires_a_level_trigger_before_mixed_fallback() { + let files = [ + new_file_handle_with_size_and_sequence(FileId::random(), 0, 10, 0, 1, 10), + new_file_handle_with_size_and_sequence(FileId::random(), 0, 10, 1, 2, 10), + ]; + let windows = assign_to_windows(files.iter(), 100); + let picker = TwcsPicker { + trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 8, + inactive_window_l1_merge_trigger: 2, + time_window_seconds: Some(100), + max_output_file_size: Some(1_000), + append_mode: false, + max_background_tasks: None, + time_range: None, + }; + + let output = picker + .build_output_with_time_range(RegionId::from_u64(1), windows, Some(100), None) + .await + .unwrap(); + + assert!(output.is_empty()); + } + #[tokio::test] async fn test_count_first_prefers_more_files_over_smaller_overlap() { let files = [ @@ -1988,6 +2609,9 @@ mod tests { let windows = assign_to_windows(files.iter(), 100); let picker = TwcsPicker { trigger_file_num: 2, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(100), max_output_file_size: None, append_mode: false, @@ -2014,6 +2638,9 @@ mod tests { let windows = assign_to_windows(files.iter(), 100); let picker = TwcsPicker { trigger_file_num: 3, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 3, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(100), max_output_file_size: None, append_mode: false, @@ -2039,6 +2666,9 @@ mod tests { let windows = assign_to_windows(files.iter(), 100); let picker = TwcsPicker { trigger_file_num: 3, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 3, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(100), max_output_file_size: None, append_mode: false, @@ -2079,6 +2709,9 @@ mod tests { let windows = assign_to_windows(files.iter(), 1000); let picker = TwcsPicker { trigger_file_num: 2, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(1000), max_output_file_size: None, append_mode: false, @@ -2258,6 +2891,9 @@ mod tests { })); let picker = TwcsPicker { trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 4, + inactive_window_l1_merge_trigger: 4, time_window_seconds: Some(1), max_output_file_size: None, append_mode: false, @@ -2267,7 +2903,7 @@ mod tests { for (case, files, expected_level, expected_len) in [ ("L0 reaches trigger", enough_l0, 0, DEFAULT_MAX_INPUT_FILES), - ("L1 reaches trigger", enough_l1, 1, 4), + ("L0 fallback precedes triggered L1", enough_l1, 0, 3), ] { let windows = assign_to_windows(files.iter(), 1); let output = picker @@ -2287,6 +2923,82 @@ mod tests { } } + #[tokio::test] + async fn test_active_window_defers_l1_below_safety_trigger() { + let files = (0..7) + .map(|idx| { + let start = idx * 100; + new_file_handle_with_size_and_sequence( + FileId::random(), + start, + start + 99, + 1, + idx as u64 + 1, + 100, + ) + }) + .collect::>(); + let windows = assign_to_windows(files.iter(), 1); + let active_window = windows.keys().next().copied(); + let picker = TwcsPicker { + trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, + time_window_seconds: Some(1), + max_output_file_size: None, + append_mode: false, + max_background_tasks: None, + time_range: None, + }; + + let output = picker + .build_output_with_time_range(RegionId::from_u64(1), windows, active_window, None) + .await + .unwrap(); + + assert!(output.is_empty()); + } + + #[tokio::test] + async fn test_active_window_l1_safety_trigger_is_independent_of_l0_trigger() { + let files = (0..8) + .map(|idx| { + let start = idx * 100; + new_file_handle_with_size_and_sequence( + FileId::random(), + start, + start + 99, + 1, + idx as u64 + 1, + 100, + ) + }) + .collect::>(); + let windows = assign_to_windows(files.iter(), 1); + let active_window = windows.keys().next().copied(); + let picker = TwcsPicker { + trigger_file_num: 16, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, + time_window_seconds: Some(1), + max_output_file_size: None, + append_mode: false, + max_background_tasks: None, + time_range: None, + }; + + let output = picker + .build_output_with_time_range(RegionId::from_u64(1), windows, active_window, None) + .await + .unwrap(); + + assert_eq!(1, output.len()); + assert_eq!(8, output[0].inputs.len()); + assert!(output[0].inputs.iter().all(|file| file.level() == 1)); + } + #[tokio::test] async fn test_picker_falls_back_to_l1_when_triggered_l0_cannot_make_progress() { let mut files = (0..4) @@ -2302,7 +3014,7 @@ mod tests { ) }) .collect::>(); - files.extend((0..4).map(|idx| { + files.extend((0..8).map(|idx| { let start = idx * 20 + 100; new_file_handle_with_size_and_sequence( FileId::random(), @@ -2310,12 +3022,16 @@ mod tests { start + 9, 1, idx as u64 + 10, - 100, + 50, ) })); let windows = assign_to_windows(files.iter(), 1); + let active_window = windows.keys().next().copied(); let picker = TwcsPicker { trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 4, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(1), max_output_file_size: Some(512), append_mode: false, @@ -2324,13 +3040,244 @@ mod tests { }; let output = picker + .build_output_with_time_range(RegionId::from_u64(1), windows, active_window, None) + .await + .unwrap(); + + assert_eq!(1, output.len()); + assert_eq!(8, output[0].inputs.len()); + assert!(output[0].inputs.iter().all(|file| file.level() == 1)); + } + + fn priority_test_files( + window: i64, + level: Level, + file_size: u64, + overlap: bool, + ) -> [FileHandle; 2] { + let start = window * 1_000; + let ranges = if overlap { + [(start, start + 499), (start + 250, start + 749)] + } else { + [(start, start + 99), (start + 200, start + 299)] + }; + ranges.map(|(start, end)| { + new_file_handle_with_size_and_sequence( + FileId::random(), + start, + end, + level, + window as u64 + 1, + file_size, + ) + }) + } + + fn priority_test_picker(max_background_tasks: usize) -> TwcsPicker { + TwcsPicker { + trigger_file_num: 2, + active_window_l1_merge_trigger: 2, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 2, + time_window_seconds: Some(1), + max_output_file_size: Some(100), + append_mode: false, + max_background_tasks: Some(max_background_tasks), + time_range: None, + } + } + + fn output_window(output: &CompactionOutput) -> i64 { + output.inputs[0] + .time_range() + .1 + .convert_to(TimeUnit::Second) + .unwrap() + .value() + } + + #[tokio::test] + async fn test_older_l0_candidate_has_priority_over_newer_pure_l1_candidate() { + let files = priority_test_files(0, 0, 40, false) + .into_iter() + .chain(priority_test_files(1, 1, 40, false)) + .collect::>(); + let windows = assign_to_windows(files.iter(), 1); + + let output = priority_test_picker(1) .build_output_with_time_range(RegionId::from_u64(1), windows, Some(1), None) .await .unwrap(); assert_eq!(1, output.len()); - assert_eq!(4, output[0].inputs.len()); - assert!(output[0].inputs.iter().all(|file| file.level() == 1)); + assert_eq!(0, output_window(&output[0])); + assert!(output[0].inputs.iter().all(|file| file.level() == 0)); + } + + #[tokio::test] + async fn test_older_l1_file_reduction_has_priority_over_newer_overlap_only_l1() { + let files = priority_test_files(0, 1, 40, false) + .into_iter() + .chain(priority_test_files(1, 1, 100, true)) + .collect::>(); + let windows = assign_to_windows(files.iter(), 1); + + let output = priority_test_picker(1) + .build_output_with_time_range(RegionId::from_u64(1), windows, Some(1), None) + .await + .unwrap(); + + assert_eq!(1, output.len()); + assert_eq!(0, output_window(&output[0])); + } + + #[tokio::test] + async fn test_multiple_slots_follow_class_priority_and_newer_first_within_class() { + let files = [ + priority_test_files(0, 0, 40, false), + priority_test_files(1, 0, 40, false), + priority_test_files(2, 1, 40, false), + priority_test_files(3, 1, 40, false), + priority_test_files(4, 1, 100, true), + priority_test_files(5, 1, 100, true), + ] + .into_iter() + .flatten() + .collect::>(); + let windows = assign_to_windows(files.iter(), 1); + + let output = priority_test_picker(5) + .build_output_with_time_range(RegionId::from_u64(1), windows, Some(5), None) + .await + .unwrap(); + let pop_priority = output + .iter() + .rev() + .map(|output| (output.inputs[0].level(), output_window(output))) + .collect::>(); + + assert_eq!(vec![(0, 1), (0, 0), (1, 3), (1, 2), (1, 5)], pop_priority); + } + + #[tokio::test] + async fn test_inactive_window_mixed_fallback_merges_within_rewrite_budget() { + let files = [ + new_file_with_level_and_rows(0, 99, 1, 1, 1_000, 1_000_000), + new_file_with_level_and_rows(0, 9, 0, 2, 10, 1), + ]; + let windows = assign_to_windows(files.iter(), 100); + let picker = TwcsPicker { + trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, + time_window_seconds: Some(100), + max_output_file_size: Some(2_000), + append_mode: false, + max_background_tasks: None, + time_range: None, + }; + + let output = picker + .build_output_with_time_range(RegionId::from_u64(1), windows, Some(100), None) + .await + .unwrap(); + + assert_eq!(1, output.len()); + assert_eq!(2, output[0].inputs.len()); + } + + #[tokio::test] + async fn test_inactive_window_over_budget_merges_l0_only() { + // L0 files are unbalanced among themselves, and merging in the huge L1 file + // would exceed the rewrite budget: only the L0 files are compacted. + let files = [ + new_file_handle_with_size_and_sequence(FileId::random(), 0, 9, 0, 1, 10_000), + new_file_handle_with_size_and_sequence(FileId::random(), 20, 29, 0, 2, 10), + new_file_handle_with_size_and_sequence(FileId::random(), 0, 99, 1, 3, 1_000_000), + ]; + let windows = assign_to_windows(files.iter(), 100); + let picker = TwcsPicker { + trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, + time_window_seconds: Some(100), + max_output_file_size: Some(100_000), + append_mode: false, + max_background_tasks: None, + time_range: None, + }; + + let output = picker + .build_output_with_time_range(RegionId::from_u64(1), windows, Some(100), None) + .await + .unwrap(); + + assert_eq!(1, output.len()); + assert_eq!(2, output[0].inputs.len()); + assert!(output[0].inputs.iter().all(|file| file.level() == 0)); + } + + #[tokio::test] + async fn test_inactive_window_over_budget_without_l0_pair_does_not_merge() { + // A single tiny L0 and a huge L1: merging is over budget and there is no + // L0 pair to converge, so the window is left as-is. + let files = [ + new_file_handle_with_size_and_sequence(FileId::random(), 0, 9, 0, 1, 10), + new_file_handle_with_size_and_sequence(FileId::random(), 0, 99, 1, 2, 1_000_000), + ]; + let windows = assign_to_windows(files.iter(), 100); + let picker = TwcsPicker { + trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, + time_window_seconds: Some(100), + max_output_file_size: Some(100_000), + append_mode: false, + max_background_tasks: None, + time_range: None, + }; + + let output = picker + .build_output_with_time_range(RegionId::from_u64(1), windows, Some(100), None) + .await + .unwrap(); + + assert!(output.is_empty()); + } + + #[tokio::test] + async fn test_inactive_window_falls_through_when_triggered_l0_pick_fails() { + // L0 reaches the inactive trigger but is unbalanced; L1 is below the + // trigger. The window must still converge through the budgeted mixed + // fallback instead of being skipped. + let files = [ + new_file_handle_with_size_and_sequence(FileId::random(), 0, 9, 0, 1, 100_000), + new_file_handle_with_size_and_sequence(FileId::random(), 20, 29, 0, 2, 10), + new_file_handle_with_size_and_sequence(FileId::random(), 0, 99, 1, 3, 50_000), + ]; + let windows = assign_to_windows(files.iter(), 100); + let picker = TwcsPicker { + trigger_file_num: 4, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, + time_window_seconds: Some(100), + max_output_file_size: Some(1_000_000), + append_mode: false, + max_background_tasks: None, + time_range: None, + }; + + let output = picker + .build_output_with_time_range(RegionId::from_u64(1), windows, Some(100), None) + .await + .unwrap(); + + assert_eq!(1, output.len()); + assert_eq!(3, output[0].inputs.len()); } #[test] @@ -2374,7 +3321,8 @@ mod tests { assert_eq!(DEFAULT_MAX_INPUT_FILES, ranges.len()); assert_eq!(Some(&(20, 29)), ranges.first()); - assert_eq!(Some(&(640, 649)), ranges.last()); + let last_start = DEFAULT_MAX_INPUT_FILES as i64 * 20; + assert_eq!(Some(&(last_start, last_start + 9)), ranges.last()); } #[test] @@ -2445,7 +3393,8 @@ mod tests { assert_eq!(DEFAULT_MAX_INPUT_FILES, ranges.len()); assert_eq!(Some(&(0, 9)), ranges.first()); - assert_eq!(Some(&(620, 629)), ranges.last()); + let last_start = (DEFAULT_MAX_INPUT_FILES as i64 - 1) * 20; + assert_eq!(Some(&(last_start, last_start + 9)), ranges.last()); } #[test] @@ -2514,6 +3463,9 @@ mod tests { let windows = assign_to_windows(files.iter(), 100); let picker = TwcsPicker { trigger_file_num: 3, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 3, + inactive_window_l1_merge_trigger: 8, time_window_seconds: Some(100), max_output_file_size: None, append_mode: false, diff --git a/src/mito2/src/engine/alter_test.rs b/src/mito2/src/engine/alter_test.rs index 6bd20100e8..736f32cc8a 100644 --- a/src/mito2/src/engine/alter_test.rs +++ b/src/mito2/src/engine/alter_test.rs @@ -1689,6 +1689,70 @@ async fn test_alter_region_ttl_options_with_format(flat_format: bool) { check_ttl(&engine, &Duration::from_secs(500)); } +#[tokio::test] +async fn test_alter_twcs_option_enables_compaction_override() { + let mut env = TestEnv::new().await; + let engine = env.create_engine(MitoConfig::default()).await; + let region_id = RegionId::new(1, 1); + + env.get_schema_metadata_manager() + .register_region_table_info( + region_id.table_id(), + "test_table", + "test_catalog", + "test_schema", + None, + env.get_kv_backend(), + ) + .await; + engine + .handle_request( + region_id, + RegionRequest::Create(CreateRequestBuilder::new().build()), + ) + .await + .unwrap(); + let compaction_override = || { + engine + .get_region(region_id) + .unwrap() + .version() + .options + .compaction_override + }; + assert!(!compaction_override()); + + engine + .handle_request( + region_id, + RegionRequest::Alter(RegionAlterRequest { + kind: AlterKind::UnsetRegionOptions { + keys: vec![UnsetRegionOption::TwcsTimeWindow], + }, + }), + ) + .await + .unwrap(); + assert!(!compaction_override()); + + engine + .handle_request( + region_id, + RegionRequest::Alter(RegionAlterRequest { + kind: AlterKind::SetRegionOptions { + options: vec![SetRegionOption::Twsc( + "compaction.twcs.time_window".to_string(), + "2h".to_string(), + )], + }, + }), + ) + .await + .unwrap(); + + assert!(compaction_override()); +} + #[tokio::test] async fn test_mixed_region_options_are_published_after_flush() { let mut env = TestEnv::new().await; @@ -1733,6 +1797,10 @@ async fn test_mixed_region_options_are_published_after_flush() { kind: AlterKind::SetRegionOptions { options: vec![ SetRegionOption::Ttl(Some(Duration::from_secs(500).into())), + SetRegionOption::Twsc( + "compaction.twcs.time_window".to_string(), + "2h".to_string(), + ), SetRegionOption::MaxRowGroupRowCount(Some(1024)), ], }, @@ -1745,6 +1813,7 @@ async fn test_mixed_region_options_are_published_after_flush() { listener.wait_flush_begin().await; let version = engine.get_region(region_id).unwrap().version(); assert_eq!(None, version.options.ttl); + assert!(!version.options.compaction_override); assert_eq!(None, version.options.max_row_group_row_count); listener.wake_flush(); @@ -1752,6 +1821,7 @@ async fn test_mixed_region_options_are_published_after_flush() { let version = engine.get_region(region_id).unwrap().version(); assert_eq!(Some(Duration::from_secs(500).into()), version.options.ttl); + assert!(version.options.compaction_override); assert_eq!(Some(1024), version.options.max_row_group_row_count); } diff --git a/src/mito2/src/engine/compaction_test.rs b/src/mito2/src/engine/compaction_test.rs index 04c97bdfbb..c67afbf8c6 100644 --- a/src/mito2/src/engine/compaction_test.rs +++ b/src/mito2/src/engine/compaction_test.rs @@ -1392,7 +1392,7 @@ async fn env_for_manual_compaction( (engine, column_schemas) } -/// The picker caps a compaction at 32 input files and drops the largest file groups to get +/// The picker caps a compaction at 16 input files and drops the largest file groups to get /// there. A deletion marker among the picked files must not be filtered out while the file /// holding the rows it masks stays behind, otherwise those rows become visible again. async fn test_compaction_input_limit_keeps_rows_deleted_with_format(flat_format: bool) { @@ -1406,9 +1406,9 @@ async fn test_compaction_input_limit_keeps_rows_deleted_with_format(flat_format: put_and_flush(&engine, region_id, &column_schemas, 0..3000).await; // Deletes 6 rows of that file. The markers land in a tiny file that overlaps it. delete_and_flush(&engine, region_id, &column_schemas, 10..16).await; - // 31 more tiny files that overlap the large one but not each other, so the window holds - // 33 file groups forming 2 runs. - for i in 2..33 { + // 15 more tiny files that overlap the large one but not each other, so the window holds + // 17 file groups forming 2 runs. + for i in 2..17 { put_and_flush(&engine, region_id, &column_schemas, i * 10..i * 10 + 6).await; } @@ -1417,7 +1417,7 @@ async fn test_compaction_input_limit_keeps_rows_deleted_with_format(flat_format: .await .unwrap(); assert_eq!( - 33, + 17, scanner.num_files(), "unexpected files: {:?}", scanner.file_ids() @@ -1429,7 +1429,7 @@ async fn test_compaction_input_limit_keeps_rows_deleted_with_format(flat_format: .scanner(region_id, ScanRequest::default()) .await .unwrap(); - // The 32 tiny files are merged into one; the large file exceeds the input file num limit + // The 16 tiny files are merged into one; the large file exceeds the input file num limit // and is left behind. assert_eq!( 2, diff --git a/src/mito2/src/region/options.rs b/src/mito2/src/region/options.rs index 7bfb8d0d65..9a8d17c954 100644 --- a/src/mito2/src/region/options.rs +++ b/src/mito2/src/region/options.rs @@ -172,6 +172,19 @@ impl RegionOptions { } ); } + let CompactionOptions::Twcs(options) = &self.compaction; + ensure!( + options.active_window_l1_merge_trigger >= 2, + InvalidRegionOptionsSnafu { + reason: "active_window.l1_merge_trigger must be at least 2", + } + ); + ensure!( + options.inactive_window_l1_merge_trigger >= 2, + InvalidRegionOptionsSnafu { + reason: "inactive_window.l1_merge_trigger must be at least 2", + } + ); Ok(()) } @@ -372,9 +385,22 @@ impl Default for CompactionOptions { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(default)] pub struct TwcsOptions { - /// Minimum file num in every time window to trigger a compaction. + /// Minimum file num in the active time window to trigger a compaction. #[serde_as(as = "DisplayFromStr")] - pub trigger_file_num: usize, + #[serde(rename = "active_window.trigger_file_num", alias = "trigger_file_num")] + pub active_window_trigger_file_num: usize, + /// Minimum L1 file num in the active window to allow a safety compaction. + #[serde_as(as = "DisplayFromStr")] + #[serde(rename = "active_window.l1_merge_trigger")] + pub active_window_l1_merge_trigger: usize, + /// Minimum file num in an inactive time window to trigger a compaction. + #[serde_as(as = "DisplayFromStr")] + #[serde(rename = "inactive_window.trigger_file_num")] + pub inactive_window_trigger_file_num: usize, + /// Minimum L1 file num to trigger a compaction in an inactive window. + #[serde_as(as = "DisplayFromStr")] + #[serde(rename = "inactive_window.l1_merge_trigger")] + pub inactive_window_l1_merge_trigger: usize, /// Compaction time window defined when creating tables. #[serde(with = "humantime_serde")] pub time_window: Option, @@ -407,7 +433,10 @@ impl TwcsOptions { impl Default for TwcsOptions { fn default() -> Self { Self { - trigger_file_num: 4, + active_window_trigger_file_num: 4, + active_window_l1_merge_trigger: 16, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, time_window: None, max_output_file_size: Some(ReadableSize::mb(512)), remote_compaction: false, @@ -702,14 +731,20 @@ mod tests { #[test] fn test_with_compaction_type() { let map = make_map(&[ - ("compaction.twcs.trigger_file_num", "8"), + ("compaction.twcs.active_window.trigger_file_num", "8"), + ("compaction.twcs.active_window.l1_merge_trigger", "16"), + ("compaction.twcs.inactive_window.trigger_file_num", "2"), + ("compaction.twcs.inactive_window.l1_merge_trigger", "12"), ("compaction.twcs.time_window", "2h"), ("compaction.type", "twcs"), ]); let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap(); let expect = RegionOptions { compaction: CompactionOptions::Twcs(TwcsOptions { - trigger_file_num: 8, + active_window_trigger_file_num: 8, + active_window_l1_merge_trigger: 16, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 12, time_window: Some(Duration::from_secs(3600 * 2)), ..Default::default() }), @@ -719,6 +754,60 @@ mod tests { assert_eq!(expect, options); } + #[test] + fn test_twcs_window_trigger_below_two_is_accepted_for_compatibility() { + // Tables created before the >= 2 ALTER-time check may have persisted + // trigger values of 1; region open must still accept them. + let map = make_map(&[ + ("compaction.twcs.active_window.trigger_file_num", "1"), + ("compaction.type", "twcs"), + ]); + let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap(); + let CompactionOptions::Twcs(twcs) = &options.compaction; + assert_eq!(1, twcs.active_window_trigger_file_num); + + let map = make_map(&[ + ("compaction.twcs.inactive_window.trigger_file_num", "1"), + ("compaction.type", "twcs"), + ]); + let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap(); + let CompactionOptions::Twcs(twcs) = &options.compaction; + assert_eq!(1, twcs.inactive_window_trigger_file_num); + } + + #[test] + fn test_active_window_l1_merge_trigger_below_two_is_rejected() { + let map = make_map(&[ + ("compaction.twcs.active_window.l1_merge_trigger", "1"), + ("compaction.type", "twcs"), + ]); + + let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err(); + assert_eq!(StatusCode::InvalidArguments, err.status_code()); + } + + #[test] + fn test_inactive_window_l1_merge_trigger_defaults_to_eight() { + let value = serde_json::to_value(TwcsOptions::default()).unwrap(); + assert_eq!( + Some("8"), + value + .get("inactive_window.l1_merge_trigger") + .and_then(|value| value.as_str()) + ); + } + + #[test] + fn test_inactive_window_l1_merge_trigger_below_two_is_rejected() { + let map = make_map(&[ + ("compaction.twcs.inactive_window.l1_merge_trigger", "1"), + ("compaction.type", "twcs"), + ]); + + let err = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap_err(); + assert_eq!(StatusCode::InvalidArguments, err.status_code()); + } + #[test] fn test_with_compaction_override_true_without_compaction_type() { let map = make_map(&[(COMPACTION_OVERRIDE, "true")]); @@ -982,7 +1071,10 @@ mod tests { ttl: Some(Duration::from_secs(3600 * 24 * 7).into()), auto_flush_interval: None, compaction: CompactionOptions::Twcs(TwcsOptions { - trigger_file_num: 8, + active_window_trigger_file_num: 8, + active_window_l1_merge_trigger: 16, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, time_window: Some(Duration::from_secs(3600 * 2)), max_output_file_size: Some(ReadableSize::gb(1)), remote_compaction: false, @@ -1042,7 +1134,10 @@ mod tests { ttl: Some(Duration::from_secs(3600 * 24 * 7).into()), auto_flush_interval: None, compaction: CompactionOptions::Twcs(TwcsOptions { - trigger_file_num: 8, + active_window_trigger_file_num: 8, + active_window_l1_merge_trigger: 8, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, time_window: Some(Duration::from_secs(3600 * 2)), max_output_file_size: None, remote_compaction: false, @@ -1077,6 +1172,9 @@ mod tests { let got: RegionOptions = serde_json::from_str(old_region_options_json_str).unwrap(); assert_eq!(None, got.write_buffer_size); assert!(!got.preserve_row_sequence); + let CompactionOptions::Twcs(twcs) = got.compaction; + assert_eq!(16, twcs.active_window_l1_merge_trigger); + assert_eq!(8, twcs.inactive_window_l1_merge_trigger); let default_json = serde_json::to_value(RegionOptions::default()).unwrap(); assert!(default_json.get(WRITE_BUFFER_SIZE_KEY).is_none()); @@ -1113,7 +1211,10 @@ mod tests { ttl: Some(Duration::from_secs(3600 * 24 * 7).into()), auto_flush_interval: None, compaction: CompactionOptions::Twcs(TwcsOptions { - trigger_file_num: 8, + active_window_trigger_file_num: 8, + active_window_l1_merge_trigger: 16, + inactive_window_trigger_file_num: 2, + inactive_window_l1_merge_trigger: 8, time_window: Some(Duration::from_secs(3600 * 2)), max_output_file_size: Some(ReadableSize::mb(7)), remote_compaction: false, diff --git a/src/mito2/src/test_util/scheduler_util.rs b/src/mito2/src/test_util/scheduler_util.rs index 9194094873..f97886b4ea 100644 --- a/src/mito2/src/test_util/scheduler_util.rs +++ b/src/mito2/src/test_util/scheduler_util.rs @@ -92,6 +92,14 @@ impl SchedulerEnv { pub(crate) fn mock_compaction_scheduler( &self, request_sender: Sender, + ) -> CompactionScheduler { + self.mock_compaction_scheduler_with_config(request_sender, MitoConfig::default()) + } + + pub(crate) fn mock_compaction_scheduler_with_config( + &self, + request_sender: Sender, + config: MitoConfig, ) -> CompactionScheduler { let scheduler = self.get_scheduler(); @@ -99,7 +107,7 @@ impl SchedulerEnv { scheduler, request_sender, Arc::new(CacheManager::default()), - Arc::new(MitoConfig::default()), + Arc::new(config), WorkerListener::default(), Plugins::new(), Arc::new(new_compaction_memory_manager(0)), diff --git a/src/mito2/src/worker/handle_alter.rs b/src/mito2/src/worker/handle_alter.rs index 57eceed337..b13c1badf8 100644 --- a/src/mito2/src/worker/handle_alter.rs +++ b/src/mito2/src/worker/handle_alter.rs @@ -246,6 +246,9 @@ impl RegionWorkerLoop { &value, region.region_id, )?; + if !value.is_empty() { + current_options.compaction_override = true; + } } SetRegionOption::Format(format_str) => { let new_format = format_str.parse::().map_err(|_| { @@ -429,10 +432,70 @@ fn set_twcs_options( region_id: RegionId, ) -> std::result::Result<(), MetadataError> { match key { - mito_engine_options::TWCS_TRIGGER_FILE_NUM => { - let files = parse_usize_with_default(key, value, default_option.trigger_file_num)?; - log_option_update(region_id, key, options.trigger_file_num, files); - options.trigger_file_num = files; + mito_engine_options::TWCS_TRIGGER_FILE_NUM + | mito_engine_options::TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM => { + let files = parse_usize_with_default( + key, + value, + default_option.active_window_trigger_file_num, + )?; + log_option_update( + region_id, + key, + options.active_window_trigger_file_num, + files, + ); + options.active_window_trigger_file_num = files; + } + mito_engine_options::TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER => { + let files = parse_usize_with_default( + key, + value, + default_option.active_window_l1_merge_trigger, + )?; + ensure!( + files >= 2, + InvalidSetRegionOptionRequestSnafu { key, value } + ); + log_option_update( + region_id, + key, + options.active_window_l1_merge_trigger, + files, + ); + options.active_window_l1_merge_trigger = files; + } + mito_engine_options::TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM => { + let files = parse_usize_with_default( + key, + value, + default_option.inactive_window_trigger_file_num, + )?; + log_option_update( + region_id, + key, + options.inactive_window_trigger_file_num, + files, + ); + options.inactive_window_trigger_file_num = files; + } + mito_engine_options::TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER => { + let files = parse_usize_with_default( + key, + value, + default_option.inactive_window_l1_merge_trigger, + )?; + ensure!( + files >= 2, + InvalidSetRegionOptionRequestSnafu { key, value } + ); + log_option_update( + region_id, + key, + options.inactive_window_l1_merge_trigger, + files, + ); + options.inactive_window_l1_merge_trigger = files; } mito_engine_options::TWCS_MAX_OUTPUT_FILE_SIZE => { let size = if value.is_empty() { @@ -551,6 +614,115 @@ fn need_change_index(kind: &AlterKind) -> bool { mod tests { use super::*; + #[test] + fn test_set_twcs_window_trigger_options() { + let mut options = TwcsOptions::default(); + let defaults = options.clone(); + let region_id = RegionId::new(1, 1); + + set_twcs_options( + &mut options, + &defaults, + "compaction.twcs.active_window.trigger_file_num", + "8", + region_id, + ) + .unwrap(); + set_twcs_options( + &mut options, + &defaults, + "compaction.twcs.active_window.l1_merge_trigger", + "16", + region_id, + ) + .unwrap(); + set_twcs_options( + &mut options, + &defaults, + "compaction.twcs.inactive_window.trigger_file_num", + "3", + region_id, + ) + .unwrap(); + set_twcs_options( + &mut options, + &defaults, + "compaction.twcs.inactive_window.l1_merge_trigger", + "12", + region_id, + ) + .unwrap(); + assert_eq!(8, options.active_window_trigger_file_num); + assert_eq!(16, options.active_window_l1_merge_trigger); + assert_eq!(3, options.inactive_window_trigger_file_num); + assert_eq!(12, options.inactive_window_l1_merge_trigger); + + set_twcs_options( + &mut options, + &defaults, + "compaction.twcs.active_window.trigger_file_num", + "", + region_id, + ) + .unwrap(); + set_twcs_options( + &mut options, + &defaults, + "compaction.twcs.active_window.l1_merge_trigger", + "", + region_id, + ) + .unwrap(); + set_twcs_options( + &mut options, + &defaults, + "compaction.twcs.inactive_window.trigger_file_num", + "", + region_id, + ) + .unwrap(); + set_twcs_options( + &mut options, + &defaults, + "compaction.twcs.inactive_window.l1_merge_trigger", + "", + region_id, + ) + .unwrap(); + assert_eq!(defaults, options); + } + + #[test] + fn test_set_twcs_window_trigger_accepts_one() { + let defaults = TwcsOptions::default(); + for key in [ + "compaction.twcs.trigger_file_num", + "compaction.twcs.active_window.trigger_file_num", + "compaction.twcs.inactive_window.trigger_file_num", + ] { + let mut options = defaults.clone(); + assert!( + set_twcs_options(&mut options, &defaults, key, "1", RegionId::new(1, 1)).is_ok(), + "{key}" + ); + } + } + + #[test] + fn test_set_twcs_l1_merge_triggers_reject_one() { + let defaults = TwcsOptions::default(); + for key in [ + "compaction.twcs.active_window.l1_merge_trigger", + "compaction.twcs.inactive_window.l1_merge_trigger", + ] { + let mut options = TwcsOptions::default(); + assert!( + set_twcs_options(&mut options, &defaults, key, "1", RegionId::new(1, 1)).is_err(), + "{key}" + ); + } + } + #[test] fn test_new_region_options_with_idempotent_append_mode() { let current_options = RegionOptions::default(); diff --git a/src/mito2/src/worker/handle_compaction.rs b/src/mito2/src/worker/handle_compaction.rs index 1f8a66887c..ad7872b723 100644 --- a/src/mito2/src/worker/handle_compaction.rs +++ b/src/mito2/src/worker/handle_compaction.rs @@ -30,6 +30,10 @@ use crate::request::{ use crate::sst::index::IndexBuildType; use crate::worker::RegionWorkerLoop; +fn made_progress(files_to_add: usize, files_to_remove: usize) -> bool { + files_to_add > 0 || files_to_remove > files_to_add +} + impl RegionWorkerLoop { pub(crate) async fn handle_compaction_pick_finished( &mut self, @@ -131,11 +135,10 @@ impl RegionWorkerLoop { return; } let execution = request.execution.clone(); - // Whether this execution reduced the SST file count. The scheduler keeps - // draining while compaction makes progress; a rewrite that did not reduce - // files (e.g. its output was split into more files than its input) must end - // the chain to avoid a no-progress loop. - let made_progress = request.edit.files_to_remove.len() > request.edit.files_to_add.len(); + let made_progress = made_progress( + request.edit.files_to_add.len(), + request.edit.files_to_remove.len(), + ); region.version_control.apply_edit( Some(request.edit.clone()), @@ -270,3 +273,25 @@ impl RegionWorkerLoop { } } } + +#[cfg(test)] +mod tests { + use super::made_progress; + + #[test] + fn test_nonempty_output_or_file_reduction_is_progress() { + for (files_to_add, files_to_remove, expected) in [ + (3, 3, true), // Equal-count rewrite. + (1, 3, true), // Ordinary reduction. + (0, 3, true), // Zero-output reduction. + (0, 0, false), // Empty edit. + (3, 2, true), // Growth rewrite with output. + ] { + assert_eq!( + expected, + made_progress(files_to_add, files_to_remove), + "files_to_remove: {files_to_remove}, files_to_add: {files_to_add}" + ); + } + } +} diff --git a/src/sql/src/error.rs b/src/sql/src/error.rs index dc819bda01..1588842a36 100644 --- a/src/sql/src/error.rs +++ b/src/sql/src/error.rs @@ -164,6 +164,15 @@ pub enum Error { location: Location, }, + #[snafu(display("Invalid database option value for {}: {}, {}", key, value, reason))] + InvalidDatabaseOptionValue { + key: String, + value: String, + reason: String, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Invalid table name: {}", name))] InvalidTableName { name: String, @@ -375,6 +384,7 @@ impl ErrorExt for Error { | InvalidExprAsOptionValue { .. } | InvalidDatabaseName { .. } | InvalidDatabaseOption { .. } + | InvalidDatabaseOptionValue { .. } | ColumnTypeMismatch { .. } | InvalidTableName { .. } | InvalidFlowName { .. } diff --git a/src/sql/src/parsers/create_parser.rs b/src/sql/src/parsers/create_parser.rs index b1ea56df2e..81ff23d3bc 100644 --- a/src/sql/src/parsers/create_parser.rs +++ b/src/sql/src/parsers/create_parser.rs @@ -35,13 +35,13 @@ use sqlparser::keywords::ALL_KEYWORDS; use sqlparser::parser::IsOptional::Mandatory; use sqlparser::parser::{Parser, ParserError}; use sqlparser::tokenizer::{Token, TokenWithSpan, Word}; -use table::requests::validate_database_option; +use table::requests::{validate_database_option, validate_database_option_value}; use crate::ast::{ColumnDef, Ident, ObjectNamePartExt}; use crate::error::{ - self, InvalidColumnOptionSnafu, InvalidDatabaseOptionSnafu, InvalidFlowQuerySnafu, - InvalidIntervalSnafu, InvalidSqlSnafu, InvalidTimeIndexSnafu, MissingTimeIndexSnafu, Result, - SyntaxSnafu, UnexpectedSnafu, UnsupportedSnafu, + self, InvalidColumnOptionSnafu, InvalidDatabaseOptionSnafu, InvalidDatabaseOptionValueSnafu, + InvalidFlowQuerySnafu, InvalidIntervalSnafu, InvalidSqlSnafu, InvalidTimeIndexSnafu, + MissingTimeIndexSnafu, Result, SyntaxSnafu, UnexpectedSnafu, UnsupportedSnafu, }; use crate::parser::{FLOW, ParserContext}; use crate::parsers::tql_parser; @@ -223,11 +223,22 @@ impl<'a> ParserContext<'a> { .map(parse_option_string) .collect::>>()?; - for key in options.keys() { + for (key, option_value) in &options { ensure!( validate_database_option(key), InvalidDatabaseOptionSnafu { key: key.clone() } ); + let option_value_str = option_value.as_string(); + validate_database_option_value(key, option_value_str).map_err(|reason| { + InvalidDatabaseOptionValueSnafu { + key: key.clone(), + value: option_value_str + .map(str::to_owned) + .unwrap_or_else(|| option_value.to_string()), + reason: reason.to_string(), + } + .build() + })?; } if let Some(append_mode) = options.get("append_mode").and_then(|x| x.as_string()) && append_mode == "true" @@ -1581,6 +1592,70 @@ mod tests { } } + #[test] + fn test_parse_create_database_option_validation() { + let overflow = format!("{}0", usize::MAX); + for key in [ + "compaction.twcs.trigger_file_num", + "compaction.twcs.active_window.trigger_file_num", + "compaction.twcs.inactive_window.trigger_file_num", + ] { + for invalid in ["invalid", "-1", overflow.as_str()] { + let sql = format!("CREATE DATABASE invalid WITH ('{key}'='{invalid}')"); + let err = ParserContext::create_with_dialect( + &sql, + &GreptimeDbDialect {}, + ParseOptions::default(), + ) + .unwrap_err(); + assert_eq!( + err.to_string(), + format!( + "Invalid database option value for {key}: {invalid}, expected a non-negative integer fitting in usize" + ) + ); + } + for valid in ["0", "1"] { + let sql = format!("CREATE DATABASE valid WITH ('{key}'='{valid}')"); + ParserContext::create_with_dialect( + &sql, + &GreptimeDbDialect {}, + ParseOptions::default(), + ) + .unwrap(); + } + } + for key in [ + "compaction.twcs.active_window.l1_merge_trigger", + "compaction.twcs.inactive_window.l1_merge_trigger", + ] { + let sql = format!("CREATE DATABASE invalid WITH ('{key}'='1')"); + let err = ParserContext::create_with_dialect( + &sql, + &GreptimeDbDialect {}, + ParseOptions::default(), + ) + .unwrap_err(); + assert_eq!( + format!( + "Invalid database option value for {key}: 1, expected an integer greater than or equal to 2" + ), + err.to_string() + ); + } + + let sql = + "CREATE DATABASE valid WITH ('compaction.twcs.active_window.l1_merge_trigger'='2')"; + ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default()) + .unwrap(); + + let sql = "CREATE DATABASE invalid WITH ('unknown'='1')"; + let err = + ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default()) + .unwrap_err(); + assert_eq!("Unrecognized database option key: unknown", err.to_string()); + } + #[test] fn test_parse_create_flow_more_testcases() { use pretty_assertions::assert_eq; diff --git a/src/store-api/src/mito_engine_options.rs b/src/store-api/src/mito_engine_options.rs index 6890f06cc3..c308e91565 100644 --- a/src/store-api/src/mito_engine_options.rs +++ b/src/store-api/src/mito_engine_options.rs @@ -15,6 +15,8 @@ //! Option keys for the mito engine. //! We define them in this mod so the create parser can use it to validate table options. +use std::collections::HashMap; + /// Option key for all WAL options. pub use common_wal::options::WAL_OPTIONS_KEY; /// Option key for append mode. @@ -37,6 +39,18 @@ pub const COMPACTION_OVERRIDE: &str = "compaction.override"; pub const COMPACTION_TYPE_TWCS: &str = "twcs"; /// Option key for twcs min file num to trigger a compaction. pub const TWCS_TRIGGER_FILE_NUM: &str = "compaction.twcs.trigger_file_num"; +/// Option key for twcs min file num to trigger compaction in the active window. +pub const TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM: &str = + "compaction.twcs.active_window.trigger_file_num"; +/// Option key for the active-window L1 safety compaction threshold. +pub const TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER: &str = + "compaction.twcs.active_window.l1_merge_trigger"; +/// Option key for twcs min file num to trigger compaction in an inactive window. +pub const TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM: &str = + "compaction.twcs.inactive_window.trigger_file_num"; +/// Option key for the inactive-window L1 compaction threshold. +pub const TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER: &str = + "compaction.twcs.inactive_window.l1_merge_trigger"; /// Option key for twcs max output file size. pub const TWCS_MAX_OUTPUT_FILE_SIZE: &str = "compaction.twcs.max_output_file_size"; /// Option key for twcs time window. @@ -76,6 +90,36 @@ pub const MAX_ROW_GROUP_ROW_COUNT_LIMIT: usize = 10 * 1024 * 1024; pub const PRESERVE_ROW_SEQUENCE: &str = "preserve_row_sequence"; // Note: Adding new options here should also check if this option should be removed in [metric_engine::engine::create::region_options_for_metadata_region]. +/// Conflicting values supplied through the legacy and canonical TWCS trigger options. +#[derive(Debug, PartialEq, Eq)] +pub struct TwcsTriggerOptionConflict { + /// Value supplied under [`TWCS_TRIGGER_FILE_NUM`]. + pub legacy_value: String, + /// Value supplied under [`TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM`]. + pub canonical_value: String, +} + +/// Normalizes the active-window TWCS trigger option to its legacy storage key. +pub fn normalize_twcs_trigger_options( + options: &mut HashMap, +) -> Result<(), TwcsTriggerOptionConflict> { + let Some(canonical_value) = options.get(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM).cloned() else { + return Ok(()); + }; + if let Some(legacy_value) = options.get(TWCS_TRIGGER_FILE_NUM) + && legacy_value != &canonical_value + { + return Err(TwcsTriggerOptionConflict { + legacy_value: legacy_value.clone(), + canonical_value, + }); + } + + options.remove(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM); + options.insert(TWCS_TRIGGER_FILE_NUM.to_string(), canonical_value); + Ok(()) +} + /// Returns true if the `key` is a valid option key for the mito engine. pub fn is_mito_engine_option_key(key: &str) -> bool { [ @@ -85,6 +129,10 @@ pub fn is_mito_engine_option_key(key: &str) -> bool { COMPACTION_TYPE, COMPACTION_OVERRIDE, TWCS_TRIGGER_FILE_NUM, + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, + TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER, + TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM, + TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW, TWCS_REMOTE_COMPACTION, @@ -113,6 +161,8 @@ pub fn is_mito_engine_option_key(key: &str) -> bool { #[cfg(test)] mod tests { + use std::collections::HashMap; + use super::*; #[test] @@ -125,6 +175,18 @@ mod tests { assert!(is_mito_engine_option_key( "compaction.twcs.trigger_file_num" )); + assert!(is_mito_engine_option_key( + "compaction.twcs.active_window.trigger_file_num" + )); + assert!(is_mito_engine_option_key( + "compaction.twcs.active_window.l1_merge_trigger" + )); + assert!(is_mito_engine_option_key( + "compaction.twcs.inactive_window.trigger_file_num" + )); + assert!(is_mito_engine_option_key( + "compaction.twcs.inactive_window.l1_merge_trigger" + )); assert!(is_mito_engine_option_key("compaction.twcs.time_window")); assert!(is_mito_engine_option_key("storage")); assert!(is_mito_engine_option_key( @@ -157,4 +219,44 @@ mod tests { assert!(is_mito_engine_option_key("preserve_row_sequence")); assert!(!is_mito_engine_option_key("foo")); } + + #[test] + fn test_normalize_twcs_trigger_aliases_to_legacy_key() { + let expected = HashMap::from([(TWCS_TRIGGER_FILE_NUM.to_string(), "4".to_string())]); + for mut options in [ + HashMap::from([(TWCS_TRIGGER_FILE_NUM.to_string(), "4".to_string())]), + HashMap::from([( + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM.to_string(), + "4".to_string(), + )]), + HashMap::from([ + (TWCS_TRIGGER_FILE_NUM.to_string(), "4".to_string()), + ( + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM.to_string(), + "4".to_string(), + ), + ]), + ] { + normalize_twcs_trigger_options(&mut options).unwrap(); + assert_eq!(expected, options); + } + } + + #[test] + fn test_normalize_twcs_trigger_conflicting_aliases() { + let mut options = HashMap::from([ + (TWCS_TRIGGER_FILE_NUM.to_string(), "4".to_string()), + ( + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM.to_string(), + "8".to_string(), + ), + ]); + let original = options.clone(); + + let error = normalize_twcs_trigger_options(&mut options).unwrap_err(); + + assert_eq!("4", error.legacy_value); + assert_eq!("8", error.canonical_value); + assert_eq!(original, options); + } } diff --git a/src/store-api/src/region_request.rs b/src/store-api/src/region_request.rs index 7cbf5b1a7f..137087bc57 100644 --- a/src/store-api/src/region_request.rs +++ b/src/store-api/src/region_request.rs @@ -58,6 +58,8 @@ use crate::metrics; use crate::mito_engine_options::{ APPEND_MODE_KEY, AUTO_FLUSH_INTERVAL_KEY, MAX_ROW_GROUP_ROW_COUNT, MAX_ROW_GROUP_ROW_COUNT_LIMIT, PRESERVE_ROW_SEQUENCE, SKIP_WAL_KEY, SST_FORMAT_KEY, TTL_KEY, + TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, + TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW, TWCS_TRIGGER_FILE_NUM, WRITE_BUFFER_SIZE_KEY, }; use crate::path_utils::table_dir; @@ -1517,9 +1519,13 @@ impl TryFrom<&PbOption> for SetRegionOption { Ok(Self::Ttl(Some(ttl))) } - TWCS_TRIGGER_FILE_NUM | TWCS_MAX_OUTPUT_FILE_SIZE | TWCS_TIME_WINDOW => { - Ok(Self::Twsc(key.clone(), value.clone())) - } + TWCS_TRIGGER_FILE_NUM + | TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM + | TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER + | TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM + | TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER + | TWCS_MAX_OUTPUT_FILE_SIZE + | TWCS_TIME_WINDOW => Ok(Self::Twsc(key.clone(), value.clone())), SST_FORMAT_KEY => Ok(Self::Format(value.clone())), APPEND_MODE_KEY => { let append_mode = value @@ -1572,6 +1578,18 @@ impl From<&UnsetRegionOption> for SetRegionOption { UnsetRegionOption::TwcsTriggerFileNum => { SetRegionOption::Twsc(unset_option.to_string(), String::new()) } + UnsetRegionOption::TwcsActiveWindowTriggerFileNum => { + SetRegionOption::Twsc(unset_option.to_string(), String::new()) + } + UnsetRegionOption::TwcsActiveWindowL1MergeTrigger => { + SetRegionOption::Twsc(unset_option.to_string(), String::new()) + } + UnsetRegionOption::TwcsInactiveWindowTriggerFileNum => { + SetRegionOption::Twsc(unset_option.to_string(), String::new()) + } + UnsetRegionOption::TwcsInactiveWindowL1MergeTrigger => { + SetRegionOption::Twsc(unset_option.to_string(), String::new()) + } UnsetRegionOption::TwcsMaxOutputFileSize => { SetRegionOption::Twsc(unset_option.to_string(), String::new()) } @@ -1594,6 +1612,10 @@ impl TryFrom<&str> for UnsetRegionOption { TTL_KEY => Ok(Self::Ttl), WRITE_BUFFER_SIZE_KEY => Ok(Self::WriteBufferSize), TWCS_TRIGGER_FILE_NUM => Ok(Self::TwcsTriggerFileNum), + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM => Ok(Self::TwcsActiveWindowTriggerFileNum), + TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER => Ok(Self::TwcsActiveWindowL1MergeTrigger), + TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM => Ok(Self::TwcsInactiveWindowTriggerFileNum), + TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER => Ok(Self::TwcsInactiveWindowL1MergeTrigger), TWCS_MAX_OUTPUT_FILE_SIZE => Ok(Self::TwcsMaxOutputFileSize), TWCS_TIME_WINDOW => Ok(Self::TwcsTimeWindow), MAX_ROW_GROUP_ROW_COUNT => Ok(Self::MaxRowGroupRowCount), @@ -1606,12 +1628,16 @@ impl TryFrom<&str> for UnsetRegionOption { #[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)] pub enum UnsetRegionOption { TwcsTriggerFileNum, + TwcsActiveWindowTriggerFileNum, + TwcsInactiveWindowTriggerFileNum, + TwcsInactiveWindowL1MergeTrigger, TwcsMaxOutputFileSize, TwcsTimeWindow, Ttl, MaxRowGroupRowCount, WriteBufferSize, PreserveRowSequence, + TwcsActiveWindowL1MergeTrigger, } impl UnsetRegionOption { @@ -1620,6 +1646,10 @@ impl UnsetRegionOption { Self::Ttl => TTL_KEY, Self::WriteBufferSize => WRITE_BUFFER_SIZE_KEY, Self::TwcsTriggerFileNum => TWCS_TRIGGER_FILE_NUM, + Self::TwcsActiveWindowTriggerFileNum => TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, + Self::TwcsActiveWindowL1MergeTrigger => TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER, + Self::TwcsInactiveWindowTriggerFileNum => TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM, + Self::TwcsInactiveWindowL1MergeTrigger => TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER, Self::TwcsMaxOutputFileSize => TWCS_MAX_OUTPUT_FILE_SIZE, Self::TwcsTimeWindow => TWCS_TIME_WINDOW, Self::MaxRowGroupRowCount => MAX_ROW_GROUP_ROW_COUNT, @@ -2063,6 +2093,58 @@ mod tests { ); } + #[test] + fn test_set_twcs_window_trigger_options_try_from() { + for key in [ + "compaction.twcs.active_window.trigger_file_num", + "compaction.twcs.active_window.l1_merge_trigger", + "compaction.twcs.inactive_window.trigger_file_num", + "compaction.twcs.inactive_window.l1_merge_trigger", + ] { + let option = PbOption { + key: key.to_string(), + value: "8".to_string(), + }; + assert_eq!( + SetRegionOption::Twsc(key.to_string(), "8".to_string()), + SetRegionOption::try_from(&option).unwrap(), + "{key}" + ); + } + } + + #[test] + fn test_unset_twcs_window_trigger_options_try_from() { + for (key, expected) in [ + ( + "compaction.twcs.active_window.trigger_file_num", + UnsetRegionOption::TwcsActiveWindowTriggerFileNum, + ), + ( + "compaction.twcs.active_window.l1_merge_trigger", + UnsetRegionOption::TwcsActiveWindowL1MergeTrigger, + ), + ( + "compaction.twcs.inactive_window.trigger_file_num", + UnsetRegionOption::TwcsInactiveWindowTriggerFileNum, + ), + ] { + assert_eq!(expected, UnsetRegionOption::try_from(key).unwrap()); + assert_eq!( + SetRegionOption::Twsc(key.to_string(), String::new()), + SetRegionOption::from(&expected) + ); + } + + let key = "compaction.twcs.inactive_window.l1_merge_trigger"; + let option = UnsetRegionOption::try_from(key).unwrap(); + assert_eq!(key, option.to_string()); + assert_eq!( + SetRegionOption::Twsc(key.to_string(), String::new()), + SetRegionOption::from(&option) + ); + } + #[test] fn test_from_proto_alter_request() { RegionAlterRequest::try_from(AlterRequest { diff --git a/src/table/src/error.rs b/src/table/src/error.rs index 7fd04b26bf..2c8da6c29b 100644 --- a/src/table/src/error.rs +++ b/src/table/src/error.rs @@ -135,6 +135,22 @@ pub enum Error { location: Location, }, + #[snafu(display( + "Conflicting table options: {}={} and {}={}", + first_key, + first_value, + second_key, + second_value + ))] + ConflictingTableOptions { + first_key: String, + first_value: String, + second_key: String, + second_value: String, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Invalid alter table({}) request: {}", table, err))] InvalidAlterRequest { table: String, @@ -223,7 +239,9 @@ impl ErrorExt for Error { Error::InvalidColumnOption { .. } => StatusCode::InvalidArguments, Error::ColumnNotExists { .. } => StatusCode::TableColumnNotFound, Error::Unsupported { .. } => StatusCode::Unsupported, - Error::ParseTableOption { .. } => StatusCode::InvalidArguments, + Error::ParseTableOption { .. } | Error::ConflictingTableOptions { .. } => { + StatusCode::InvalidArguments + } Error::MissingTimeIndexColumn { .. } => StatusCode::IllegalState, Error::SetSkippingOptions { .. } | Error::UnsetSkippingOptions { .. } diff --git a/src/table/src/metadata.rs b/src/table/src/metadata.rs index 46175ed116..69803c7855 100644 --- a/src/table/src/metadata.rs +++ b/src/table/src/metadata.rs @@ -32,6 +32,7 @@ use store_api::metric_engine_consts::PHYSICAL_TABLE_METADATA_KEY; use store_api::mito_engine_options::{ APPEND_MODE_KEY, AUTO_FLUSH_INTERVAL_KEY, COMPACTION_TYPE, COMPACTION_TYPE_TWCS, MAX_ROW_GROUP_ROW_COUNT, MERGE_MODE_KEY, PRESERVE_ROW_SEQUENCE, SKIP_WAL_KEY, SST_FORMAT_KEY, + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_TRIGGER_FILE_NUM, }; use store_api::region_request::{SetRegionOption, UnsetRegionOption}; use store_api::storage::{ColumnDescriptor, ColumnDescriptorBuilder, ColumnId}; @@ -362,8 +363,22 @@ impl TableMeta { new_options.ttl = *new_ttl; } SetRegionOption::Twsc(key, value) => { + let persisted_key = if matches!( + key.as_str(), + TWCS_TRIGGER_FILE_NUM | TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM + ) { + new_options.extra_options.remove(TWCS_TRIGGER_FILE_NUM); + new_options + .extra_options + .remove(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM); + TWCS_TRIGGER_FILE_NUM + } else { + key + }; if !value.is_empty() { - new_options.extra_options.insert(key.clone(), value.clone()); + new_options + .extra_options + .insert(persisted_key.to_string(), value.clone()); // Ensure node restart correctly. new_options.extra_options.insert( COMPACTION_TYPE.to_string(), @@ -371,7 +386,7 @@ impl TableMeta { ); } else { // Invalidate the previous change option if an empty value has been set. - new_options.extra_options.remove(key.as_str()); + new_options.extra_options.remove(persisted_key); } } SetRegionOption::Format(value) => { @@ -2224,6 +2239,49 @@ mod tests { ); } + #[test] + fn test_set_twcs_trigger_persists_legacy_key() { + for key in [TWCS_TRIGGER_FILE_NUM, TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM] { + let mut table_options = TableOptions::default(); + table_options.extra_options.insert( + TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM.to_string(), + "4".to_string(), + ); + let meta = TableMetaBuilder::empty() + .schema(Arc::new(new_test_schema())) + .primary_key_indices(vec![0]) + .engine("engine") + .next_column_id(3) + .options(table_options) + .build() + .unwrap(); + let alter_kind = AlterKind::SetTableOptions { + options: vec![SetRegionOption::Twsc(key.to_string(), "8".to_string())], + }; + + let new_meta = meta + .builder_with_alter_kind("my_table", &alter_kind) + .unwrap() + .build() + .unwrap(); + + assert_eq!( + Some("8"), + new_meta + .options + .extra_options + .get(TWCS_TRIGGER_FILE_NUM) + .map(String::as_str) + ); + assert!( + !new_meta + .options + .extra_options + .contains_key(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM) + ); + } + } + #[test] fn test_set_and_unset_max_row_group_row_count() { let meta = TableMetaBuilder::empty() diff --git a/src/table/src/requests.rs b/src/table/src/requests.rs index aea45130ed..3c9d0679ae 100644 --- a/src/table/src/requests.rs +++ b/src/table/src/requests.rs @@ -39,12 +39,14 @@ use store_api::mito_engine_options::{ APPEND_MODE_KEY, COMPACTION_TYPE, MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD, MEMTABLE_BULK_ENCODE_ROW_THRESHOLD, MEMTABLE_BULK_MAX_MERGE_GROUPS, MEMTABLE_BULK_MERGE_THRESHOLD, MEMTABLE_TYPE, MERGE_MODE_KEY, SST_FORMAT_KEY, - TWCS_FALLBACK_TO_LOCAL, TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW, TWCS_TRIGGER_FILE_NUM, - is_mito_engine_option_key, + TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, + TWCS_FALLBACK_TO_LOCAL, TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER, + TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW, + TWCS_TRIGGER_FILE_NUM, is_mito_engine_option_key, normalize_twcs_trigger_options, }; use store_api::region_request::{SetRegionOption, UnsetRegionOption}; -use crate::error::{ParseTableOptionSnafu, Result}; +use crate::error::{ConflictingTableOptionsSnafu, ParseTableOptionSnafu, Result}; use crate::metadata::{TableId, TableVersion}; use crate::table_reference::TableReference; @@ -118,6 +120,10 @@ static VALID_DB_OPT_KEYS: Lazy> = Lazy::new(|| { set.insert(TWCS_FALLBACK_TO_LOCAL); set.insert(TWCS_TIME_WINDOW); set.insert(TWCS_TRIGGER_FILE_NUM); + set.insert(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM); + set.insert(TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER); + set.insert(TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM); + set.insert(TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER); set.insert(TWCS_MAX_OUTPUT_FILE_SIZE); set.insert(SST_FORMAT_KEY); set @@ -128,6 +134,32 @@ pub fn validate_database_option(key: &str) -> bool { VALID_DB_OPT_KEYS.contains(&key) } +/// Validates a database option value, returning the violated constraint on error. +pub fn validate_database_option_value( + key: &str, + value: Option<&str>, +) -> std::result::Result<(), &'static str> { + let (minimum, constraint) = match key { + TWCS_TRIGGER_FILE_NUM + | TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM + | TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM => { + (0, "expected a non-negative integer fitting in usize") + } + TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER | TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER => { + (2, "expected an integer greater than or equal to 2") + } + _ => return Ok(()), + }; + if value + .and_then(|value| value.parse::().ok()) + .is_some_and(|files| files >= minimum) + { + Ok(()) + } else { + Err(constraint) + } +} + /// Returns true if the `key` is a valid key for any engine or storage. pub fn validate_table_option(key: &str) -> bool { if is_supported_in_s3(key) { @@ -183,11 +215,21 @@ impl TableOptions { ) -> Result { let mut options = TableOptions::default(); - let kvs: HashMap = iter + let mut kvs: HashMap = iter .into_iter() .map(|(k, v)| (k.to_string(), v.to_string())) .collect(); + normalize_twcs_trigger_options(&mut kvs).map_err(|conflict| { + ConflictingTableOptionsSnafu { + first_key: TWCS_TRIGGER_FILE_NUM, + first_value: conflict.legacy_value, + second_key: TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, + second_value: conflict.canonical_value, + } + .build() + })?; + if let Some(write_buffer_size) = kvs.get(WRITE_BUFFER_SIZE_KEY) { let size = ReadableSize::from_str(write_buffer_size).map_err(|_| { ParseTableOptionSnafu { @@ -716,6 +758,9 @@ pub struct CopyQueryToRequest { mod tests { use std::time::Duration; + use common_error::ext::ErrorExt; + use common_error::status_code::StatusCode; + use super::*; #[test] @@ -748,9 +793,60 @@ mod tests { MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD )); assert!(validate_database_option(MEMTABLE_BULK_MAX_MERGE_GROUPS)); + assert!(validate_database_option( + "compaction.twcs.active_window.trigger_file_num" + )); + assert!(validate_database_option( + "compaction.twcs.active_window.l1_merge_trigger" + )); + assert!(validate_database_option( + "compaction.twcs.inactive_window.trigger_file_num" + )); + assert!(validate_database_option( + "compaction.twcs.inactive_window.l1_merge_trigger" + )); assert!(!validate_database_option("foo")); } + #[test] + fn test_database_trigger_value_boundaries() { + let maximum = usize::MAX.to_string(); + let overflow = format!("{maximum}0"); + for (key, minimum) in [ + (TWCS_TRIGGER_FILE_NUM, 0), + (TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, 0), + (TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM, 0), + (TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER, 2), + (TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER, 2), + ] { + for invalid in [ + None, + Some(""), + Some("invalid"), + Some("-1"), + Some(overflow.as_str()), + ] { + assert!( + validate_database_option_value(key, invalid).is_err(), + "{key}: {invalid:?}" + ); + } + for valid in ["2", maximum.as_str()] { + assert!( + validate_database_option_value(key, Some(valid)).is_ok(), + "{key}: {valid}" + ); + } + for boundary in ["0", "1"] { + assert_eq!( + validate_database_option_value(key, Some(boundary)).is_ok(), + minimum == 0, + "{key}: {boundary}" + ); + } + } + } + #[test] fn test_serialize_table_options() { let options = TableOptions { @@ -820,6 +916,37 @@ mod tests { assert_eq!(options, serialized); } + #[test] + fn test_table_options_normalizes_twcs_trigger_aliases() { + for options in [ + vec![(TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, "4")], + vec![ + (TWCS_TRIGGER_FILE_NUM, "4"), + (TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, "4"), + ], + ] { + let table_options = TableOptions::try_from_iter(options).unwrap(); + assert_eq!( + HashMap::from([(TWCS_TRIGGER_FILE_NUM.to_string(), "4".to_string())]), + table_options.extra_options + ); + } + } + + #[test] + fn test_table_options_rejects_conflicting_twcs_trigger_aliases() { + let error = TableOptions::try_from_iter([ + (TWCS_TRIGGER_FILE_NUM, "4"), + (TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, "8"), + ]) + .unwrap_err(); + assert_eq!(StatusCode::InvalidArguments, error.status_code()); + assert_eq!( + "Conflicting table options: compaction.twcs.trigger_file_num=4 and compaction.twcs.active_window.trigger_file_num=8", + error.to_string() + ); + } + #[test] fn test_table_options_to_string() { let options = TableOptions { diff --git a/tests/cases/standalone/common/alter/alter_database.result b/tests/cases/standalone/common/alter/alter_database.result index 2fccce10de..0c07b75cbc 100644 --- a/tests/cases/standalone/common/alter/alter_database.result +++ b/tests/cases/standalone/common/alter/alter_database.result @@ -139,23 +139,79 @@ ALTER DATABASE alter_database SET 'compaction.twcs.trigger_file_num'='8'; Affected Rows: 0 +ALTER DATABASE alter_database SET 'compaction.twcs.active_window.trigger_file_num'='16'; + +Affected Rows: 0 + +ALTER DATABASE alter_database SET 'compaction.twcs.active_window.l1_merge_trigger'='12'; + +Affected Rows: 0 + +ALTER DATABASE alter_database SET 'compaction.twcs.inactive_window.trigger_file_num'='6'; + +Affected Rows: 0 + +ALTER DATABASE alter_database SET 'compaction.twcs.inactive_window.l1_merge_trigger'='10'; + +Affected Rows: 0 + +SHOW CREATE DATABASE alter_database; + ++----------------+--------------------------------------------------------------+ +| Database | Create Database | ++----------------+--------------------------------------------------------------+ +| alter_database | CREATE DATABASE IF NOT EXISTS alter_database | +| | WITH( | +| | 'compaction.twcs.active_window.l1_merge_trigger' = '12', | +| | 'compaction.twcs.inactive_window.l1_merge_trigger' = '10', | +| | 'compaction.twcs.inactive_window.trigger_file_num' = '6', | +| | 'compaction.twcs.time_window' = '2h', | +| | 'compaction.twcs.trigger_file_num' = '16', | +| | 'compaction.type' = 'twcs' | +| | ) | ++----------------+--------------------------------------------------------------+ + +-- SQLNESS ARG restart=true +SHOW CREATE DATABASE alter_database; + ++----------------+--------------------------------------------------------------+ +| Database | Create Database | ++----------------+--------------------------------------------------------------+ +| alter_database | CREATE DATABASE IF NOT EXISTS alter_database | +| | WITH( | +| | 'compaction.twcs.active_window.l1_merge_trigger' = '12', | +| | 'compaction.twcs.inactive_window.l1_merge_trigger' = '10', | +| | 'compaction.twcs.inactive_window.trigger_file_num' = '6', | +| | 'compaction.twcs.time_window' = '2h', | +| | 'compaction.twcs.trigger_file_num' = '16', | +| | 'compaction.type' = 'twcs' | +| | ) | ++----------------+--------------------------------------------------------------+ + +ALTER DATABASE alter_database SET 'compaction.twcs.trigger_file_num'='8'; + +Affected Rows: 0 + ALTER DATABASE alter_database SET 'compaction.twcs.max_output_file_size'='512MB'; Affected Rows: 0 SHOW CREATE DATABASE alter_database; -+----------------+-----------------------------------------------------+ -| Database | Create Database | -+----------------+-----------------------------------------------------+ -| alter_database | CREATE DATABASE IF NOT EXISTS alter_database | -| | WITH( | -| | 'compaction.twcs.max_output_file_size' = '512MB', | -| | 'compaction.twcs.time_window' = '2h', | -| | 'compaction.twcs.trigger_file_num' = '8', | -| | 'compaction.type' = 'twcs' | -| | ) | -+----------------+-----------------------------------------------------+ ++----------------+--------------------------------------------------------------+ +| Database | Create Database | ++----------------+--------------------------------------------------------------+ +| alter_database | CREATE DATABASE IF NOT EXISTS alter_database | +| | WITH( | +| | 'compaction.twcs.active_window.l1_merge_trigger' = '12', | +| | 'compaction.twcs.inactive_window.l1_merge_trigger' = '10', | +| | 'compaction.twcs.inactive_window.trigger_file_num' = '6', | +| | 'compaction.twcs.max_output_file_size' = '512MB', | +| | 'compaction.twcs.time_window' = '2h', | +| | 'compaction.twcs.trigger_file_num' = '8', | +| | 'compaction.type' = 'twcs' | +| | ) | ++----------------+--------------------------------------------------------------+ ALTER DATABASE alter_database SET 'compaction.twcs.time_window'='1d'; @@ -163,37 +219,55 @@ Affected Rows: 0 SHOW CREATE DATABASE alter_database; -+----------------+-----------------------------------------------------+ -| Database | Create Database | -+----------------+-----------------------------------------------------+ -| alter_database | CREATE DATABASE IF NOT EXISTS alter_database | -| | WITH( | -| | 'compaction.twcs.max_output_file_size' = '512MB', | -| | 'compaction.twcs.time_window' = '1d', | -| | 'compaction.twcs.trigger_file_num' = '8', | -| | 'compaction.type' = 'twcs' | -| | ) | -+----------------+-----------------------------------------------------+ ++----------------+--------------------------------------------------------------+ +| Database | Create Database | ++----------------+--------------------------------------------------------------+ +| alter_database | CREATE DATABASE IF NOT EXISTS alter_database | +| | WITH( | +| | 'compaction.twcs.active_window.l1_merge_trigger' = '12', | +| | 'compaction.twcs.inactive_window.l1_merge_trigger' = '10', | +| | 'compaction.twcs.inactive_window.trigger_file_num' = '6', | +| | 'compaction.twcs.max_output_file_size' = '512MB', | +| | 'compaction.twcs.time_window' = '1d', | +| | 'compaction.twcs.trigger_file_num' = '8', | +| | 'compaction.type' = 'twcs' | +| | ) | ++----------------+--------------------------------------------------------------+ -- SQLNESS ARG restart=true SHOW CREATE DATABASE alter_database; -+----------------+-----------------------------------------------------+ -| Database | Create Database | -+----------------+-----------------------------------------------------+ -| alter_database | CREATE DATABASE IF NOT EXISTS alter_database | -| | WITH( | -| | 'compaction.twcs.max_output_file_size' = '512MB', | -| | 'compaction.twcs.time_window' = '1d', | -| | 'compaction.twcs.trigger_file_num' = '8', | -| | 'compaction.type' = 'twcs' | -| | ) | -+----------------+-----------------------------------------------------+ ++----------------+--------------------------------------------------------------+ +| Database | Create Database | ++----------------+--------------------------------------------------------------+ +| alter_database | CREATE DATABASE IF NOT EXISTS alter_database | +| | WITH( | +| | 'compaction.twcs.active_window.l1_merge_trigger' = '12', | +| | 'compaction.twcs.inactive_window.l1_merge_trigger' = '10', | +| | 'compaction.twcs.inactive_window.trigger_file_num' = '6', | +| | 'compaction.twcs.max_output_file_size' = '512MB', | +| | 'compaction.twcs.time_window' = '1d', | +| | 'compaction.twcs.trigger_file_num' = '8', | +| | 'compaction.type' = 'twcs' | +| | ) | ++----------------+--------------------------------------------------------------+ ALTER DATABASE alter_database UNSET 'compaction.twcs.trigger_file_num'; Affected Rows: 0 +ALTER DATABASE alter_database UNSET 'compaction.twcs.active_window.l1_merge_trigger'; + +Affected Rows: 0 + +ALTER DATABASE alter_database UNSET 'compaction.twcs.inactive_window.trigger_file_num'; + +Affected Rows: 0 + +ALTER DATABASE alter_database UNSET 'compaction.twcs.inactive_window.l1_merge_trigger'; + +Affected Rows: 0 + SHOW CREATE DATABASE alter_database; +----------------+-----------------------------------------------------+ diff --git a/tests/cases/standalone/common/alter/alter_database.sql b/tests/cases/standalone/common/alter/alter_database.sql index 33b309153e..42881eba15 100644 --- a/tests/cases/standalone/common/alter/alter_database.sql +++ b/tests/cases/standalone/common/alter/alter_database.sql @@ -42,6 +42,21 @@ SHOW CREATE DATABASE alter_database; ALTER DATABASE alter_database SET 'compaction.twcs.trigger_file_num'='8'; +ALTER DATABASE alter_database SET 'compaction.twcs.active_window.trigger_file_num'='16'; + +ALTER DATABASE alter_database SET 'compaction.twcs.active_window.l1_merge_trigger'='12'; + +ALTER DATABASE alter_database SET 'compaction.twcs.inactive_window.trigger_file_num'='6'; + +ALTER DATABASE alter_database SET 'compaction.twcs.inactive_window.l1_merge_trigger'='10'; + +SHOW CREATE DATABASE alter_database; + +-- SQLNESS ARG restart=true +SHOW CREATE DATABASE alter_database; + +ALTER DATABASE alter_database SET 'compaction.twcs.trigger_file_num'='8'; + ALTER DATABASE alter_database SET 'compaction.twcs.max_output_file_size'='512MB'; SHOW CREATE DATABASE alter_database; @@ -55,6 +70,12 @@ SHOW CREATE DATABASE alter_database; ALTER DATABASE alter_database UNSET 'compaction.twcs.trigger_file_num'; +ALTER DATABASE alter_database UNSET 'compaction.twcs.active_window.l1_merge_trigger'; + +ALTER DATABASE alter_database UNSET 'compaction.twcs.inactive_window.trigger_file_num'; + +ALTER DATABASE alter_database UNSET 'compaction.twcs.inactive_window.l1_merge_trigger'; + SHOW CREATE DATABASE alter_database; ALTER DATABASE alter_database UNSET 'compaction.twcs.time_window'; diff --git a/tests/cases/standalone/common/alter/alter_table_options.result b/tests/cases/standalone/common/alter/alter_table_options.result index b169562c89..85e64db429 100644 --- a/tests/cases/standalone/common/alter/alter_table_options.result +++ b/tests/cases/standalone/common/alter/alter_table_options.result @@ -234,6 +234,111 @@ SHOW CREATE TABLE ato; | | ) | +-------+-----------------------------------------------------+ +ALTER TABLE ato SET 'compaction.twcs.trigger_file_num'='4'; + +Affected Rows: 0 + +ALTER TABLE ato SET 'compaction.twcs.active_window.trigger_file_num'='1'; + +Affected Rows: 0 + +ALTER TABLE ato SET 'compaction.twcs.active_window.l1_merge_trigger'='8'; + +Affected Rows: 0 + +ALTER TABLE ato SET 'compaction.twcs.inactive_window.trigger_file_num'='1'; + +Affected Rows: 0 + +ALTER TABLE ato SET 'compaction.twcs.inactive_window.l1_merge_trigger'='12'; + +Affected Rows: 0 + +SHOW CREATE TABLE ato; + ++-------+--------------------------------------------------------------+ +| Table | Create Table | ++-------+--------------------------------------------------------------+ +| ato | CREATE TABLE IF NOT EXISTS "ato" ( | +| | "i" INT NULL, | +| | "j" TIMESTAMP(3) NOT NULL, | +| | TIME INDEX ("j"), | +| | PRIMARY KEY ("i") | +| | ) | +| | | +| | ENGINE=mito | +| | WITH( | +| | 'compaction.twcs.active_window.l1_merge_trigger' = '8', | +| | 'compaction.twcs.inactive_window.l1_merge_trigger' = '12', | +| | 'compaction.twcs.inactive_window.trigger_file_num' = '1', | +| | 'compaction.twcs.max_output_file_size' = '500MB', | +| | 'compaction.twcs.trigger_file_num' = '1', | +| | 'compaction.type' = 'twcs', | +| | ttl = '1s' | +| | ) | ++-------+--------------------------------------------------------------+ + +-- SQLNESS ARG restart=true +SHOW CREATE TABLE ato; + ++-------+--------------------------------------------------------------+ +| Table | Create Table | ++-------+--------------------------------------------------------------+ +| ato | CREATE TABLE IF NOT EXISTS "ato" ( | +| | "i" INT NULL, | +| | "j" TIMESTAMP(3) NOT NULL, | +| | TIME INDEX ("j"), | +| | PRIMARY KEY ("i") | +| | ) | +| | | +| | ENGINE=mito | +| | WITH( | +| | 'compaction.twcs.active_window.l1_merge_trigger' = '8', | +| | 'compaction.twcs.inactive_window.l1_merge_trigger' = '12', | +| | 'compaction.twcs.inactive_window.trigger_file_num' = '1', | +| | 'compaction.twcs.max_output_file_size' = '500MB', | +| | 'compaction.twcs.trigger_file_num' = '1', | +| | 'compaction.type' = 'twcs', | +| | ttl = '1s' | +| | ) | ++-------+--------------------------------------------------------------+ + +ALTER TABLE ato UNSET 'compaction.twcs.active_window.trigger_file_num'; + +Affected Rows: 0 + +ALTER TABLE ato UNSET 'compaction.twcs.active_window.l1_merge_trigger'; + +Affected Rows: 0 + +ALTER TABLE ato UNSET 'compaction.twcs.inactive_window.trigger_file_num'; + +Affected Rows: 0 + +ALTER TABLE ato UNSET 'compaction.twcs.inactive_window.l1_merge_trigger'; + +Affected Rows: 0 + +SHOW CREATE TABLE ato; + ++-------+-----------------------------------------------------+ +| Table | Create Table | ++-------+-----------------------------------------------------+ +| ato | CREATE TABLE IF NOT EXISTS "ato" ( | +| | "i" INT NULL, | +| | "j" TIMESTAMP(3) NOT NULL, | +| | TIME INDEX ("j"), | +| | PRIMARY KEY ("i") | +| | ) | +| | | +| | ENGINE=mito | +| | WITH( | +| | 'compaction.twcs.max_output_file_size' = '500MB', | +| | 'compaction.type' = 'twcs', | +| | ttl = '1s' | +| | ) | ++-------+-----------------------------------------------------+ + -- SQLNESS ARG restart=true SHOW CREATE TABLE ato; diff --git a/tests/cases/standalone/common/alter/alter_table_options.sql b/tests/cases/standalone/common/alter/alter_table_options.sql index c3f67faeee..dd32ee0d4c 100644 --- a/tests/cases/standalone/common/alter/alter_table_options.sql +++ b/tests/cases/standalone/common/alter/alter_table_options.sql @@ -50,6 +50,31 @@ ALTER TABLE ato SET 'compaction.twcs.trigger_file_num'=''; SHOW CREATE TABLE ato; +ALTER TABLE ato SET 'compaction.twcs.trigger_file_num'='4'; + +ALTER TABLE ato SET 'compaction.twcs.active_window.trigger_file_num'='1'; + +ALTER TABLE ato SET 'compaction.twcs.active_window.l1_merge_trigger'='8'; + +ALTER TABLE ato SET 'compaction.twcs.inactive_window.trigger_file_num'='1'; + +ALTER TABLE ato SET 'compaction.twcs.inactive_window.l1_merge_trigger'='12'; + +SHOW CREATE TABLE ato; + +-- SQLNESS ARG restart=true +SHOW CREATE TABLE ato; + +ALTER TABLE ato UNSET 'compaction.twcs.active_window.trigger_file_num'; + +ALTER TABLE ato UNSET 'compaction.twcs.active_window.l1_merge_trigger'; + +ALTER TABLE ato UNSET 'compaction.twcs.inactive_window.trigger_file_num'; + +ALTER TABLE ato UNSET 'compaction.twcs.inactive_window.l1_merge_trigger'; + +SHOW CREATE TABLE ato; + -- SQLNESS ARG restart=true SHOW CREATE TABLE ato; diff --git a/tests/cases/standalone/common/create/create_with_options.result b/tests/cases/standalone/common/create/create_with_options.result index dff8ae377f..1101e6ca41 100644 --- a/tests/cases/standalone/common/create/create_with_options.result +++ b/tests/cases/standalone/common/create/create_with_options.result @@ -81,6 +81,50 @@ drop table test_mito_options; Affected Rows: 0 +create table if not exists test_window_compaction_options( + host string, + ts timestamp, + TIME INDEX (ts), + PRIMARY KEY(host) +) +engine=mito +with( + 'compaction.type'='twcs', + 'compaction.twcs.active_window.trigger_file_num'='4', + 'compaction.twcs.active_window.l1_merge_trigger'='8', + 'compaction.twcs.inactive_window.trigger_file_num'='3', + 'compaction.twcs.inactive_window.l1_merge_trigger'='12' +); + +Affected Rows: 0 + +show create table test_window_compaction_options; + ++--------------------------------+---------------------------------------------------------------+ +| Table | Create Table | ++--------------------------------+---------------------------------------------------------------+ +| test_window_compaction_options | CREATE TABLE IF NOT EXISTS "test_window_compaction_options" ( | +| | "host" STRING NULL, | +| | "ts" TIMESTAMP(3) NOT NULL, | +| | TIME INDEX ("ts"), | +| | PRIMARY KEY ("host") | +| | ) | +| | | +| | ENGINE=mito | +| | WITH( | +| | 'compaction.override' = 'true', | +| | 'compaction.twcs.active_window.l1_merge_trigger' = '8', | +| | 'compaction.twcs.inactive_window.l1_merge_trigger' = '12', | +| | 'compaction.twcs.inactive_window.trigger_file_num' = '3', | +| | 'compaction.twcs.trigger_file_num' = '4', | +| | 'compaction.type' = 'twcs' | +| | ) | ++--------------------------------+---------------------------------------------------------------+ + +drop table test_window_compaction_options; + +Affected Rows: 0 + create table if not exists test_compaction_override_without_type( host string, ts timestamp, @@ -109,3 +153,24 @@ with('compaction.type'='twcs', 'compaction.twcs.trigger_file_num'='8d'); Error: 1004(InvalidArguments), Invalid options: invalid digit found in string +create table conflicting_twcs_trigger_aliases( + host string, + ts timestamp, + TIME INDEX (ts), + PRIMARY KEY(host) +) +engine=mito +with( + 'compaction.twcs.trigger_file_num'='4', + 'compaction.twcs.active_window.trigger_file_num'='8' +); + +Error: 1004(InvalidArguments), Conflicting table options: compaction.twcs.trigger_file_num=4 and compaction.twcs.active_window.trigger_file_num=8 + +create database conflicting_twcs_trigger_aliases with( + 'compaction.twcs.trigger_file_num'='4', + 'compaction.twcs.active_window.trigger_file_num'='8' +); + +Error: 1004(InvalidArguments), Conflicting schema options: compaction.twcs.trigger_file_num=4 and compaction.twcs.active_window.trigger_file_num=8 + diff --git a/tests/cases/standalone/common/create/create_with_options.sql b/tests/cases/standalone/common/create/create_with_options.sql index 071f103ab6..4370da03bf 100644 --- a/tests/cases/standalone/common/create/create_with_options.sql +++ b/tests/cases/standalone/common/create/create_with_options.sql @@ -67,6 +67,25 @@ with( drop table test_mito_options; +create table if not exists test_window_compaction_options( + host string, + ts timestamp, + TIME INDEX (ts), + PRIMARY KEY(host) +) +engine=mito +with( + 'compaction.type'='twcs', + 'compaction.twcs.active_window.trigger_file_num'='4', + 'compaction.twcs.active_window.l1_merge_trigger'='8', + 'compaction.twcs.inactive_window.trigger_file_num'='3', + 'compaction.twcs.inactive_window.l1_merge_trigger'='12' +); + +show create table test_window_compaction_options; + +drop table test_window_compaction_options; + create table if not exists test_compaction_override_without_type( host string, ts timestamp, @@ -88,3 +107,20 @@ create table if not exists invalid_compaction( ) engine=mito with('compaction.type'='twcs', 'compaction.twcs.trigger_file_num'='8d'); + +create table conflicting_twcs_trigger_aliases( + host string, + ts timestamp, + TIME INDEX (ts), + PRIMARY KEY(host) +) +engine=mito +with( + 'compaction.twcs.trigger_file_num'='4', + 'compaction.twcs.active_window.trigger_file_num'='8' +); + +create database conflicting_twcs_trigger_aliases with( + 'compaction.twcs.trigger_file_num'='4', + 'compaction.twcs.active_window.trigger_file_num'='8' +); diff --git a/tests/compatibility/cases/twcs_active_window_options/case.toml b/tests/compatibility/cases/twcs_active_window_options/case.toml new file mode 100644 index 0000000000..1c9b0233c5 --- /dev/null +++ b/tests/compatibility/cases/twcs_active_window_options/case.toml @@ -0,0 +1,8 @@ +name = "twcs_active_window_options" +reason = "Verify current binaries can reopen tables whose TWCS trigger option was written by old binaries." +introduced_by = "feat: introduce compaction active window" +topologies = ["distributed", "standalone"] +from_range = ["=v1.1.4"] +to_range = [">=v1.3.0"] +features = ["table", "metadata", "compaction", "twcs"] +owner = "storage" diff --git a/tests/compatibility/cases/twcs_active_window_options/setup.sql b/tests/compatibility/cases/twcs_active_window_options/setup.sql new file mode 100644 index 0000000000..6a08654945 --- /dev/null +++ b/tests/compatibility/cases/twcs_active_window_options/setup.sql @@ -0,0 +1,16 @@ +CREATE TABLE t_twcs_active_window_options ( + host STRING, + ts TIMESTAMP TIME INDEX, + cpu DOUBLE, + PRIMARY KEY(host) +) ENGINE=mito +WITH( + 'compaction.type' = 'twcs', + 'compaction.twcs.trigger_file_num' = '3' +); + +INSERT INTO t_twcs_active_window_options VALUES + ('host1', '2026-08-01 00:00:00+0000', 1.0), + ('host2', '2026-08-01 00:01:00+0000', 2.0); + +ADMIN FLUSH_TABLE('t_twcs_active_window_options'); diff --git a/tests/compatibility/cases/twcs_active_window_options/verify.result b/tests/compatibility/cases/twcs_active_window_options/verify.result new file mode 100644 index 0000000000..a1c26f95f8 --- /dev/null +++ b/tests/compatibility/cases/twcs_active_window_options/verify.result @@ -0,0 +1,53 @@ +SELECT host, ts, cpu +FROM t_twcs_active_window_options +ORDER BY host, ts; + ++-------+---------------------+-----+ +| host | ts | cpu | ++-------+---------------------+-----+ +| host1 | 2026-08-01T00:00:00 | 1.0 | +| host2 | 2026-08-01T00:01:00 | 2.0 | ++-------+---------------------+-----+ + +SHOW CREATE TABLE t_twcs_active_window_options; + ++------------------------------+-------------------------------------------------------------+ +| Table | Create Table | ++------------------------------+-------------------------------------------------------------+ +| t_twcs_active_window_options | CREATE TABLE IF NOT EXISTS "t_twcs_active_window_options" ( | +| | "host" STRING NULL, | +| | "ts" TIMESTAMP(3) NOT NULL, | +| | "cpu" DOUBLE NULL, | +| | TIME INDEX ("ts"), | +| | PRIMARY KEY ("host") | +| | ) | +| | | +| | ENGINE=mito | +| | WITH( | +| | 'compaction.override' = 'true', | +| | 'compaction.twcs.trigger_file_num' = '3', | +| | 'compaction.type' = 'twcs' | +| | ) | ++------------------------------+-------------------------------------------------------------+ + +ALTER TABLE t_twcs_active_window_options +SET 'compaction.twcs.active_window.trigger_file_num' = '4'; + +Affected Rows: 0 + +INSERT INTO t_twcs_active_window_options VALUES + ('host3', '2026-08-01 00:02:00+0000', 3.0); + +Affected Rows: 1 + +SELECT host, ts, cpu +FROM t_twcs_active_window_options +ORDER BY host, ts; + ++-------+---------------------+-----+ +| host | ts | cpu | ++-------+---------------------+-----+ +| host1 | 2026-08-01T00:00:00 | 1.0 | +| host2 | 2026-08-01T00:01:00 | 2.0 | +| host3 | 2026-08-01T00:02:00 | 3.0 | ++-------+---------------------+-----+ diff --git a/tests/compatibility/cases/twcs_active_window_options/verify.sql b/tests/compatibility/cases/twcs_active_window_options/verify.sql new file mode 100644 index 0000000000..af0dfa419c --- /dev/null +++ b/tests/compatibility/cases/twcs_active_window_options/verify.sql @@ -0,0 +1,15 @@ +SELECT host, ts, cpu +FROM t_twcs_active_window_options +ORDER BY host, ts; + +SHOW CREATE TABLE t_twcs_active_window_options; + +ALTER TABLE t_twcs_active_window_options +SET 'compaction.twcs.active_window.trigger_file_num' = '4'; + +INSERT INTO t_twcs_active_window_options VALUES + ('host3', '2026-08-01 00:02:00+0000', 3.0); + +SELECT host, ts, cpu +FROM t_twcs_active_window_options +ORDER BY host, ts;