From 4bd36cfd20c987b19247e6fe9ef0cad4438047cb Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" <6406592+v0y4g3r@users.noreply.github.com> Date: Tue, 13 Oct 2026 12:28:36 +0000 Subject: [PATCH] feat: support compaction options in HTTP hints (#9560) * feat: support compaction options in HTTP hints Signed-off-by: Lei, HUANG * refactor: share TWCS compaction option prefix Use the store-api constant for both HTTP hint inference and Mito region option validation so the two paths cannot drift. Signed-off-by: Lei, HUANG --------- Signed-off-by: Lei, HUANG --- src/mito2/src/region/options.rs | 3 +- src/operator/src/insert.rs | 70 ++++++++++++++++++--- src/servers/src/hint_headers.rs | 8 ++- src/store-api/src/mito_engine_options.rs | 78 +++++++++++++----------- 4 files changed, 109 insertions(+), 50 deletions(-) diff --git a/src/mito2/src/region/options.rs b/src/mito2/src/region/options.rs index 7e050fb4a76..42f4adcd04a 100644 --- a/src/mito2/src/region/options.rs +++ b/src/mito2/src/region/options.rs @@ -36,7 +36,7 @@ use store_api::metric_engine_consts::{ MEMTABLE_PARTITION_TREE_PRIMARY_KEY_ENCODING, PRIMARY_KEY_ENCODING, }; use store_api::mito_engine_options::{ - COMPACTION_OVERRIDE, FloatFieldEncoding, MAX_ROW_GROUP_ROW_COUNT_LIMIT, + COMPACTION_OVERRIDE, COMPACTION_TWCS_PREFIX, FloatFieldEncoding, MAX_ROW_GROUP_ROW_COUNT_LIMIT, }; use store_api::storage::{ColumnId, RegionId}; use strum::EnumString; @@ -47,7 +47,6 @@ use crate::sst::FormatType; use crate::sst::parquet::DEFAULT_ROW_GROUP_SIZE; const DEFAULT_INDEX_SEGMENT_ROW_COUNT: usize = 1024; -const COMPACTION_TWCS_PREFIX: &str = "compaction.twcs."; const MEMTABLE_PARTITION_TREE_PREFIX: &str = "memtable.partition_tree."; const MEMTABLE_BULK_PREFIX: &str = "memtable.bulk."; const MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD: &str = "memtable.bulk.encode_bytes_threshold"; diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs index 2c01a35b8d9..2b382dbdb5f 100644 --- a/src/operator/src/insert.rs +++ b/src/operator/src/insert.rs @@ -67,8 +67,8 @@ use store_api::metric_engine_consts::{ LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY, }; use store_api::mito_engine_options::{ - APPEND_MODE_KEY, COMPACTION_TYPE, COMPACTION_TYPE_TWCS, MERGE_MODE_KEY, TTL_KEY, - TWCS_TIME_WINDOW, + APPEND_MODE_KEY, COMPACTION_OPTION_KEYS, COMPACTION_TWCS_PREFIX, COMPACTION_TYPE, + COMPACTION_TYPE_TWCS, MERGE_MODE_KEY, TTL_KEY, }; use store_api::storage::{RegionId, TableId}; use table::TableRef; @@ -1856,6 +1856,21 @@ pub fn fill_table_options_for_create( } } + if !matches!(create_type, AutoCreateTableType::Logical(_)) { + let mut has_twcs_option = false; + for key in COMPACTION_OPTION_KEYS { + if let Some(value) = ctx.extension(key) { + has_twcs_option |= key.starts_with(COMPACTION_TWCS_PREFIX); + table_options.insert(key.to_string(), value.to_string()); + } + } + if has_twcs_option { + table_options + .entry(COMPACTION_TYPE.to_string()) + .or_insert_with(|| COMPACTION_TYPE_TWCS.to_string()); + } + } + // Semantic keys use their own vocabulary instead of the fixed option list. for (key, value) in ctx.extensions() { if is_semantic_option_key(&key) && validate_semantic_option(&key, &value) { @@ -1877,14 +1892,6 @@ pub fn fill_table_options_for_create( if let Some(merge_mode) = ctx.extension(MERGE_MODE_KEY) { table_options.insert(MERGE_MODE_KEY.to_string(), merge_mode.to_string()); } - if let Some(time_window) = ctx.extension(TWCS_TIME_WINDOW) { - table_options.insert(TWCS_TIME_WINDOW.to_string(), time_window.to_string()); - // We need to set the compaction type explicitly. - table_options.insert( - COMPACTION_TYPE.to_string(), - COMPACTION_TYPE_TWCS.to_string(), - ); - } } // Set append_mode to true for log table. // because log tables should keep rows with the same ts and tags. @@ -3051,6 +3058,49 @@ mod tests { assert_eq!(requests, original); } + #[test] + fn test_trace_create_options_copy_compaction_hints() { + let mut ctx = QueryContext::with(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME); + ctx.set_extension("compaction.twcs.time_window", "1h"); + ctx.set_extension("compaction.twcs.max_output_file_size", "256MB"); + ctx.set_extension("compaction.twcs.remote_compaction", "true"); + ctx.set_extension("compaction.unknown", "ignored"); + let ctx = Arc::new(ctx); + let mut table_options = Default::default(); + + fill_table_options_for_create( + &mut table_options, + &AutoCreateTableType::Trace { + alter_existing: true, + }, + &ctx, + ); + + assert_eq!( + Some("twcs"), + table_options.get(COMPACTION_TYPE).map(String::as_str) + ); + assert_eq!( + Some("1h"), + table_options + .get("compaction.twcs.time_window") + .map(String::as_str) + ); + assert_eq!( + Some("256MB"), + table_options + .get("compaction.twcs.max_output_file_size") + .map(String::as_str) + ); + assert_eq!( + Some("true"), + table_options + .get("compaction.twcs.remote_compaction") + .map(String::as_str) + ); + assert!(!table_options.contains_key("compaction.unknown")); + } + #[test] fn test_skip_wal_does_not_change_table_options() { check_skip_wal_does_not_change_table_options(false); diff --git a/src/servers/src/hint_headers.rs b/src/servers/src/hint_headers.rs index 3d13dd9bf5e..4d37447e01d 100644 --- a/src/servers/src/hint_headers.rs +++ b/src/servers/src/hint_headers.rs @@ -145,12 +145,12 @@ mod tests { headers.insert( "x-greptime-hints", HeaderValue::from_static(" auto_create_table=true, ttl =3600d, append_mode=true , merge_mode=false , physical_table= table1,\ - read_preference=leader"), + read_preference=leader, compaction.twcs.time_window=1h"), ); let hints = extract_hints(&headers); - assert_eq!(hints.len(), 6); + assert_eq!(hints.len(), 7); assert_eq!( hints[0], ("auto_create_table".to_string(), "true".to_string()) @@ -166,6 +166,10 @@ mod tests { hints[5], ("read_preference".to_string(), "leader".to_string()) ); + assert_eq!( + hints[6], + ("compaction.twcs.time_window".to_string(), "1h".to_string()) + ); } #[test] diff --git a/src/store-api/src/mito_engine_options.rs b/src/store-api/src/mito_engine_options.rs index 07cfe6a6516..457b33665c3 100644 --- a/src/store-api/src/mito_engine_options.rs +++ b/src/store-api/src/mito_engine_options.rs @@ -35,6 +35,8 @@ pub const COMPACTION_TYPE: &str = "compaction.type"; pub const COMPACTION_OVERRIDE: &str = "compaction.override"; /// TWCS compaction strategy. pub const COMPACTION_TYPE_TWCS: &str = "twcs"; +/// Prefix for TWCS compaction options. +pub const COMPACTION_TWCS_PREFIX: &str = "compaction.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. @@ -57,6 +59,20 @@ pub const TWCS_TIME_WINDOW: &str = "compaction.twcs.time_window"; pub const TWCS_REMOTE_COMPACTION: &str = "compaction.twcs.remote_compaction"; /// Option key for twcs fallback to local. pub const TWCS_FALLBACK_TO_LOCAL: &str = "compaction.twcs.fallback_to_local"; +/// Table options that configure compaction. +pub const COMPACTION_OPTION_KEYS: [&str; 11] = [ + 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, + TWCS_FALLBACK_TO_LOCAL, +]; /// Option key for memtable type. pub const MEMTABLE_TYPE: &str = "memtable.type"; /// Option key for bulk memtable merge threshold. @@ -145,42 +161,32 @@ pub fn normalize_twcs_trigger_options( /// Returns true if the `key` is a valid option key for the mito engine. pub fn is_mito_engine_option_key(key: &str) -> bool { - [ - "ttl", - WRITE_BUFFER_SIZE_KEY, - AUTO_FLUSH_INTERVAL_KEY, - 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, - TWCS_FALLBACK_TO_LOCAL, - "storage", - "index.inverted_index.ignore_column_ids", - "index.inverted_index.segment_row_count", - WAL_OPTIONS_KEY, - MEMTABLE_TYPE, - MEMTABLE_BULK_MERGE_THRESHOLD, - MEMTABLE_BULK_ENCODE_ROW_THRESHOLD, - MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD, - MEMTABLE_BULK_MAX_MERGE_GROUPS, - MEMTABLE_PARTITION_TREE_INDEX_MAX_KEYS_PER_SHARD, - MEMTABLE_PARTITION_TREE_DATA_FREEZE_THRESHOLD, - MEMTABLE_PARTITION_TREE_FORK_DICTIONARY_BYTES, - // We don't allow to create a mito table with sparse primary key encoding directly. - APPEND_MODE_KEY, - MERGE_MODE_KEY, - SST_FORMAT_KEY, - MAX_ROW_GROUP_ROW_COUNT, - PRESERVE_ROW_SEQUENCE, - EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, - ] - .contains(&key) + COMPACTION_OPTION_KEYS.contains(&key) + || [ + "ttl", + WRITE_BUFFER_SIZE_KEY, + AUTO_FLUSH_INTERVAL_KEY, + "storage", + "index.inverted_index.ignore_column_ids", + "index.inverted_index.segment_row_count", + WAL_OPTIONS_KEY, + MEMTABLE_TYPE, + MEMTABLE_BULK_MERGE_THRESHOLD, + MEMTABLE_BULK_ENCODE_ROW_THRESHOLD, + MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD, + MEMTABLE_BULK_MAX_MERGE_GROUPS, + MEMTABLE_PARTITION_TREE_INDEX_MAX_KEYS_PER_SHARD, + MEMTABLE_PARTITION_TREE_DATA_FREEZE_THRESHOLD, + MEMTABLE_PARTITION_TREE_FORK_DICTIONARY_BYTES, + // We don't allow to create a mito table with sparse primary key encoding directly. + APPEND_MODE_KEY, + MERGE_MODE_KEY, + SST_FORMAT_KEY, + MAX_ROW_GROUP_ROW_COUNT, + PRESERVE_ROW_SEQUENCE, + EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, + ] + .contains(&key) } #[cfg(test)]