feat(mito2): introduce TWCS active window compaction (#9011)

* feat(mito2): support independent TWCS trigger_file_num for active and inactive windows

Split the single TWCS trigger_file_num into per-window-state thresholds:
the active window keeps the existing trigger (default 4, legacy
compaction.twcs.trigger_file_num stays a compatible alias), while
inactive windows use a new trigger (default 2). Inactive windows
additionally fall back from balanced L0-only/L1-only candidates to a
progress-making unbalanced mixed candidate so historical windows can
converge; the active window retains the row/byte balance guards.

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

* fix(mito2): bound inactive TWCS window convergence by rewrite budget

Inactive windows that cannot compact within one level previously either
stayed stuck (a threshold-qualified but unbalanced level returned no
candidate without trying any fallback) or fell back to a mixed merge
with no balance checks at all, which could rewrite a huge compacted file
to absorb tiny fresh files.

Inactive windows now converge progressively: threshold-qualified
balanced picks, sub-threshold balanced single-level picks, a mixed merge
whose total rewrite must fit in the output file budget, and finally an
L0-only merge without balance checks. Windows that qualify for none of
these are left uncompacted, bounding write amplification.

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

* test(mito2): cover TWCS window trigger options in alter_table_options sqlness case

Exercise SET/UNSET of compaction.twcs.active_window.trigger_file_num and
compaction.twcs.inactive_window.trigger_file_num end to end, including
that setting the canonical active key removes the legacy
compaction.twcs.trigger_file_num alias from the table options.

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

* fix(mito2): derive TWCS active window from the max-sequence file

The active window was determined by the max event-time window among
level-0 files. Between an L0 compaction removing its inputs and the next
flush landing, level 0 is empty, so the active window transiently became
None and every window fell back to the inactive rules - triggering
full-window convergence merges during ongoing ingestion whose outputs
are then superseded by new data.

Flush and compaction outputs both inherit the max input sequence, so the
file with the highest sequence across all levels always tracks the most
recent write. Use its window as the active window, falling back to the
previous L0-based rule when no file carries a sequence (legacy files).

The new helper deliberately computes window keys with the
assign_to_windows convention (truncate to seconds, then align up),
because the result is compared against window keys produced there; the
older ceil-based helper is kept unchanged for the legacy fallback path.

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

* feat(mito2): add active-window L1 compaction safety trigger

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

* test(compat): cover TWCS active window options

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

* fix(mito2): align TWCS window trigger validation

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

* fix(mito2): resolve database TWCS trigger aliases

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

* fix(mito2): prioritize newer compaction windows

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

* fix(mito2): repick serial compaction outputs

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

* feat(mito2): configure inactive-window L1 trigger

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

* feat(mito2): prioritize TWCS compaction candidates

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

* fix(meta): preserve TWCS trigger downgrade compatibility

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

* fix(sql): distinguish invalid database option values

Separate database option key and value validation so recognized keys report the invalid value and its constraint. Add parser coverage for invalid, valid, and unknown options.

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

* docs(meta): explain TWCS legacy key compatibility

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

* refactor(mito2): clarify active window trigger field

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

* fix(options): normalize TWCS trigger aliases

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

* fix(mito2): ignore ineligible files for active window

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

* fix(mito2): mark explicit TWCS options as overrides

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

* fix(mito2): normalize zero compaction output size

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

* fix(options): validate database TWCS trigger values

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

* style(store-api): collapse TWCS alias conflict condition

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

---------

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

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)
+            );
+        }
+    }
 }

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 { .. } => {

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<String, String>> for SchemaNameValue {
     type Error = Error;
 
     fn try_from(value: &HashMap<String, String>) -> std::result::Result<Self, Self::Error> {
+        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

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<PbOption> 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();

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<String, String> = schema_opts
+            let mut map: HashMap<String, String> = 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 {

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,

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)

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();

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, &region_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;

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);
 }
 

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,

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<Duration>,
@@ -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,

src/mito2/src/test_util/scheduler_util.rs

@@ -92,6 +92,14 @@ impl SchedulerEnv {
     pub(crate) fn mock_compaction_scheduler(
         &self,
         request_sender: Sender<WorkerRequestWithTime>,
+    ) -> CompactionScheduler {
+        self.mock_compaction_scheduler_with_config(request_sender, MitoConfig::default())
+    }
+
+    pub(crate) fn mock_compaction_scheduler_with_config(
+        &self,
+        request_sender: Sender<WorkerRequestWithTime>,
+        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)),

src/mito2/src/worker/handle_alter.rs

@@ -246,6 +246,9 @@ impl<S: LogStore> RegionWorkerLoop<S> {
                         &value,
                         region.region_id,
                     )?;
+                    if !value.is_empty() {
+                        current_options.compaction_override = true;
+                    }
                 }
                 SetRegionOption::Format(format_str) => {
                     let new_format = format_str.parse::<FormatType>().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();

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<S> RegionWorkerLoop<S> {
     pub(crate) async fn handle_compaction_pick_finished(
         &mut self,
@@ -131,11 +135,10 @@ impl<S> RegionWorkerLoop<S> {
             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<S> RegionWorkerLoop<S> {
         }
     }
 }
+
+#[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}"
+            );
+        }
+    }
+}

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 { .. }

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::<Result<HashMap<String, OptionValue>>>()?;
 
-        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;

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<String, String>,
+) -> 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);
+    }
 }

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 {

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 { .. }

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()

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<HashSet<&str>> = 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::<usize>().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<TableOptions> {
         let mut options = TableOptions::default();
 
-        let kvs: HashMap<String, String> = iter
+        let mut kvs: HashMap<String, String> = 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 {

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;
 
 +----------------+-----------------------------------------------------+

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';

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;
 

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;
 

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
+

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'
+);

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"

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');

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 |
++-------+---------------------+-----+

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;