From f9c2a9427af1e2825853ecbeb58d2a21af33a252 Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Tue, 22 Sep 2026 19:29:05 +0800 Subject: [PATCH] fix(flow): validate retained source windows before recovery Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/flow/src/batching_mode/task.rs | 127 +++++++++++++++++- src/flow/src/batching_mode/task/test.rs | 165 +++++++++++++++++++++++- 2 files changed, 290 insertions(+), 2 deletions(-) diff --git a/src/flow/src/batching_mode/task.rs b/src/flow/src/batching_mode/task.rs index 4fe4d0dde17..e2b903d6b9c 100644 --- a/src/flow/src/batching_mode/task.rs +++ b/src/flow/src/batching_mode/task.rs @@ -18,13 +18,15 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use api::v1::{CreateTableExpr, TableName}; use catalog::CatalogManagerRef; +use catalog::kvbackend::KvBackendCatalogManager; use client::OutputWithMetrics; use common_error::ext::BoxedError; +use common_meta::key::schema_name::SchemaNameKey; use common_query::OutputData; use common_query::logical_plan::breakup_insert_plan; use common_telemetry::tracing::warn; use common_telemetry::{debug, info}; -use common_time::Timestamp; +use common_time::{TimeToLive, Timestamp}; use datafusion::datasource::DefaultTableSource; use datafusion::sql::unparser::expr_to_sql; use datafusion_common::tree_node::{Transformed, TreeNode}; @@ -583,6 +585,129 @@ impl BatchingTask { .map(Some) } + /// Verifies that recovery can still read all source data required by its retained scope. + pub async fn validate_recovery_retention( + &self, + retention_lower: Option, + windows: &[(Timestamp, Timestamp)], + ) -> Result<(), Error> { + for name in &self.config.source_table_names { + let table = self + .config + .catalog_manager + .table(&name[0], &name[1], &name[2], None) + .await + .map_err(BoxedError::new) + .context(ExternalSnafu)? + .context(UnexpectedSnafu { + reason: format!( + "Flow {} source table {} is unavailable for recovery retention validation", + self.config.flow_id, + name.join(".") + ), + })?; + let ttl = if let Some(ttl) = table.table_info().meta.options.ttl { + ttl + } else { + let manager = self + .config + .catalog_manager + .as_any() + .downcast_ref::() + .context(UnexpectedSnafu { + reason: format!( + "Flow {} cannot resolve inherited TTL for source table {} during recovery", + self.config.flow_id, + name.join(".") + ), + })?; + manager + .table_metadata_manager_ref() + .schema_manager() + .get(SchemaNameKey::new(&name[0], &name[1])) + .await + .map_err(BoxedError::new) + .context(ExternalSnafu)? + .context(UnexpectedSnafu { + reason: format!( + "Flow {} schema {}.{} is unavailable for recovery retention validation", + self.config.flow_id, name[0], name[1] + ), + })? + .ttl + .map(Into::into) + .unwrap_or(TimeToLive::Forever) + }; + + match ttl { + TimeToLive::Forever => {} + TimeToLive::Instant => { + return UnexpectedSnafu { + reason: format!( + "Flow {} source table {} has instant TTL and cannot be recovered", + self.config.flow_id, + name.join(".") + ), + } + .fail(); + } + TimeToLive::Duration(ttl) => { + let expire_after = self.config.expire_after.context(UnexpectedSnafu { + reason: format!( + "Flow {} source table {} has TTL {ttl:?}, but recovery has no expire_after", + self.config.flow_id, + name.join(".") + ), + })?; + let expire_after = u64::try_from(expire_after).map_err(|_| { + UnexpectedSnafu { + reason: format!( + "Flow {} has negative expire_after {expire_after} during recovery retention validation", + self.config.flow_id + ), + } + .build() + })?; + if ttl <= Duration::from_secs(expire_after) { + return UnexpectedSnafu { + reason: format!( + "Flow {} source table {} TTL {ttl:?} must exceed expire_after {expire_after}s for recovery", + self.config.flow_id, + name.join(".") + ), + } + .fail(); + } + let mut oldest = retention_lower.context(UnexpectedSnafu { + reason: format!( + "Flow {} source table {} has finite TTL {ttl:?}, but recovery has no retained lower bound", + self.config.flow_id, + name.join(".") + ), + })?; + for (start, _) in windows { + oldest = oldest.min(*start); + } + let cutoff = Timestamp::current_millis() + .sub_duration(ttl) + .map_err(BoxedError::new) + .context(ExternalSnafu)?; + if oldest <= cutoff { + return UnexpectedSnafu { + reason: format!( + "Flow {} recovery requires source table {} data from {oldest:?}, older than TTL {ttl:?} cutoff {cutoff:?}", + self.config.flow_id, + name.join(".") + ), + } + .fail(); + } + } + } + } + Ok(()) + } + /// Discovers the source time windows touched by the exact sequence range `(C, H]`. /// /// This compatibility wrapper captures the retention bound before discovery. diff --git a/src/flow/src/batching_mode/task/test.rs b/src/flow/src/batching_mode/task/test.rs index ea4d759effe..3ec8ac8bae6 100644 --- a/src/flow/src/batching_mode/task/test.rs +++ b/src/flow/src/batching_mode/task/test.rs @@ -17,6 +17,7 @@ use std::pin::Pin; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::task::{Context, Poll}; +use std::time::Duration; use catalog::memory::MemoryCatalogManager; use catalog::{DeregisterTableRequest, RegisterTableRequest}; @@ -28,7 +29,7 @@ use common_error::status_code::StatusCode; use common_query::Output; use common_recordbatch::RecordBatch; use common_recordbatch::adapter::{RecordBatchMetrics, RegionWatermarkEntry}; -use common_time::Timestamp; +use common_time::{TimeToLive, Timestamp}; use datatypes::data_type::ConcreteDataType as CDT; use datatypes::schema::{ColumnSchema, Schema}; use datatypes::vectors::{ @@ -819,6 +820,47 @@ fn register_auto_created_aggregate_sink(query_engine: &QueryEngineRef, table_nam memory_catalog.register_table_sync(request).unwrap(); } +async fn configure_source_ttl(query_engine: &QueryEngineRef, ttl: Option) { + let catalog_manager = query_engine.engine_state().catalog_manager(); + let source = catalog_manager + .table( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + "numbers_with_ts", + None, + ) + .await + .unwrap() + .unwrap(); + let mut info = (*source.table_info()).clone(); + info.meta.options.ttl = ttl; + let source = Arc::new(Table::new( + Arc::new(info), + FilterPushDownType::Unsupported, + source.data_source(), + )); + let memory_catalog = catalog_manager + .as_any() + .downcast_ref::() + .unwrap(); + memory_catalog + .deregister_table_sync(DeregisterTableRequest { + catalog: DEFAULT_CATALOG_NAME.to_string(), + schema: DEFAULT_SCHEMA_NAME.to_string(), + table_name: "numbers_with_ts".to_string(), + }) + .unwrap(); + memory_catalog + .register_table_sync(RegisterTableRequest { + catalog: DEFAULT_CATALOG_NAME.to_string(), + schema: DEFAULT_SCHEMA_NAME.to_string(), + table_name: "numbers_with_ts".to_string(), + table_id: source.table_info().table_id(), + table: source, + }) + .unwrap(); +} + async fn configure_source_capability( query_engine: &QueryEngineRef, engine: &str, @@ -875,6 +917,127 @@ async fn configure_source_capability( .unwrap(); } +#[tokio::test] +async fn test_validate_recovery_retention_explicit_source_ttls() { + struct Case { + name: &'static str, + ttl: Option, + expire_after: Option, + retention_age: Option, + window_age: Option, + should_pass: bool, + } + + let cases = [ + Case { + name: "finite TTL exceeding expire_after is admissible", + ttl: Some(TimeToLive::Duration(Duration::from_secs(2 * 60 * 60))), + expire_after: Some(60 * 60), + retention_age: Some(Duration::from_secs(10 * 60)), + window_age: None, + should_pass: true, + }, + Case { + name: "TTL equal to expire_after is rejected", + ttl: Some(TimeToLive::Duration(Duration::from_secs(60 * 60))), + expire_after: Some(60 * 60), + retention_age: Some(Duration::from_secs(10 * 60)), + window_age: None, + should_pass: false, + }, + Case { + name: "TTL shorter than expire_after is rejected", + ttl: Some(TimeToLive::Duration(Duration::from_secs(59 * 60))), + expire_after: Some(60 * 60), + retention_age: Some(Duration::from_secs(10 * 60)), + window_age: None, + should_pass: false, + }, + Case { + name: "aligned window older than TTL is rejected", + ttl: Some(TimeToLive::Duration(Duration::from_secs(2 * 60 * 60))), + expire_after: Some(60 * 60), + retention_age: Some(Duration::from_secs(10 * 60)), + window_age: Some(Duration::from_secs(3 * 60 * 60)), + should_pass: false, + }, + Case { + name: "old retention cutoff is rejected even without windows", + ttl: Some(TimeToLive::Duration(Duration::from_secs(2 * 60 * 60))), + expire_after: Some(60 * 60), + retention_age: Some(Duration::from_secs(3 * 60 * 60)), + window_age: None, + should_pass: false, + }, + Case { + name: "explicit forever is admissible", + ttl: Some(TimeToLive::Forever), + expire_after: None, + retention_age: None, + window_age: None, + should_pass: true, + }, + Case { + name: "instant TTL is rejected", + ttl: Some(TimeToLive::Instant), + expire_after: Some(60 * 60), + retention_age: Some(Duration::from_secs(10 * 60)), + window_age: None, + should_pass: false, + }, + ]; + + for case in cases { + let TestTaskParts { + mut task, + query_engine, + .. + } = new_test_task_engine_and_plan_with_query( + "SELECT number, ts FROM numbers_with_ts", + "sink", + ) + .await; + Arc::get_mut(&mut task.config).unwrap().expire_after = case.expire_after; + configure_source_ttl(&query_engine, case.ttl).await; + + let now = Timestamp::current_millis(); + let retention_lower = case.retention_age.map(|age| now.sub_duration(age).unwrap()); + let windows = case + .window_age + .map(|age| { + let start = now.sub_duration(age).unwrap(); + vec![(start, now)] + }) + .unwrap_or_default(); + assert_eq!( + task.validate_recovery_retention(retention_lower, &windows) + .await + .is_ok(), + case.should_pass, + "{}", + case.name + ); + } +} + +#[tokio::test] +async fn test_validate_recovery_retention_rejects_unknown_inherited_ttl() { + let TestTaskParts { + mut task, + query_engine, + .. + } = new_test_task_engine_and_plan_with_query("SELECT number, ts FROM numbers_with_ts", "sink") + .await; + Arc::get_mut(&mut task.config).unwrap().expire_after = Some(60 * 60); + configure_source_ttl(&query_engine, None).await; + + assert!( + task.validate_recovery_retention(Some(Timestamp::current_millis()), &[],) + .await + .is_err() + ); +} + fn dirty_marker() -> DirtyTimeWindows { let mut dirty = DirtyTimeWindows::default(); dirty.set_dirty();