diff --git a/src/catalog/src/system_schema/information_schema/flows.rs b/src/catalog/src/system_schema/information_schema/flows.rs index 8002b17888..12a8783b58 100644 --- a/src/catalog/src/system_schema/information_schema/flows.rs +++ b/src/catalog/src/system_schema/information_schema/flows.rs @@ -16,7 +16,7 @@ use std::sync::{Arc, Weak}; use common_catalog::consts::INFORMATION_SCHEMA_FLOW_TABLE_ID; use common_error::ext::BoxedError; -use common_meta::ddl::create_flow::FlowType; +use common_meta::ddl::create_flow::{FlowType, effective_eval_schedule_from_flow_info}; use common_meta::key::FlowId; use common_meta::key::flow::FlowMetadataManager; use common_meta::key::flow::flow_info::FlowInfoValue; @@ -168,6 +168,11 @@ impl InformationSchemaFlows { if_not_exists: true, expire_after: flow_info.expire_after(), eval_interval: flow_info.eval_interval(), + eval_offset: effective_eval_schedule_from_flow_info(flow_info) + .map_err(BoxedError::new) + .context(InternalSnafu)? + .map(|schedule| schedule.anchor_secs) + .filter(|anchor_secs| *anchor_secs != 0), comment, flow_options: sql::statements::OptionMap::from_filtered_string_map( flow_info.options(), @@ -464,3 +469,97 @@ impl DfPartitionStream for InformationSchemaFlows { )) } } + +#[cfg(test)] +mod tests { + use std::collections::{BTreeMap, HashMap}; + + use common_meta::key::flow::flow_info::{FlowMissedTickPolicy, FlowScheduleConfig, FlowStatus}; + use sql::parser::ParseOptions; + use table::table_name::TableName; + + use super::*; + + fn flow_info_for_show_create( + raw_sql: &str, + eval_interval_secs: Option, + anchor_secs: i64, + options: HashMap, + ) -> FlowInfoValue { + let created_time = chrono::DateTime::from_timestamp(1_700_000_000, 0).unwrap(); + FlowInfoValue { + source_table_ids: vec![], + all_source_table_names: vec![], + unresolved_source_table_names: vec![], + sink_table_name: TableName::new("greptime", "public", "sink"), + flownode_ids: BTreeMap::new(), + catalog_name: "greptime".to_string(), + query_context: None, + flow_name: "my_flow".to_string(), + raw_sql: raw_sql.to_string(), + expire_after: None, + eval_interval_secs, + comment: String::new(), + options, + status: FlowStatus::Active, + created_time, + updated_time: created_time, + eval_schedule: eval_interval_secs.map(|interval| FlowScheduleConfig { + anchor_secs, + start_secs: anchor_secs + interval, + missed_tick_policy: FlowMissedTickPolicy::BoundedCatchUp, + catchup_max_runs: 3, + catchup_max_lag_secs: 300, + }), + } + } + + #[test] + fn test_generate_show_create_flow_with_eval_offset() { + // `raw_sql` stores only the query part (the `AS` clause). + let raw_sql = "SELECT max(c1) FROM public.src"; + let flow_info = flow_info_for_show_create(raw_sql, Some(3600), 120, HashMap::new()); + let sql = InformationSchemaFlows::generate_show_create_flow(&flow_info).unwrap(); + assert!( + sql.contains("EVAL OFFSET '120 s'"), + "EVAL OFFSET must be emitted, got:\n{sql}" + ); + assert!( + !sql.contains("__greptime_internal_eval_offset_secs"), + "internal key must be absent, got:\n{sql}" + ); + + let stmts = ParserContext::create_with_dialect( + &sql, + &GreptimeDbDialect {}, + ParseOptions::default(), + ) + .unwrap(); + let Statement::CreateFlow(reparsed) = &stmts[0] else { + panic!("unexpected stmt: {:?}", stmts[0]); + }; + assert_eq!(reparsed.eval_interval, Some(3600)); + assert_eq!(reparsed.eval_offset, Some(120)); + } + + #[test] + fn test_generate_show_create_flow_omits_zero_offset() { + let raw_sql = "SELECT max(c1) FROM public.src"; + let flow_info = flow_info_for_show_create(raw_sql, Some(3600), 0, HashMap::new()); + let sql = InformationSchemaFlows::generate_show_create_flow(&flow_info).unwrap(); + assert!( + !sql.contains("EVAL OFFSET"), + "zero offset must be omitted, got:\n{sql}" + ); + let stmts = ParserContext::create_with_dialect( + &sql, + &GreptimeDbDialect {}, + ParseOptions::default(), + ) + .unwrap(); + let Statement::CreateFlow(reparsed) = &stmts[0] else { + panic!("unexpected stmt: {:?}", stmts[0]); + }; + assert_eq!(reparsed.eval_offset, None); + } +} diff --git a/src/common/meta/src/ddl/create_flow.rs b/src/common/meta/src/ddl/create_flow.rs index a2f306306d..2679e61e74 100644 --- a/src/common/meta/src/ddl/create_flow.rs +++ b/src/common/meta/src/ddl/create_flow.rs @@ -21,6 +21,7 @@ use api::v1::ExpireAfter; use api::v1::flow::flow_request::Body as PbFlowRequest; use api::v1::flow::{CreateRequest, FlowRequest, FlowRequestHeader}; use async_trait::async_trait; +use chrono::{DateTime, Utc}; use common_catalog::format_full_flow_name; use common_procedure::error::{FromJsonSnafu, ToJsonSnafu}; use common_procedure::{ @@ -32,7 +33,7 @@ use common_telemetry::tracing_context::TracingContext; use futures::future::join_all; use itertools::Itertools; use serde::{Deserialize, Serialize}; -use snafu::{ResultExt, ensure}; +use snafu::{OptionExt, ResultExt, ensure}; use strum::AsRefStr; use table::metadata::TableId; use table::table_name::TableName; @@ -222,7 +223,7 @@ impl CreateFlowProcedure { .prev_flow_info_value .as_ref() .map(|v| v.get_inner_ref()), - ); + )?; self.data.state = if self.data.is_pending() { self.data.peers.clear(); @@ -465,6 +466,14 @@ pub fn get_flow_type_from_options(flow_task: &CreateFlowTask) -> Result Result<()> { .fail(); } + for key in [INTERNAL_EVAL_OFFSET_KEY, INTERNAL_EVAL_SCHEDULE_KEY] { + if flow_task.flow_options.contains_key(key) { + return UnexpectedSnafu { + err_msg: format!("flow option '{key}' is reserved for internal use"), + } + .fail(); + } + } + + // EVAL OFFSET semantics: only legal with EVAL INTERVAL and must be in + // `[0, eval_interval_secs)`. Never modulo-normalized. + if let Some(offset_secs) = flow_task.eval_offset_secs { + let Some(eval_interval_secs) = flow_task.eval_interval_secs else { + return UnexpectedSnafu { + err_msg: "EVAL OFFSET requires EVAL INTERVAL to be specified".to_string(), + } + .fail(); + }; + if !(0..eval_interval_secs).contains(&offset_secs) { + return UnexpectedSnafu { + err_msg: format!( + "EVAL OFFSET must be in range [0, EVAL INTERVAL), got {offset_secs} seconds with EVAL INTERVAL {eval_interval_secs} seconds" + ), + } + .fail(); + } + } + for key in flow_task.flow_options.keys() { match key.as_str() { DEFER_ON_MISSING_SOURCE_KEY @@ -538,12 +575,16 @@ pub fn validate_flow_options(flow_task: &CreateFlowTask) -> Result<()> { /// Computes the ceiling of `time` to the next schedule boundary aligned with `anchor + k * interval`. /// All values are Unix timestamps in seconds. -fn ceil_to_boundary(time: i64, anchor: i64, interval: i64) -> i64 { +/// +/// Fallible: if the next boundary after `time` does not fit in `i64`, an +/// explicit error is returned instead of clamping to a non-phase value such as +/// `i64::MAX`. +pub(crate) fn ceil_to_boundary(time: i64, anchor: i64, interval: i64) -> Result { if interval <= 0 { - return time; + return Ok(time); } if time <= anchor { - return anchor; + return Ok(anchor); } let diff = i128::from(time) - i128::from(anchor); @@ -551,7 +592,41 @@ fn ceil_to_boundary(time: i64, anchor: i64, interval: i64) -> i64 { let k = (diff + interval - 1) / interval; let boundary = i128::from(anchor) + k * interval; - boundary.clamp(i128::from(i64::MIN), i128::from(i64::MAX)) as i64 + i64::try_from(boundary).map_err(|_| { + UnexpectedSnafu { + err_msg: format!( + "Cannot align time {time} to the next `anchor + k * interval` boundary (anchor={anchor}, interval={interval}): result {boundary} does not fit in i64" + ), + } + .build() + }) +} + +/// Rounds a `Utc` instant up to the next whole second (Unix seconds), with +/// nanosecond precision. +/// +/// `timestamp()` / `timestamp_millis()` floor the sub-second fraction; using +/// them unchanged could produce a `start_secs` in the past relative to the +/// exact prepare instant, which would make the very first evaluation due before +/// the flow finished being prepared. Uses checked arithmetic so an instant at +/// the very end of the `i64` second range yields an explicit error instead of +/// wrapping. +pub(crate) fn ceil_to_whole_sec(now: DateTime) -> Result { + ceil_whole_sec_from_parts(now.timestamp(), now.timestamp_subsec_nanos() != 0) +} + +/// Pure ceiling computation factored out of [`ceil_to_whole_sec`] so the +/// overflow path is testable: `chrono::DateTime` cannot represent +/// instants near `i64::MAX` seconds, but the arithmetic below guards it anyway. +pub(crate) fn ceil_whole_sec_from_parts(secs: i64, has_fraction: bool) -> Result { + if !has_fraction { + return Ok(secs); + } + secs.checked_add(1).context(error::UnexpectedSnafu { + err_msg: format!( + "Cannot round instant at second {secs} up to the next whole second: timestamp overflow" + ), + }) } /// Returns the effective typed schedule config for flow metadata. @@ -561,26 +636,31 @@ fn ceil_to_boundary(time: i64, anchor: i64, interval: i64) -> i64 { /// and defaults. This avoids recovery-time wall-clock drift. pub fn effective_eval_schedule_from_flow_info( flow_info: &FlowInfoValue, -) -> Option { +) -> Result> { if let Some(schedule) = &flow_info.eval_schedule { - return Some(schedule.clone()); + return Ok(Some(schedule.clone())); } - let eval_interval_secs = flow_info.eval_interval_secs?; + let Some(eval_interval_secs) = flow_info.eval_interval_secs else { + return Ok(None); + }; if eval_interval_secs <= 0 { - return None; + return Ok(None); } + // Round the created_time up to the next whole second (same helper as new + // flow resolution) before aligning to the epoch-anchored boundary. + let created_ceil = ceil_to_whole_sec(flow_info.created_time)?; let start_secs = ceil_to_boundary( - flow_info.created_time.timestamp(), + created_ceil, FlowScheduleConfig::DEFAULT_ANCHOR_SECS, eval_interval_secs, - ); + )?; - Some(FlowScheduleConfig::default_with_start( + Ok(Some(FlowScheduleConfig::default_with_start( start_secs, eval_interval_secs, - )) + ))) } /// Resolve `FlowScheduleConfig` into `task.eval_schedule`. @@ -591,58 +671,77 @@ pub fn effective_eval_schedule_from_flow_info( /// The function is idempotent: if `task.eval_schedule` is already `Some`, /// it returns immediately (important for procedure retry / dump-restore). /// -/// Schedule configuration is NOT read from `task.flow_options` (those keys are -/// no longer user-facing options). For new flows, pure defaults are computed. -/// For OR REPLACE, the previous typed config is preserved when interval+anchor -/// are unchanged; otherwise defaults are recomputed. +/// The schedule phase (anchor) is the `EVAL OFFSET` value: boundaries are +/// `offset + k * interval` (Unix epoch seconds), independent of timezone/DST. +/// An omitted offset means zero. It is NOT anchored to create time and does +/// not drift. Schedule configuration is NOT read from `task.flow_options` +/// (those keys are no longer user-facing options). +/// +/// For OR REPLACE, the previous typed config is preserved when interval+offset +/// are unchanged; otherwise the schedule is recomputed. +/// +/// Fallible: if the next phase boundary cannot fit in `i64` (extremely far +/// future), an explicit error is returned and flow creation fails instead of +/// silently clamping to a non-phase timestamp. pub(crate) fn resolve_schedule_defaults_into_task( task: &mut CreateFlowTask, prev_flow_info: Option<&FlowInfoValue>, -) { +) -> Result<()> { // Idempotent: if already computed, skip recomputation. if task.eval_schedule.is_some() { - return; + return Ok(()); } let Some(eval_interval_secs) = task.eval_interval_secs else { - return; + return Ok(()); }; if eval_interval_secs <= 0 { - return; + return Ok(()); } - let anchor_secs = FlowScheduleConfig::DEFAULT_ANCHOR_SECS; + let anchor_secs = task.eval_offset_secs.unwrap_or(0); + + // Defense: `validate_flow_options` (called before this in `on_prepare`) + // already rejects out-of-range offsets; guard here to never schedule on a + // garbage anchor. + if !(0..eval_interval_secs).contains(&anchor_secs) { + return Ok(()); + } // For OR REPLACE: if interval+anchor unchanged, preserve the entire // existing typed config so start / policy / limits are stable. if task.or_replace && let Some(prev) = prev_flow_info - && let Some(old_sched) = effective_eval_schedule_from_flow_info(prev) + && let Some(old_sched) = effective_eval_schedule_from_flow_info(prev)? { let old_interval = prev.eval_interval_secs.unwrap_or(0); if old_interval == eval_interval_secs && old_sched.anchor_secs == anchor_secs { task.eval_schedule = Some(old_sched); - return; + return Ok(()); } } - // --- Compute start --- - let start_secs = - // New flow, or OR REPLACE with changed interval/anchor: start at the next - // aligned boundary after prepare time. The value is written into - // task.eval_schedule once so procedure retry does not recompute it. - ceil_to_boundary(chrono::Utc::now().timestamp(), anchor_secs, eval_interval_secs); + // New flow, or OR REPLACE with changed interval/offset: start at the next + // aligned boundary strictly after the exact prepare instant, rounded up to + // the next whole second so the first boundary is never in the past. The + // value is written into task.eval_schedule once so procedure retry does not + // recompute it. + let prepare_secs = ceil_to_whole_sec(chrono::Utc::now())?; + let start_secs = ceil_to_boundary(prepare_secs, anchor_secs, eval_interval_secs)?; - task.eval_schedule = Some(FlowScheduleConfig::default_with_start( + task.eval_schedule = Some(FlowScheduleConfig::with_anchor( + anchor_secs, start_secs, eval_interval_secs, )); + Ok(()) } fn user_runtime_flow_options(options: &HashMap) -> HashMap { let mut options = options.clone(); options.remove(DEFER_ON_MISSING_SOURCE_KEY); options.remove(INTERNAL_EVAL_SCHEDULE_KEY); + options.remove(INTERNAL_EVAL_OFFSET_KEY); options } @@ -772,14 +871,16 @@ impl From<&CreateFlowData> for (FlowInfoValue, Vec<(FlowPartitionId, FlowRouteVa let sql = value.task.sql.clone(); let eval_schedule = value.task.eval_schedule.clone(); - // Start with a clean options map. The transient schedule payload is - // only for the meta→flownode CreateRequest boundary and must not be - // persisted in FlowInfoValue.options. + // Start with a clean options map. The transient schedule/offset payloads + // are only for the meta→flownode / frontend→meta boundaries and must not + // be persisted in FlowInfoValue.options. let mut options: HashMap = value .task .flow_options .iter() - .filter(|(k, _)| k.as_str() != INTERNAL_EVAL_SCHEDULE_KEY) + .filter(|(k, _)| { + k.as_str() != INTERNAL_EVAL_SCHEDULE_KEY && k.as_str() != INTERNAL_EVAL_OFFSET_KEY + }) .map(|(k, v)| (k.clone(), v.clone())) .collect(); diff --git a/src/common/meta/src/ddl/tests/create_flow.rs b/src/common/meta/src/ddl/tests/create_flow.rs index b526856bda..e7004656fe 100644 --- a/src/common/meta/src/ddl/tests/create_flow.rs +++ b/src/common/meta/src/ddl/tests/create_flow.rs @@ -25,9 +25,9 @@ use table::table_name::TableName; use crate::ddl::DdlContext; use crate::ddl::create_flow::{ CreateFlowData, CreateFlowProcedure, CreateFlowState, DEFER_ON_MISSING_SOURCE_KEY, - FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY, FlowType, defer_on_missing_source, - effective_eval_schedule_from_flow_info, resolve_schedule_defaults_into_task, - validate_flow_options, + FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY, FlowType, ceil_to_whole_sec, + defer_on_missing_source, effective_eval_schedule_from_flow_info, + resolve_schedule_defaults_into_task, validate_flow_options, }; use crate::ddl::test_util::create_table::test_create_table_task; use crate::ddl::test_util::flownode_handler::NaiveFlownodeHandler; @@ -67,6 +67,7 @@ pub(crate) fn test_create_flow_task( create_if_not_exists, expire_after: Some(300), eval_interval_secs: None, + eval_offset_secs: None, comment: "".to_string(), sql: "select 1".to_string(), flow_options: Default::default(), @@ -333,7 +334,6 @@ fn test_validate_flow_options_rejects_schedule_and_internal_keys_as_unknown() { "eval_interval_missed_tick_policy", "eval_interval_catchup_max_runs", "eval_interval_catchup_max_lag", - "__greptime_internal_eval_schedule", ] { let mut task = test_create_flow_task( "my_flow", @@ -354,6 +354,78 @@ fn test_validate_flow_options_rejects_schedule_and_internal_keys_as_unknown() { } } +#[test] +fn test_validate_flow_options_rejects_internal_transport_keys() { + for key in [ + "__greptime_internal_eval_schedule", + "__greptime_internal_eval_offset_secs", + ] { + let mut task = test_create_flow_task( + "my_flow", + vec![], + TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_sink_table"), + false, + ); + task.eval_interval_secs = Some(300); + task.flow_options + .insert(key.to_string(), "value".to_string()); + + let err = validate_flow_options(&task).unwrap_err(); + assert!( + err.to_string() + .contains(&format!("flow option '{key}' is reserved for internal use")), + "unexpected error for {key}: {err}" + ); + } +} + +#[test] +fn test_validate_flow_options_rejects_invalid_eval_offset() { + let mut task = test_create_flow_task( + "my_flow", + vec![], + TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_sink_table"), + false, + ); + task.eval_interval_secs = None; + task.eval_offset_secs = Some(60); + let err = validate_flow_options(&task).unwrap_err(); + assert!( + err.to_string() + .contains("EVAL OFFSET requires EVAL INTERVAL"), + "unexpected error: {err}" + ); + + for offset_secs in [-1, 300, 301] { + let mut task = test_create_flow_task( + "my_flow", + vec![], + TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_sink_table"), + false, + ); + task.eval_interval_secs = Some(300); + task.eval_offset_secs = Some(offset_secs); + let err = validate_flow_options(&task).unwrap_err(); + assert!( + err.to_string() + .contains("EVAL OFFSET must be in range [0, EVAL INTERVAL)"), + "unexpected error for offset {offset_secs}: {err}" + ); + } + + for offset_secs in [0, 1, 299] { + let mut task = test_create_flow_task( + "my_flow", + vec![], + TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_sink_table"), + false, + ); + task.eval_interval_secs = Some(300); + task.eval_offset_secs = Some(offset_secs); + validate_flow_options(&task).unwrap(); + } +} + #[test] fn test_validate_flow_options_rejects_non_positive_eval_interval() { for secs in [0, -1] { @@ -381,7 +453,7 @@ fn test_resolved_schedule_defaults_in_create_request() { task.eval_interval_secs = Some(300); // Simulate on_prepare: resolve defaults into task. - resolve_schedule_defaults_into_task(&mut task, None); + resolve_schedule_defaults_into_task(&mut task, None).unwrap(); // Verify typed schedule config is populated. let sched = task.eval_schedule.as_ref().unwrap(); @@ -410,7 +482,7 @@ fn test_resolved_schedule_defaults_in_create_request() { // Idempotent: second call does not overwrite. let start_before = sched.start_secs; - resolve_schedule_defaults_into_task(&mut task, None); + resolve_schedule_defaults_into_task(&mut task, None).unwrap(); assert_eq!( task.eval_schedule.as_ref().unwrap().start_secs, start_before @@ -483,6 +555,221 @@ fn test_resolved_schedule_defaults_in_create_request() { } } +#[test] +fn test_resolve_schedule_with_eval_offset_uses_epoch_phase() { + // 1h interval + 120s offset must yield the `:02` phase: anchor 120 and a + // start on `120 + k * 3600`, and no earlier boundary than the prepare ceil. + let prepare_ceil = ceil_to_whole_sec(chrono::Utc::now()).unwrap(); + + let mut task = test_create_flow_task( + "my_flow", + vec![], + TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_sink_table"), + false, + ); + task.eval_interval_secs = Some(3600); + task.eval_offset_secs = Some(120); + + resolve_schedule_defaults_into_task(&mut task, None).unwrap(); + let sched = task.eval_schedule.as_ref().unwrap(); + assert_eq!(sched.anchor_secs, 120); + assert!(sched.start_secs >= prepare_ceil); + assert_eq!((sched.start_secs - 120) % 3600, 0); + + // The offset key must not leak into user runtime options / persisted info. + let flow_context = FlowQueryContext { + catalog: DEFAULT_CATALOG_NAME.to_string(), + schema: DEFAULT_SCHEMA_NAME.to_string(), + timezone: "UTC".to_string(), + extensions: HashMap::new(), + channel: 0, + snapshot_seqs: HashMap::new(), + sst_min_sequences: HashMap::new(), + }; + let data = CreateFlowData { + state: CreateFlowState::CreateFlows, + task, + flow_id: Some(1024), + peers: vec![], + source_table_ids: vec![], + unresolved_source_table_names: vec![], + flow_context: flow_context.clone(), + prev_flow_info_value: None, + did_replace: false, + flow_type: Some(FlowType::Batching), + }; + let request: CreateRequest = (&data).into(); + assert!( + !request + .flow_options + .contains_key("__greptime_internal_eval_offset_secs"), + "CreateRequest must not contain internal eval offset key" + ); + let (flow_info, _) = <(FlowInfoValue, Vec<(_, _)>)>::from(&data); + assert_eq!(flow_info.eval_schedule.as_ref().unwrap().anchor_secs, 120); + assert!( + !flow_info + .options + .contains_key("__greptime_internal_eval_offset_secs"), + "FlowInfoValue.options must not contain internal eval offset key" + ); +} + +#[test] +fn test_resolve_schedule_or_replace_preserves_and_recomputes() { + let sink = TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_sink_table"); + + let mut task = test_create_flow_task("my_flow", vec![], sink.clone(), false); + task.or_replace = false; + task.eval_interval_secs = Some(3600); + task.eval_offset_secs = Some(120); + resolve_schedule_defaults_into_task(&mut task, None).unwrap(); + let original_sched = task.eval_schedule.clone().unwrap(); + assert_eq!(original_sched.anchor_secs, 120); + + let prev_info = flow_info_for_schedule_test( + Some(original_sched.clone()), + Some(3600), + HashMap::new(), + 1_700_000_000, + ); + + let mut replace_same = test_create_flow_task("my_flow", vec![], sink.clone(), false); + replace_same.or_replace = true; + replace_same.eval_interval_secs = Some(3600); + replace_same.eval_offset_secs = Some(120); + resolve_schedule_defaults_into_task(&mut replace_same, Some(&prev_info)).unwrap(); + assert_eq!(replace_same.eval_schedule, Some(original_sched.clone())); + + let mut replace_changed = test_create_flow_task("my_flow", vec![], sink.clone(), false); + replace_changed.or_replace = true; + replace_changed.eval_interval_secs = Some(3600); + replace_changed.eval_offset_secs = Some(180); + resolve_schedule_defaults_into_task(&mut replace_changed, Some(&prev_info)).unwrap(); + let recomputed = replace_changed.eval_schedule.as_ref().unwrap(); + assert_eq!(recomputed.anchor_secs, 180); + assert_eq!((recomputed.start_secs - 180) % 3600, 0); + assert_ne!(recomputed.start_secs, original_sched.start_secs); + + let mut replace_interval = test_create_flow_task("my_flow", vec![], sink.clone(), false); + replace_interval.or_replace = true; + replace_interval.eval_interval_secs = Some(1800); + replace_interval.eval_offset_secs = Some(120); + resolve_schedule_defaults_into_task(&mut replace_interval, Some(&prev_info)).unwrap(); + assert_eq!( + replace_interval.eval_schedule.as_ref().unwrap().anchor_secs, + 120 + ); + + let mut replace_omitted = test_create_flow_task("my_flow", vec![], sink, false); + replace_omitted.or_replace = true; + replace_omitted.eval_interval_secs = Some(3600); + replace_omitted.eval_offset_secs = None; + resolve_schedule_defaults_into_task(&mut replace_omitted, Some(&prev_info)).unwrap(); + let reset = replace_omitted.eval_schedule.as_ref().unwrap(); + assert_eq!(reset.anchor_secs, 0); + assert_eq!(reset.start_secs % 3600, 0); +} + +#[test] +fn test_create_flow_task_serde_defaults_for_old_data() { + // Old serialized CreateFlowTask data without `eval_offset_secs` must + // deserialize with `eval_offset_secs: None`. + let json = r#"{ + "catalog_name": "catalog", + "flow_name": "flow", + "source_table_names": [], + "sink_table_name": {"catalog_name": "catalog", "schema_name": "schema", "table_name": "sink"}, + "or_replace": false, + "create_if_not_exists": false, + "expire_after": null, + "eval_interval_secs": 300, + "comment": "", + "sql": "select 1", + "flow_options": {} + }"#; + let task: CreateFlowTask = serde_json::from_str(json).unwrap(); + assert_eq!(task.eval_offset_secs, None); + assert_eq!(task.eval_interval_secs, Some(300)); + let encoded = serde_json::to_string(&task).unwrap(); + let decoded: CreateFlowTask = serde_json::from_str(&encoded).unwrap(); + assert_eq!(decoded.eval_offset_secs, None); + + // A task with an offset round-trips through JSON (procedure dump/restore). + let mut task = test_create_flow_task( + "my_flow", + vec![], + TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_sink_table"), + false, + ); + task.eval_interval_secs = Some(3600); + task.eval_offset_secs = Some(120); + let encoded = serde_json::to_string(&task).unwrap(); + let decoded: CreateFlowTask = serde_json::from_str(&encoded).unwrap(); + assert_eq!(decoded.eval_offset_secs, Some(120)); + assert_eq!(decoded.eval_interval_secs, Some(3600)); +} + +#[test] +fn test_create_flow_task_proto_roundtrip_preserves_offset_key() { + // Path A: the operator inserts the trusted offset key into flow_options + // (as produced by `to_create_flow_task_expr`); the proto round-trip + // (frontend -> metasrv DDL task) must preserve it. + let mut task = test_create_flow_task( + "my_flow", + vec![], + TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_sink_table"), + false, + ); + task.eval_interval_secs = Some(3600); + task.flow_options.insert( + "__greptime_internal_eval_offset_secs".to_string(), + "120".to_string(), + ); + + let pb: api::v1::meta::CreateFlowTask = task.clone().into(); + assert_eq!( + pb.create_flow + .as_ref() + .unwrap() + .flow_options + .get("__greptime_internal_eval_offset_secs"), + Some(&"120".to_string()) + ); + + let parsed = CreateFlowTask::try_from(pb).unwrap(); + assert_eq!(parsed.eval_offset_secs, Some(120)); + assert!( + !parsed + .flow_options + .contains_key("__greptime_internal_eval_offset_secs") + ); + assert_eq!(parsed.eval_interval_secs, Some(3600)); + + // Path B: a typed task carrying `eval_offset_secs` (e.g. after procedure + // dump/restore) re-inserts the key when converted back to proto. + let mut typed = test_create_flow_task( + "my_flow", + vec![], + TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_sink_table"), + false, + ); + typed.eval_interval_secs = Some(3600); + typed.eval_offset_secs = Some(120); + let pb: api::v1::meta::CreateFlowTask = typed.clone().into(); + assert_eq!( + pb.create_flow + .as_ref() + .unwrap() + .flow_options + .get("__greptime_internal_eval_offset_secs"), + Some(&"120".to_string()) + ); + let reparsed = CreateFlowTask::try_from(pb).unwrap(); + assert_eq!(reparsed.eval_offset_secs, Some(120)); + assert_eq!(reparsed.flow_options, typed.flow_options); +} + #[test] fn test_effective_eval_schedule_prefers_typed_config() { let typed = FlowScheduleConfig { @@ -496,7 +783,9 @@ fn test_effective_eval_schedule_prefers_typed_config() { let flow_info = flow_info_for_schedule_test(Some(typed.clone()), Some(60), unrelated_options, 1); - let effective = effective_eval_schedule_from_flow_info(&flow_info).unwrap(); + let effective = effective_eval_schedule_from_flow_info(&flow_info) + .unwrap() + .unwrap(); assert_eq!(effective, typed); } @@ -504,7 +793,9 @@ fn test_effective_eval_schedule_prefers_typed_config() { fn test_effective_eval_schedule_uses_created_time_default() { let flow_info = flow_info_for_schedule_test(None, Some(60), HashMap::new(), 61); - let effective = effective_eval_schedule_from_flow_info(&flow_info).unwrap(); + let effective = effective_eval_schedule_from_flow_info(&flow_info) + .unwrap() + .unwrap(); assert_eq!(effective.anchor_secs, 0); assert_eq!(effective.start_secs, 120); assert_eq!( @@ -518,7 +809,11 @@ fn test_effective_eval_schedule_uses_created_time_default() { #[test] fn test_effective_eval_schedule_none_without_eval_interval() { let flow_info = flow_info_for_schedule_test(None, None, HashMap::new(), 61); - assert!(effective_eval_schedule_from_flow_info(&flow_info).is_none()); + assert!( + effective_eval_schedule_from_flow_info(&flow_info) + .unwrap() + .is_none() + ); } #[test] @@ -526,8 +821,12 @@ fn test_effective_eval_schedule_deterministic_on_old_metadata() { // Old metadata: no typed eval_schedule, but has eval_interval_secs and // a fixed created_time. Repeated calls must produce the same config. let flow_info = flow_info_for_schedule_test(None, Some(60), HashMap::new(), 61); - let first = effective_eval_schedule_from_flow_info(&flow_info).unwrap(); - let second = effective_eval_schedule_from_flow_info(&flow_info).unwrap(); + let first = effective_eval_schedule_from_flow_info(&flow_info) + .unwrap() + .unwrap(); + let second = effective_eval_schedule_from_flow_info(&flow_info) + .unwrap() + .unwrap(); assert_eq!(first, second); // Sanity: the derived values are based on created_time=61 + interval=60. assert_eq!(first.anchor_secs, 0); @@ -556,7 +855,9 @@ fn test_old_flow_info_json_without_eval_schedule_deserializes() { let decoded: FlowInfoValue = serde_json::from_value(json).unwrap(); assert!(decoded.eval_schedule.is_none()); - let effective = effective_eval_schedule_from_flow_info(&decoded).unwrap(); + let effective = effective_eval_schedule_from_flow_info(&decoded) + .unwrap() + .unwrap(); assert_eq!(effective.anchor_secs, 0); assert_eq!(effective.start_secs, 120); assert_eq!( @@ -974,6 +1275,7 @@ fn create_test_flow_task_for_serialization() -> CreateFlowTask { create_if_not_exists: false, expire_after: None, eval_interval_secs: None, + eval_offset_secs: None, comment: "test comment".to_string(), sql: "SELECT * FROM source_table".to_string(), flow_options: HashMap::new(), diff --git a/src/common/meta/src/key/flow/flow_info.rs b/src/common/meta/src/key/flow/flow_info.rs index ee6a291dd4..605b40a882 100644 --- a/src/common/meta/src/key/flow/flow_info.rs +++ b/src/common/meta/src/key/flow/flow_info.rs @@ -172,8 +172,16 @@ impl FlowScheduleConfig { } pub fn default_with_start(start_secs: i64, eval_interval_secs: i64) -> Self { + Self::with_anchor(Self::DEFAULT_ANCHOR_SECS, start_secs, eval_interval_secs) + } + + /// Builds a schedule anchored at `anchor_secs` (the `EVAL OFFSET` phase, + /// in Unix epoch seconds) with `start_secs` as the first scheduled time. + /// Callers must ensure `0 <= anchor_secs < eval_interval_secs` and that + /// `start_secs` lies on an `anchor_secs + k * eval_interval_secs` boundary. + pub fn with_anchor(anchor_secs: i64, start_secs: i64, eval_interval_secs: i64) -> Self { Self { - anchor_secs: Self::DEFAULT_ANCHOR_SECS, + anchor_secs, start_secs, missed_tick_policy: FlowMissedTickPolicy::BoundedCatchUp, catchup_max_runs: Self::DEFAULT_CATCHUP_MAX_RUNS, diff --git a/src/common/meta/src/rpc/ddl.rs b/src/common/meta/src/rpc/ddl.rs index 47bd537b9b..16ac82be19 100644 --- a/src/common/meta/src/rpc/ddl.rs +++ b/src/common/meta/src/rpc/ddl.rs @@ -1296,6 +1296,13 @@ pub struct CreateFlowTask { /// Duration in seconds. Data older than this duration will not be used. pub expire_after: Option, pub eval_interval_secs: Option, + /// Phase offset of the evaluation schedule within `eval_interval_secs`, + /// in seconds. Must be in `[0, eval_interval_secs)`. `None` means a zero + /// offset (epoch-anchored schedule). + /// Transported through the transient option map (no proto field), see + /// `INTERNAL_EVAL_OFFSET_KEY`. + #[serde(default)] + pub eval_offset_secs: Option, pub comment: String, pub sql: String, pub flow_options: HashMap, @@ -1321,11 +1328,27 @@ impl TryFrom for CreateFlowTask { eval_interval, comment, sql, - flow_options, + mut flow_options, } = pb.create_flow.context(error::InvalidProtoMsgSnafu { err_msg: "expected create_flow", })?; + // Parse and strip the trusted transient offset key inserted by the + // operator after user option validation. It must never persist in + // user-visible options. + let eval_offset_secs = + match flow_options.remove(crate::ddl::create_flow::INTERNAL_EVAL_OFFSET_KEY) { + Some(value) => Some(value.parse::().map_err(|_| { + error::UnexpectedSnafu { + err_msg: format!( + "Invalid internal eval offset payload '{value}': expected whole seconds" + ), + } + .build() + })?), + None => None, + }; + Ok(CreateFlowTask { catalog_name, flow_name, @@ -1339,6 +1362,7 @@ impl TryFrom for CreateFlowTask { create_if_not_exists, expire_after: expire_after.map(|e| e.value), eval_interval_secs: eval_interval.map(|e| e.seconds), + eval_offset_secs, comment, sql, flow_options, @@ -1358,12 +1382,21 @@ impl From for PbCreateFlowTask { create_if_not_exists, expire_after, eval_interval_secs: eval_interval, + eval_offset_secs, comment, sql, - flow_options, + mut flow_options, .. }: CreateFlowTask, ) -> Self { + // Re-insert the transient offset key so the proto round-trip (e.g. DDL + // task submission between frontend and metasrv) preserves the offset. + if let Some(offset_secs) = eval_offset_secs { + flow_options.insert( + crate::ddl::create_flow::INTERNAL_EVAL_OFFSET_KEY.to_string(), + offset_secs.to_string(), + ); + } PbCreateFlowTask { create_flow: Some(CreateFlowExpr { catalog_name, diff --git a/src/flow/src/adapter/flownode_impl.rs b/src/flow/src/adapter/flownode_impl.rs index 7e83d5131d..18e7119c35 100644 --- a/src/flow/src/adapter/flownode_impl.rs +++ b/src/flow/src/adapter/flownode_impl.rs @@ -387,7 +387,9 @@ impl FlowDualEngine { comment: Some(info.comment().clone()), sql: info.raw_sql().clone(), flow_options: info.options().clone(), - eval_schedule: effective_eval_schedule_from_flow_info(&info), + eval_schedule: effective_eval_schedule_from_flow_info(&info) + .map_err(BoxedError::new) + .context(ExternalSnafu)?, query_ctx: info .query_context() .clone() diff --git a/src/flow/src/batching_mode/eval_schedule.rs b/src/flow/src/batching_mode/eval_schedule.rs index b1bf86c061..85b000971a 100644 --- a/src/flow/src/batching_mode/eval_schedule.rs +++ b/src/flow/src/batching_mode/eval_schedule.rs @@ -17,7 +17,7 @@ pub use common_meta::key::flow::flow_info::{FlowMissedTickPolicy, FlowScheduleConfig}; use snafu::ensure; -use crate::error::{InvalidQuerySnafu, Result}; +use crate::error::{InvalidQuerySnafu, Result, UnexpectedSnafu}; /// Schedule for an `EVAL INTERVAL` flow. #[derive(Debug, Clone, PartialEq)] @@ -72,6 +72,29 @@ impl EvalSchedule { ) } ); + // The anchor defines the epoch phase `anchor + k * interval`; it + // must be a valid offset within one interval. + ensure!( + c.anchor_secs >= 0 && c.anchor_secs < interval_secs, + InvalidQuerySnafu { + reason: format!( + "Invalid FlowScheduleConfig.anchor_secs: must be in [0, {interval_secs}), got {}", + c.anchor_secs + ) + } + ); + // The start must be phase-consistent with the anchor (on an + // `anchor + k * interval` boundary) and not before the anchor. + ensure!( + c.start_secs >= c.anchor_secs + && (c.start_secs - c.anchor_secs) % interval_secs == 0, + InvalidQuerySnafu { + reason: format!( + "Invalid FlowScheduleConfig.start_secs: must be on an anchor + k * interval boundary and >= anchor, got start={}, anchor={}, interval={}", + c.start_secs, c.anchor_secs, interval_secs + ) + } + ); Self { interval_secs, @@ -96,30 +119,53 @@ impl EvalSchedule { })) } - /// Returns the next scheduled time strictly after `cursor_secs`. - pub fn next_scheduled_time_after(&self, cursor_secs: i64) -> i64 { + /// Returns the next scheduled time strictly after `cursor_secs`, on the + /// `anchor + k * interval` lattice. + /// + /// Fallible: a non-positive interval or a next boundary that does not fit + /// in `i64` yields an explicit error instead of a saturated non-phase + /// value such as `i64::MAX`. + pub fn next_scheduled_time_after(&self, cursor_secs: i64) -> Result { next_in_sequence(cursor_secs, self.start_secs, self.interval_secs) } } -fn next_in_sequence(cursor: i64, start: i64, interval: i64) -> i64 { - if interval <= 0 { - return cursor.saturating_add(1).max(start); - } - if cursor < start { - return start; - } +/// The smallest `start + k * interval` value that is strictly after `cursor` +/// (`start` itself lies on the `anchor + k * interval` lattice, so every +/// result is phase-consistent with the anchor). All arithmetic happens in +/// `i128`: `cursor - start` cannot overflow and the result is either exactly +/// on the lattice or an explicit error. +fn next_in_sequence(cursor: i64, start: i64, interval: i64) -> Result { + ensure!( + interval > 0, + InvalidQuerySnafu { + reason: format!("Invalid eval interval: must be positive, got {interval}") + } + ); + let interval = i128::from(interval); + let start = i128::from(start); + let cursor = i128::from(cursor); - let k = (cursor - start) / interval; - start.saturating_add((k + 1).saturating_mul(interval)) + let next = if cursor < start { + start + } else { + let k = (cursor - start) / interval; + start + (k + 1) * interval + }; + + i64::try_from(next).map_err(|_| { + UnexpectedSnafu { + reason: format!( + "Cannot advance the eval schedule past cursor {cursor}: the next scheduled time {next} does not fit in i64 (start={start}, interval={interval})" + ), + } + .build() + }) } -fn first_due_in_sequence(cursor: i64, start: i64, interval: i64) -> i64 { - if interval <= 0 { - return cursor.saturating_add(1).max(start); - } +fn first_due_in_sequence(cursor: i64, start: i64, interval: i64) -> Result { if cursor < start { - start + Ok(start) } else { next_in_sequence(cursor, start, interval) } @@ -139,59 +185,112 @@ pub struct DueScheduledTimes { } /// Select due scheduled times `<= wall_now_secs` without materializing all missed ticks. +/// +/// Fallible: a non-positive interval or a scheduled time that does not fit in +/// `i64` yields an explicit error instead of silently producing saturated +/// non-phase timestamps. pub fn select_due_scheduled_times( schedule: &EvalSchedule, cursor_secs: i64, wall_now_secs: i64, -) -> Option { - if schedule.interval_secs <= 0 { - return None; - } +) -> Result { + let interval = schedule.interval_secs; + ensure!( + interval > 0, + InvalidQuerySnafu { + reason: format!("Invalid eval interval: must be positive, got {interval}") + } + ); - let first_due = first_due_in_sequence(cursor_secs, schedule.start_secs, schedule.interval_secs); + let first_due = first_due_in_sequence(cursor_secs, schedule.start_secs, interval)?; if first_due > wall_now_secs { - return Some(DueScheduledTimes { + return Ok(DueScheduledTimes { scheduled_times_secs: vec![], skipped: 0, }); } - let total_count = ((wall_now_secs - first_due) / schedule.interval_secs) as u64 + 1; + // Count and select due scheduled times in i128 so every value stays + // exactly on the `anchor + k * interval` lattice; a value beyond `i64` is + // an explicit error, never a saturated non-phase timestamp. + let first_due = i128::from(first_due); + let wall_now = i128::from(wall_now_secs); + let interval = i128::from(interval); + + let total_count = (wall_now - first_due) / interval + 1; + // `first_due >= 0` and `wall_now <= i64::MAX`, so this always fits in u64. + let total_count = u64::try_from(total_count).map_err(|_| { + UnexpectedSnafu { + reason: format!( + "Cannot count due eval scheduled times up to {wall_now}: {total_count} does not fit in u64" + ), + } + .build() + })?; + match schedule.missed_tick_policy { FlowMissedTickPolicy::Skip => { - let last = first_due + (total_count as i64 - 1) * schedule.interval_secs; - Some(DueScheduledTimes { + // Keep only the latest due scheduled time; it is still on-lattice + // and `<= wall_now`. + let last = i64::try_from(first_due + i128::from(total_count - 1) * interval) + .map_err(|_| { + UnexpectedSnafu { + reason: format!( + "Cannot compute the latest due eval scheduled time (first_due={first_due}, interval={interval}, count={total_count}): result does not fit in i64" + ), + } + .build() + })?; + Ok(DueScheduledTimes { scheduled_times_secs: vec![last], - skipped: total_count.saturating_sub(1), + skipped: total_count - 1, }) } FlowMissedTickPolicy::BoundedCatchUp => { - let cutoff = wall_now_secs.saturating_sub(schedule.max_lag_secs); + // The cutoff is computed in i128: `wall_now - max_lag` may + // legitimately underflow i64 (a cutoff before the Unix epoch) and + // must not saturate to a wrong value. + let cutoff = wall_now - i128::from(schedule.max_lag_secs); let skipped_by_cutoff = if first_due >= cutoff { 0 } else { - ((cutoff - first_due + schedule.interval_secs - 1) / schedule.interval_secs) as u64 + // ceil((cutoff - first_due) / interval), capped at u64::MAX + // before the `.min(total_count)` below. + let skipped = (cutoff - first_due + interval - 1) / interval; + u64::try_from(skipped).unwrap_or(u64::MAX) } .min(total_count); - let remaining = total_count.saturating_sub(skipped_by_cutoff); + let remaining = total_count - skipped_by_cutoff; if remaining == 0 { - return Some(DueScheduledTimes { + return Ok(DueScheduledTimes { scheduled_times_secs: vec![], skipped: total_count, }); } - // max_lag_secs decides which missed scheduled times are recent enough to - // run; max_runs caps how many of those times we execute - // back-to-back in one scheduler pass. - let keep_count = remaining.min(schedule.max_runs as u64); - let keep_start = skipped_by_cutoff + remaining.saturating_sub(keep_count); - let scheduled_times_secs = (0..keep_count) - .map(|i| first_due + (keep_start as i64 + i as i64) * schedule.interval_secs) - .collect::>(); + // max_lag decides which missed scheduled times are recent enough to + // run; max_runs caps how many of those times execute back-to-back + // in one scheduler pass. + let keep_count = remaining.min(u64::from(schedule.max_runs)); + let keep_start = skipped_by_cutoff + remaining - keep_count; + let mut scheduled_times_secs = Vec::with_capacity(keep_count as usize); + for i in 0..keep_count { + let t = i64::try_from( + first_due + (i128::from(keep_start) + i128::from(i)) * interval, + ) + .map_err(|_| { + UnexpectedSnafu { + reason: format!( + "Cannot compute a due eval scheduled time (first_due={first_due}, interval={interval}, index={i}): result does not fit in i64" + ), + } + .build() + })?; + scheduled_times_secs.push(t); + } - Some(DueScheduledTimes { + Ok(DueScheduledTimes { scheduled_times_secs, skipped: total_count - keep_count, }) @@ -200,12 +299,15 @@ pub fn select_due_scheduled_times( } /// Ceils `time` to the next `anchor + k * interval` boundary. -pub fn ceil_to_boundary(time: i64, anchor: i64, interval: i64) -> i64 { +/// +/// Fallible: if the next boundary does not fit in `i64`, an explicit error is +/// returned instead of clamping to a non-phase value such as `i64::MAX`. +pub fn ceil_to_boundary(time: i64, anchor: i64, interval: i64) -> Result { if interval <= 0 { - return time; + return Ok(time); } if time <= anchor { - return anchor; + return Ok(anchor); } let diff = i128::from(time) - i128::from(anchor); @@ -213,7 +315,14 @@ pub fn ceil_to_boundary(time: i64, anchor: i64, interval: i64) -> i64 { let k = (diff + interval - 1) / interval; let boundary = i128::from(anchor) + k * interval; - boundary.clamp(i128::from(i64::MIN), i128::from(i64::MAX)) as i64 + i64::try_from(boundary).map_err(|_| { + crate::error::UnexpectedSnafu { + reason: format!( + "Cannot align time {time} to the next `anchor + k * interval` boundary (anchor={anchor}, interval={interval}): result {boundary} does not fit in i64" + ), + } + .build() + }) } #[cfg(test)] @@ -239,7 +348,8 @@ mod test { fn config(policy: FlowMissedTickPolicy) -> FlowScheduleConfig { FlowScheduleConfig { anchor_secs: 10, - start_secs: 70, + // phase-consistent: 310 = anchor(10) + 1 * interval(300) + start_secs: 310, missed_tick_policy: policy, catchup_max_runs: 4, catchup_max_lag_secs: 600, @@ -248,14 +358,15 @@ mod test { #[test] fn ceil_to_boundary_handles_anchor_and_interval_edges() { - assert_eq!(ceil_to_boundary(-10, 0, 60), 0); - assert_eq!(ceil_to_boundary(0, 0, 60), 0); - assert_eq!(ceil_to_boundary(1, 0, 60), 60); - assert_eq!(ceil_to_boundary(60, 0, 60), 60); - assert_eq!(ceil_to_boundary(101, 100, 60), 160); - assert_eq!(ceil_to_boundary(50, 0, 0), 50); - assert_eq!(ceil_to_boundary(i64::MAX, 0, 60), i64::MAX); - assert_eq!(ceil_to_boundary(i64::MAX - 1, i64::MIN, 60), i64::MAX); + assert_eq!(ceil_to_boundary(-10, 0, 60).unwrap(), 0); + assert_eq!(ceil_to_boundary(0, 0, 60).unwrap(), 0); + assert_eq!(ceil_to_boundary(1, 0, 60).unwrap(), 60); + assert_eq!(ceil_to_boundary(60, 0, 60).unwrap(), 60); + assert_eq!(ceil_to_boundary(101, 100, 60).unwrap(), 160); + assert_eq!(ceil_to_boundary(50, 0, 0).unwrap(), 50); + // Never clamp to the non-phase i64::MAX: the next boundary does not fit. + assert!(ceil_to_boundary(i64::MAX, 0, 60).is_err()); + assert!(ceil_to_boundary(i64::MAX - 1, i64::MIN, 60).is_err()); } #[test] @@ -269,7 +380,7 @@ mod test { .unwrap(); assert_eq!(from_typed.interval_secs, 300); assert_eq!(from_typed.anchor_secs, 10); - assert_eq!(from_typed.start_secs, 70); + assert_eq!(from_typed.start_secs, 310); assert_eq!(from_typed.missed_tick_policy, FlowMissedTickPolicy::Skip); assert_eq!(from_typed.max_runs, 4); assert_eq!(from_typed.max_lag_secs, 600); @@ -291,12 +402,72 @@ mod test { assert!(EvalSchedule::from_config(Some(300), Some(&c)).is_err()); } + #[test] + fn nonzero_anchor_due_selection_follows_phase() { + // anchor=120 (i.e. `EVAL OFFSET '2 minutes'`), interval=3600: + // boundaries at :02 every hour. start=3720 (120 + 3600). + let s = EvalSchedule { + interval_secs: 3600, + anchor_secs: 120, + start_secs: 3720, + missed_tick_policy: FlowMissedTickPolicy::BoundedCatchUp, + max_runs: 3, + max_lag_secs: 3600, + }; + assert_eq!( + select_due_scheduled_times(&s, 0, 100) + .unwrap() + .scheduled_times_secs, + Vec::::new() + ); + // From 3720 on, every selected time must be on the :02 phase. + let due = select_due_scheduled_times(&s, 0, 3720).unwrap(); + assert_eq!(due.scheduled_times_secs, vec![3720]); + let due = select_due_scheduled_times(&s, 3720, 7320).unwrap(); + assert_eq!(due.scheduled_times_secs, vec![7320]); + for t in &due.scheduled_times_secs { + assert_eq!((t - 120) % 3600, 0); + } + assert_eq!(s.next_scheduled_time_after(3720).unwrap(), 7320); + assert_eq!(s.next_scheduled_time_after(7300).unwrap(), 7320); + } + #[test] fn next_scheduled_time_after_respects_start_sequence() { let s = schedule(50, FlowMissedTickPolicy::BoundedCatchUp, 3, 300); - assert_eq!(s.next_scheduled_time_after(0), 50); - assert_eq!(s.next_scheduled_time_after(50), 110); - assert_eq!(s.next_scheduled_time_after(100), 110); + assert_eq!(s.next_scheduled_time_after(0).unwrap(), 50); + assert_eq!(s.next_scheduled_time_after(50).unwrap(), 110); + assert_eq!(s.next_scheduled_time_after(100).unwrap(), 110); + } + + #[test] + fn near_i64_max_advancement_is_exact_or_explicit_error() { + // anchor=0, interval=60: the next boundary after i64::MAX - 60 is + // 9223372036854775800, still in range and exactly on the lattice. + let s = schedule(0, FlowMissedTickPolicy::Skip, 5, 3600); + let cursor = i64::MAX - 60; + let next = s.next_scheduled_time_after(cursor).unwrap(); + assert_eq!(next, 9223372036854775800); + assert_eq!(next % 60, 0); + + // Advancing past the last representable boundary is an explicit error, + // never a saturated non-phase value like i64::MAX. + let err = s + .next_scheduled_time_after(9223372036854775800) + .unwrap_err(); + assert!(err.to_string().contains("does not fit in i64")); + + // A non-positive interval is an explicit error, not a saturating + // `cursor + 1` result. + let invalid = EvalSchedule { + interval_secs: 0, + anchor_secs: 0, + start_secs: 0, + missed_tick_policy: FlowMissedTickPolicy::Skip, + max_runs: 3, + max_lag_secs: 900, + }; + assert!(invalid.next_scheduled_time_after(0).is_err()); } #[test] diff --git a/src/flow/src/batching_mode/task.rs b/src/flow/src/batching_mode/task.rs index 196c9ea42d..fb64dc9c20 100644 --- a/src/flow/src/batching_mode/task.rs +++ b/src/flow/src/batching_mode/task.rs @@ -79,6 +79,61 @@ fn wall_clock_unix_secs() -> i64 { .as_secs() as i64 } +/// Initial scheduler cursor for `start_scheduled_loop`: exactly one interval +/// before `start_secs` so the first due scheduled time is `start_secs` itself. +/// +/// Fallible: a `start_secs - interval_secs` difference that does not fit in +/// `i64` is an explicit error instead of a saturated cursor that would make +/// the first due scheduled time `start_secs + interval_secs` and silently skip +/// the `start_secs` tick. +fn initial_schedule_cursor(start_secs: i64, interval_secs: i64) -> Result { + let cursor = i128::from(start_secs) - i128::from(interval_secs); + i64::try_from(cursor).map_err(|_| { + UnexpectedSnafu { + reason: format!( + "Cannot compute the initial eval schedule cursor one interval before start {start_secs} (interval={interval_secs}): {cursor} does not fit in i64" + ), + } + .build() + }) +} + +/// Whole seconds to sleep until the next scheduled time `next`, measured from +/// the current wall clock `wall_now_secs`. +/// +/// Fallible: `next` must be strictly after `wall_now_secs` and the difference +/// must fit in `u64`. In practice `i64::MAX - i64::MIN` is exactly `u64::MAX`, +/// so the difference always fits once `next > wall_now_secs`; the explicit +/// error keeps the scheduled loop panic-free and wrap-free regardless. +fn sleep_delta_secs(next: i64, wall_now_secs: i64) -> Result { + let delta = i128::from(next) - i128::from(wall_now_secs); + u64::try_from(delta).map_err(|_| { + UnexpectedSnafu { + reason: format!( + "Cannot sleep until the next scheduled time {next}: the delta from wall clock {wall_now_secs} is {delta} seconds, which does not fit in u64" + ), + } + .build() + }) +} + +/// Scheduled time in seconds converted to milliseconds for the +/// `FLOW_SCHEDULED_TIME_MILLIS` extension. +/// +/// Fallible: a seconds value whose millisecond product does not fit in `i64` +/// is an explicit error instead of a saturated `i64::MAX` that would silently +/// misrepresent the logical scheduled time. +fn scheduled_time_millis(scheduled_time_secs: i64) -> Result { + scheduled_time_secs.checked_mul(1000).ok_or_else(|| { + UnexpectedSnafu { + reason: format!( + "Cannot convert scheduled time {scheduled_time_secs}s to milliseconds: the product exceeds i64" + ), + } + .build() + }) +} + /// The task's config, immutable once created #[derive(Clone)] pub struct TaskConfig { @@ -1020,8 +1075,20 @@ impl BatchingTask { }; // Initial cursor is one interval before start so the first due - // scheduled time is `start_secs`. - let mut cursor_secs = schedule.start_secs.saturating_sub(schedule.interval_secs); + // scheduled time is `start_secs`. An unrepresentable difference is an + // explicit error, never a saturated cursor that would silently skip + // the first scheduled tick. + let mut cursor_secs = + match initial_schedule_cursor(schedule.start_secs, schedule.interval_secs) { + Ok(cursor) => cursor, + Err(e) => { + warn!( + "Flow {}: invalid eval schedule, exiting loop: {}", + flow_id_str, e + ); + return; + } + }; info!( "Flow {}: entering scheduled loop, interval={}s, start={}, anchor={}, policy={:?}, max_runs={}, max_lag={}s", @@ -1042,11 +1109,11 @@ impl BatchingTask { let wall_now_secs = wall_clock_unix_secs(); let due = match select_due_scheduled_times(&schedule, cursor_secs, wall_now_secs) { - Some(d) => d, - None => { + Ok(d) => d, + Err(e) => { warn!( - "Flow {}: Invalid schedule (interval <= 0), exiting loop", - flow_id_str + "Flow {}: invalid eval schedule, exiting loop: {}", + flow_id_str, e ); return; } @@ -1063,14 +1130,32 @@ impl BatchingTask { } // No due yet — sleep until the next scheduled time. - let next = schedule.next_scheduled_time_after(cursor_secs); + let next = match schedule.next_scheduled_time_after(cursor_secs) { + Ok(next) => next, + Err(e) => { + warn!( + "Flow {}: cannot advance eval schedule past cursor {cursor_secs}: {e}; exiting loop", + flow_id_str + ); + return; + } + }; if next <= wall_now_secs { // Shouldn't happen given select_due_scheduled_times returned empty, // but guard against clock skew / logic error. cursor_secs = wall_now_secs; continue; } - let wait_secs = (next - wall_now_secs) as u64; + let wait_secs = match sleep_delta_secs(next, wall_now_secs) { + Ok(wait_secs) => wait_secs, + Err(e) => { + warn!( + "Flow {}: cannot sleep until next scheduled time {}: {e}; exiting loop", + flow_id_str, next + ); + return; + } + }; let wait_dur = Duration::from_secs(wait_secs); debug!( "Flow {}: no due scheduled times, sleeping for {}s until next scheduled time at {}", @@ -1259,6 +1344,19 @@ impl BatchingTask { } } + // Convert to milliseconds before touching the task state so an + // unrepresentable scheduled time fails as an explicit error without + // ever installing a saturated (off-phase) extension value. + let scheduled_time_millis = match scheduled_time_millis(scheduled_time_secs) { + Ok(millis) => millis, + Err(e) => { + return ExecuteOnceOutcome { + new_query: None, + result: Err(e), + }; + } + }; + // Clone the current QueryContext and add the scheduled time // extension, then swap it into the task state for this attempt. let old_ctx = { @@ -1267,7 +1365,7 @@ impl BatchingTask { let mut new_ctx = (*old).clone(); new_ctx.set_extension( query::options::FLOW_SCHEDULED_TIME_MILLIS, - (scheduled_time_secs.saturating_mul(1000)).to_string(), + scheduled_time_millis.to_string(), ); state.query_ctx = Arc::new(new_ctx); old diff --git a/src/flow/src/batching_mode/task/test.rs b/src/flow/src/batching_mode/task/test.rs index 28468aa595..24cdbf883e 100644 --- a/src/flow/src/batching_mode/task/test.rs +++ b/src/flow/src/batching_mode/task/test.rs @@ -43,6 +43,7 @@ use crate::batching_mode::checkpoint::{ CHECKPOINT_DECISION_ADVANCE, CHECKPOINT_DECISION_FALLBACK, CHECKPOINT_REASON_NONE, FlowCheckpointDecision, FlowQueryFallbackReason, }; +use crate::batching_mode::eval_schedule::{FlowMissedTickPolicy, FlowScheduleConfig}; use crate::batching_mode::state::CheckpointMode; use crate::batching_mode::time_window::find_time_window_expr; use crate::test_utils::create_test_query_engine; @@ -142,6 +143,77 @@ async fn test_incremental_read_is_disabled_by_default() { assert!(task.state.read().unwrap().is_incremental_disabled()); } +#[tokio::test] +async fn test_non_aggregate_scheduled_sql_honors_eval_offset_phase() { + // A non-aggregate SQL flow with `EVAL INTERVAL` runs as an explicit + // full-query flow on the batching scheduler. The typed schedule must reach + // the task config unchanged and the offset must not be silently ignored: + // due scheduled times follow the `anchor + k * interval` phase. + let query = "SELECT number, ts FROM numbers_with_ts"; + let query_engine = create_test_query_engine(); + let ctx = QueryContext::arc(); + let plan = sql_to_df_plan(ctx.clone(), query_engine.clone(), query, true) + .await + .unwrap(); + let (_tx, rx) = tokio::sync::oneshot::channel(); + + let schedule = EvalSchedule::from_config( + Some(3600), + Some(&FlowScheduleConfig { + anchor_secs: 120, // `EVAL OFFSET '2 minutes'` + start_secs: 3720, // 120 + 1 * 3600 + missed_tick_policy: FlowMissedTickPolicy::BoundedCatchUp, + catchup_max_runs: 3, + catchup_max_lag_secs: 3600, + }), + ) + .unwrap() + .unwrap(); + + let task = BatchingTask::try_new(TaskArgs { + flow_id: 1, + query, + plan, + time_window_expr: None, + expire_after: None, + sink_table_name: [ + "greptime".to_string(), + "public".to_string(), + "scheduled_non_aggr_sink".to_string(), + ], + source_table_names: vec![[ + "greptime".to_string(), + "public".to_string(), + "numbers_with_ts".to_string(), + ]], + query_ctx: ctx, + catalog_manager: query_engine.engine_state().catalog_manager().clone(), + shutdown_rx: rx, + batch_opts: Arc::new(BatchingModeOptions::default()), + flow_eval_interval: Some(Duration::from_secs(3600)), + eval_schedule: Some(schedule), + }) + .unwrap(); + + let stored = task.config.eval_schedule.as_ref().unwrap(); + assert_eq!(stored.anchor_secs, 120); + assert_eq!(stored.start_secs, 3720); + assert_eq!(stored.interval_secs, 3600); + + // Due-time selection proves the offset is honored: the first due time is + // on the `:02` phase (120 + k * 3600), never 0 or 3600. + let due = select_due_scheduled_times(stored, 0, 3720).unwrap(); + assert_eq!(due.scheduled_times_secs, vec![3720]); + for t in &due.scheduled_times_secs { + assert_eq!((t - 120) % 3600, 0); + } + let due = select_due_scheduled_times(stored, 3720, 7320).unwrap(); + assert_eq!(due.scheduled_times_secs, vec![7320]); + for t in &due.scheduled_times_secs { + assert_eq!((t - 120) % 3600, 0); + } +} + #[tokio::test] async fn test_dirty_time_windows_uses_batch_opts() { let task = new_test_task_engine_and_plan_with_query_and_opts( @@ -764,6 +836,51 @@ async fn test_scheduled_time_now_is_bound_to_selected_attempt() { assert!(!sent_sql.is_empty()); } +/// The scheduled-loop logical-time arithmetic must never clamp, wrap, or +/// panic near the `i64` boundaries: unrepresentable values are explicit +/// errors, representable values are exact. +#[test] +fn test_scheduled_loop_arithmetic_near_i64_boundary() { + assert_eq!(initial_schedule_cursor(3720, 3600).unwrap(), 120); + assert_eq!( + initial_schedule_cursor(i64::MAX, 3600).unwrap(), + i64::MAX - 3600 + ); + // start == i64::MIN cannot go one interval earlier: explicit error, never + // a saturated cursor equal to start that would silently skip the first + // scheduled tick. + assert!(initial_schedule_cursor(i64::MIN, 3600).is_err()); + assert_eq!( + initial_schedule_cursor(i64::MIN + 3600, 3600).unwrap(), + i64::MIN + ); + + assert_eq!( + scheduled_time_millis(1_700_000_000).unwrap(), + 1_700_000_000_000 + ); + // Largest seconds value whose millisecond product still fits in i64. + let max_secs = i64::MAX / 1000; + assert_eq!(scheduled_time_millis(max_secs).unwrap(), max_secs * 1000); + // One more second overflows: explicit error, never a saturated i64::MAX. + let err = scheduled_time_millis(max_secs + 1).unwrap_err(); + assert!(err.to_string().contains("milliseconds"), "{err}"); + assert!(scheduled_time_millis(i64::MAX).is_err()); + + // The widest representable gap (i64::MIN..=i64::MAX) is exactly u64::MAX; + // subtracting in i64 would panic in debug and wrap in release, so the + // i128 path must return the exact u64 value. + assert_eq!( + sleep_delta_secs(i64::MAX, i64::MIN).unwrap(), + u64::MAX, + "i64::MAX - i64::MIN must be exactly u64::MAX, not a wrapped value" + ); + assert_eq!(sleep_delta_secs(7320, 3720).unwrap(), 3600); + // A negative delta (next <= wall_now, violating the caller's guard) is an + // explicit error instead of a wrapped huge `as u64` sleep. + assert!(sleep_delta_secs(100, 200).is_err()); +} + fn output_with_region_watermarks( watermarks: impl IntoIterator)>, ) -> OutputWithMetrics { diff --git a/src/operator/src/expr_helper.rs b/src/operator/src/expr_helper.rs index 3221598cf2..925e062cf5 100644 --- a/src/operator/src/expr_helper.rs +++ b/src/operator/src/expr_helper.rs @@ -1098,6 +1098,7 @@ pub fn to_create_flow_task_expr( let eval_interval = create_flow.eval_interval; + let flow_options = stringify_flow_options(create_flow.flow_options)?; Ok(CreateFlowExpr { catalog_name: query_ctx.current_catalog().to_string(), flow_name: sanitize_flow_name(create_flow.flow_name)?, @@ -1109,7 +1110,7 @@ pub fn to_create_flow_task_expr( eval_interval: eval_interval.map(|seconds| api::v1::EvalInterval { seconds }), comment: create_flow.comment.unwrap_or_default(), sql: create_flow.query.to_string(), - flow_options: stringify_flow_options(create_flow.flow_options)?, + flow_options, }) } diff --git a/src/operator/src/statement/ddl.rs b/src/operator/src/statement/ddl.rs index f89f5de73a..59695a6e7b 100644 --- a/src/operator/src/statement/ddl.rs +++ b/src/operator/src/statement/ddl.rs @@ -42,6 +42,7 @@ use common_meta::cache_invalidator::CacheInvalidatorRef; use common_meta::cache_invalidator::Context; use common_meta::ddl::create_flow::{ DEFER_ON_MISSING_SOURCE_KEY, FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY, FlowType, + INTERNAL_EVAL_OFFSET_KEY, INTERNAL_EVAL_SCHEDULE_KEY, }; use common_meta::instruction::CacheIdent; #[cfg(feature = "enterprise")] @@ -205,7 +206,10 @@ fn validate_and_normalize_flow_options( options .into_iter() .map(|(key, value)| { - if key == FlowType::FLOW_TYPE_KEY { + if matches!( + key.as_str(), + FlowType::FLOW_TYPE_KEY | INTERNAL_EVAL_OFFSET_KEY | INTERNAL_EVAL_SCHEDULE_KEY + ) { return InvalidSqlSnafu { err_msg: format!("flow option '{key}' is reserved for internal use"), } @@ -232,12 +236,51 @@ fn validate_and_normalize_flow_options( .collect() } +/// Validates `EVAL OFFSET` semantics at the operator boundary: an offset is +/// only legal together with `EVAL INTERVAL` and must lie in +/// `[0, eval_interval)`. Never modulo-normalized. +fn validate_eval_offset( + eval_offset_secs: Option, + eval_interval_secs: Option, +) -> Result<()> { + if let Some(offset_secs) = eval_offset_secs { + let Some(eval_interval_secs) = eval_interval_secs else { + return InvalidSqlSnafu { + err_msg: "EVAL OFFSET requires EVAL INTERVAL to be specified".to_string(), + } + .fail(); + }; + if !(0..eval_interval_secs).contains(&offset_secs) { + return InvalidSqlSnafu { + err_msg: format!( + "EVAL OFFSET must be in range [0, EVAL INTERVAL), got {offset_secs} seconds with EVAL INTERVAL {eval_interval_secs} seconds" + ), + } + .fail(); + } + } + Ok(()) +} + fn determine_flow_type_for_source_state( flow_name: &str, flow_options: &HashMap, has_missing_source_table: bool, has_instant_ttl_source_table: bool, + force_batching: bool, ) -> Result> { + if has_instant_ttl_source_table && force_batching { + // The batching scheduler cannot read instant-TTL source tables, so + // reject this combination even when another source table is missing. + return InvalidSqlSnafu { + err_msg: format!( + "flow '{}' with EVAL INTERVAL requires the batching scheduler, but source tables with ttl=instant are not supported under batching mode; use a TTL longer than the flush interval", + flow_name + ), + } + .fail(); + } + if has_missing_source_table { let defer_on_missing_source = flow_options .get(DEFER_ON_MISSING_SOURCE_KEY) @@ -741,23 +784,31 @@ impl StatementExecutor { query_context: QueryContextRef, ) -> Result { // TODO(ruihang): do some verification + let eval_offset_secs = stmt.eval_offset; let expr = expr_helper::to_create_flow_task_expr(stmt, &query_context)?; - self.create_flow_inner(expr, query_context).await + // The typed `EVAL OFFSET` comes straight from the parsed SQL AST; it is + // the only trusted source of the offset at this boundary. + self.create_flow_procedure(expr, eval_offset_secs, query_context) + .await?; + Ok(Output::new_with_affected_rows(0)) } + /// Direct gRPC entry point for `CREATE FLOW`. pub async fn create_flow_inner( &self, expr: CreateFlowExpr, query_context: QueryContextRef, ) -> Result { - self.create_flow_procedure(expr, query_context).await?; + self.create_flow_procedure(expr, None, query_context) + .await?; Ok(Output::new_with_affected_rows(0)) } async fn create_flow_procedure( &self, mut expr: CreateFlowExpr, + eval_offset_secs: Option, query_context: QueryContextRef, ) -> Result { let eval_interval_secs = expr.eval_interval.as_ref().map(|e| e.seconds); @@ -772,6 +823,14 @@ impl StatementExecutor { .fail(); } + // Validate EVAL OFFSET semantics at the operator boundary (defense in + // depth; the parser already enforces them deterministically). The + // offset arrives as a typed parameter, never from `flow_options`. + validate_eval_offset(eval_offset_secs, eval_interval_secs)?; + + // Validate user options. `validate_and_normalize_flow_options` also + // rejects the internal transport keys, so a spoofed key smuggled into + // the expr can never be honored here. expr.flow_options = validate_and_normalize_flow_options(expr.flow_options, eval_interval_secs)?; @@ -783,10 +842,11 @@ impl StatementExecutor { expr.flow_options .insert(FlowType::FLOW_TYPE_KEY.to_string(), flow_type.to_string()); - let task = CreateFlowTask::try_from(PbCreateFlowTask { + let mut task = CreateFlowTask::try_from(PbCreateFlowTask { create_flow: Some(expr), }) .context(error::InvalidExprSnafu)?; + task.eval_offset_secs = eval_offset_secs; let executor_context = to_executor_context(query_context, TriggerReason::Manual); let request = SubmitDdlTaskRequest::new(DdlTask::new_create_flow(task)); @@ -796,14 +856,15 @@ impl StatementExecutor { .context(error::ExecuteDdlSnafu) } - /// Determine the flow type based on the SQL query - /// - /// If it contains aggregation or distinct, then it is a batch flow, otherwise it is a streaming flow + /// Determines the flow type from source-table state, schedule requirements, + /// and SQL shape. async fn determine_flow_type( &self, expr: &CreateFlowExpr, query_ctx: QueryContextRef, ) -> Result { + let has_eval_interval = expr.eval_interval.is_some(); + let mut has_missing_source_table = false; let mut has_instant_ttl_source_table = false; @@ -827,13 +888,18 @@ impl StatementExecutor { if table.table_info().meta.options.ttl == Some(common_time::TimeToLive::Instant) { warn!( - "Source table `{}` for flow `{}`'s ttl=instant, fallback to streaming mode", + "Source table `{}` for flow `{}`'s ttl=instant, {}", format_full_table_name( &src_table_name.catalog_name, &src_table_name.schema_name, &src_table_name.table_name ), - expr.flow_name + expr.flow_name, + if has_eval_interval { + "rejecting flow because EVAL INTERVAL requires the batching scheduler which does not support instant TTL" + } else { + "fallback to streaming mode" + } ); has_instant_ttl_source_table = true; } @@ -844,10 +910,20 @@ impl StatementExecutor { &expr.flow_options, has_missing_source_table, has_instant_ttl_source_table, + has_eval_interval, )? { return Ok(flow_type); } + // A flow with `EVAL INTERVAL` (and therefore possibly `EVAL OFFSET`) + // always uses the batching scheduler: the fixed epoch-phase schedule is + // only honored by the batching scheduler, regardless of the SQL shape. + // Non-aggregate SQL is supported as an explicit full-query flow (the + // batching engine requires a time-window expression or EVAL INTERVAL). + if has_eval_interval { + return Ok(FlowType::Batching); + } + let engine = &self.query_engine; let stmts = ParserContext::create_with_dialect( &expr.sql, @@ -3229,7 +3305,6 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;"; "eval_interval_missed_tick_policy", "eval_interval_catchup_max_runs", "eval_interval_catchup_max_lag", - "__greptime_internal_eval_schedule", ] { let err = validate_and_normalize_flow_options( HashMap::from([(key.to_string(), "value".to_string())]), @@ -3246,10 +3321,31 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;"; } #[test] - fn test_determine_flow_type_for_source_state_missing_sources_require_opt_in() { - let err = determine_flow_type_for_source_state("my_flow", &HashMap::new(), true, false) + fn test_internal_transport_keys_rejected_as_reserved() { + for key in [ + "__greptime_internal_eval_schedule", + "__greptime_internal_eval_offset_secs", + ] { + let err = validate_and_normalize_flow_options( + HashMap::from([(key.to_string(), "120".to_string())]), + Some(300), + ) .unwrap_err(); + assert!( + err.to_string() + .contains(&format!("flow option '{key}' is reserved for internal use")), + "unexpected error for {key}: {err}" + ); + } + } + + #[test] + fn test_determine_flow_type_for_source_state_missing_sources_require_opt_in() { + let err = + determine_flow_type_for_source_state("my_flow", &HashMap::new(), true, false, false) + .unwrap_err(); + assert!(err.to_string().contains( "missing source tables for flow 'my_flow'; use WITH (defer_on_missing_source = true) to create a pending flow" )); @@ -3261,19 +3357,43 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;"; HashMap::from([(DEFER_ON_MISSING_SOURCE_KEY.to_string(), "true".to_string())]); assert_eq!( - determine_flow_type_for_source_state("my_flow", &flow_options, true, true).unwrap(), + determine_flow_type_for_source_state("my_flow", &flow_options, true, true, false) + .unwrap(), Some(FlowType::Batching) ); + let err = determine_flow_type_for_source_state("my_flow", &flow_options, true, true, true) + .unwrap_err(); + assert!(err.to_string().contains( + "flow 'my_flow' with EVAL INTERVAL requires the batching scheduler, but source tables with ttl=instant are not supported under batching mode" + )); } #[test] fn test_determine_flow_type_for_source_state_instant_ttl_without_missing_sources() { assert_eq!( - determine_flow_type_for_source_state("my_flow", &HashMap::new(), false, true).unwrap(), + determine_flow_type_for_source_state("my_flow", &HashMap::new(), false, true, false) + .unwrap(), Some(FlowType::Streaming) ); } + #[test] + fn test_determine_flow_type_for_source_state_instant_ttl_with_eval_interval_is_rejected() { + // A flow with `EVAL INTERVAL` is forced onto the batching scheduler, + // which cannot read instant-TTL source tables: reject instead of + // silently falling back to streaming (where the schedule/offset would + // be ignored). + let err = + determine_flow_type_for_source_state("my_flow", &HashMap::new(), false, true, true) + .unwrap_err(); + assert!( + err.to_string().contains( + "flow 'my_flow' with EVAL INTERVAL requires the batching scheduler, but source tables with ttl=instant are not supported under batching mode" + ), + "unexpected error: {err}" + ); + } + #[test] fn test_name_is_match() { assert!(!NAME_PATTERN_REG.is_match("/adaf")); diff --git a/src/query/src/sql.rs b/src/query/src/sql.rs index 93e74ca33e..7527d44d82 100644 --- a/src/query/src/sql.rs +++ b/src/query/src/sql.rs @@ -32,8 +32,9 @@ use common_catalog::format_full_table_name; use common_datasource::file_format::{FileFormat, Format, infer_schemas}; use common_datasource::lister::{Lister, Source}; use common_datasource::object_store::{LocalFileAccess, build_backend_with_path}; +use common_error::ext::BoxedError; use common_meta::SchemaOptions; -use common_meta::ddl::create_flow::FlowType; +use common_meta::ddl::create_flow::{FlowType, effective_eval_schedule_from_flow_info}; use common_meta::key::flow::flow_info::FlowInfoValue; use common_query::Output; use common_query::prelude::greptime_timestamp; @@ -1227,6 +1228,11 @@ pub fn show_create_flow( if_not_exists: true, expire_after: flow_val.expire_after(), eval_interval: flow_val.eval_interval(), + eval_offset: effective_eval_schedule_from_flow_info(&flow_val) + .map_err(BoxedError::new) + .context(error::QueryExecutionSnafu)? + .map(|schedule| schedule.anchor_secs) + .filter(|anchor_secs| *anchor_secs != 0), comment, flow_options: OptionMap::from_filtered_string_map( flow_val.options(), diff --git a/src/sql/src/parsers/create_parser.rs b/src/sql/src/parsers/create_parser.rs index 0e0ece1b33..b1ea56df2e 100644 --- a/src/sql/src/parsers/create_parser.rs +++ b/src/sql/src/parsers/create_parser.rs @@ -325,11 +325,55 @@ impl<'a> ParserContext<'a> { None }; - let eval_interval = if self - .parser - .consume_tokens(&[Token::make_keyword("EVAL"), Token::make_keyword("INTERVAL")]) - { - Some(self.parse_interval_no_month("EVAL INTERVAL")?) + let (eval_interval, eval_interval_has_fractional_secs) = + if self.consume_eval_pair("INTERVAL") { + let (secs, has_fractional) = + self.parse_interval_no_month_whole_secs("EVAL INTERVAL")?; + (Some(secs), has_fractional) + } else { + (None, false) + }; + + // `EVAL OFFSET` is only legal together with `EVAL INTERVAL`. The phase + // offset must be a whole number of seconds; when an offset is present + // the interval must also be whole seconds (never silently truncated). + let eval_offset = if self.consume_eval_pair("OFFSET") { + let Some(eval_interval) = eval_interval else { + return InvalidIntervalSnafu { + reason: "EVAL OFFSET requires EVAL INTERVAL to be specified".to_string(), + } + .fail(); + }; + let (offset_secs, has_fractional) = + self.parse_interval_no_month_whole_secs("EVAL OFFSET")?; + if has_fractional { + return InvalidIntervalSnafu { + reason: "EVAL OFFSET must be a whole number of seconds".to_string(), + } + .fail(); + } + if eval_interval_has_fractional_secs { + return InvalidIntervalSnafu { + reason: "EVAL INTERVAL must be a whole number of seconds when EVAL OFFSET is specified" + .to_string(), + } + .fail(); + } + if !(0..eval_interval).contains(&offset_secs) { + return InvalidIntervalSnafu { + reason: format!( + "EVAL OFFSET must be in range [0, EVAL INTERVAL), got {offset_secs} seconds with EVAL INTERVAL {eval_interval} seconds" + ), + } + .fail(); + } + // Canonicalize a zero offset to `None` (the default epoch-anchored + // schedule) so that parse/display/reparse round-trips are stable. + if offset_secs == 0 { + None + } else { + Some(offset_secs) + } } else { None }; @@ -372,6 +416,7 @@ impl<'a> ParserContext<'a> { if_not_exists, expire_after, eval_interval, + eval_offset, comment, flow_options: flow_option_map(flow_options), query, @@ -437,6 +482,15 @@ impl<'a> ParserContext<'a> { /// Parse the interval expr to duration in seconds. fn parse_interval_no_month(&mut self, context: &str) -> Result { + Ok(self.parse_interval_no_month_whole_secs(context)?.0) + } + + /// Parses an interval that must not contain months and returns the total + /// whole seconds together with whether the interval contains a sub-second + /// fraction. Whole seconds are computed by truncating the nanosecond part; + /// callers that require exact whole-second precision (e.g. `EVAL OFFSET`) + /// must reject `has_fractional_secs`. + fn parse_interval_no_month_whole_secs(&mut self, context: &str) -> Result<(i64, bool)> { let interval = self.parse_interval_month_day_nano()?.0; if interval.months != 0 { return InvalidIntervalSnafu { @@ -444,13 +498,30 @@ impl<'a> ParserContext<'a> { } .fail(); } - Ok( - interval.nanoseconds / 1_000_000_000 - + interval.days as i64 * 60 * 60 * 24 - + interval.months as i64 * 60 * 60 * 24 * 3044 / 1000, // 1 month=365.25/12=30.44 days - // this is to keep the same as https://docs.rs/humantime/latest/humantime/fn.parse_duration.html - // which we use in database to parse i.e. ttl interval and many other intervals - ) + let has_fractional_secs = interval.nanoseconds % 1_000_000_000 != 0; + let whole_secs = interval.nanoseconds / 1_000_000_000 + interval.days as i64 * 60 * 60 * 24; + Ok((whole_secs, has_fractional_secs)) + } + + /// Consumes an `EVAL ` token pair case-insensitively. + /// + /// `EVAL` is not a sqlparser keyword, so a plain + /// `consume_tokens([Token::make_keyword("EVAL"), ...])` comparison would be + /// case-sensitive on the word value and reject lower-case `eval interval`. + fn consume_eval_pair(&mut self, second: &str) -> bool { + let matches = matches!( + ( + &self.parser.peek_token().token, + &self.parser.peek_nth_token(1).token, + ), + (Token::Word(w1), Token::Word(w2)) + if w1.value.eq_ignore_ascii_case("EVAL") && w2.value.eq_ignore_ascii_case(second) + ); + if matches { + self.parser.next_token(); + self.parser.next_token(); + } + matches } /// Parse interval expr to [`IntervalMonthDayNano`]. @@ -1721,6 +1792,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;", if_not_exists: expected.if_not_exists, expire_after: expected.expire_after, eval_interval: None, + eval_offset: None, comment: expected.comment, flow_options: expected.flow_options, // ignore query parse result @@ -1766,6 +1838,9 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;", /// Duration in seconds as `i64` /// If not set, flow will be evaluated based on time window size and other args. pub eval_interval: Option, + /// Phase offset of the flow evaluation schedule within `eval_interval`. + /// Duration in seconds as `i64`. + pub eval_offset: Option, /// Comment string pub comment: Option, /// Flow creation options @@ -1792,6 +1867,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;", if_not_exists: true, expire_after: Some(300), eval_interval: None, + eval_offset: None, comment: Some("test comment".to_string()), flow_options: OptionMap::default(), }, @@ -1814,6 +1890,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;", if_not_exists: true, expire_after: Some(300), eval_interval: None, + eval_offset: None, comment: Some("test comment".to_string()), flow_options: OptionMap::default(), }, @@ -1837,6 +1914,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;", if_not_exists: true, expire_after: Some(300), eval_interval: Some(10), + eval_offset: None, comment: Some("test comment".to_string()), flow_options: OptionMap::default(), }, @@ -1860,6 +1938,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;", if_not_exists: true, expire_after: Some(300), eval_interval: Some(10), + eval_offset: None, comment: Some("test comment".to_string()), flow_options: OptionMap::default(), }, @@ -1883,6 +1962,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;", if_not_exists: false, expire_after: Some(2 * 86400 + 3600 + 2 * 60), eval_interval: None, + eval_offset: None, comment: None, flow_options: OptionMap::default(), }, @@ -1905,6 +1985,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;", if_not_exists: false, expire_after: None, eval_interval: Some(10), + eval_offset: None, comment: None, flow_options: string_option_map([ ("defer_on_missing_source", "true"), @@ -1924,6 +2005,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;", if_not_exists: expected.if_not_exists, expire_after: expected.expire_after, eval_interval: expected.eval_interval, + eval_offset: expected.eval_offset, comment: expected.comment, flow_options: expected.flow_options, // ignore query parse result @@ -1937,6 +2019,142 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;", } } + #[test] + fn test_parse_create_flow_with_eval_offset() { + use pretty_assertions::assert_eq; + fn parse_create_flow(sql: &str) -> CreateFlow { + let stmts = ParserContext::create_with_dialect( + sql, + &GreptimeDbDialect {}, + ParseOptions::default(), + ) + .unwrap(); + assert_eq!(1, stmts.len()); + match &stmts[0] { + Statement::CreateFlow(c) => c.clone(), + _ => panic!("{:?}", stmts[0]), + } + } + let sql = r#" +CREATE FLOW task_1 +SINK TO schema_1.table_1 +EVAL INTERVAL '1 hour' +EVAL OFFSET '2 minutes' +AS +SELECT max(c1), min(c2) FROM schema_2.table_2;"#; + let create_task = parse_create_flow(sql); + assert_eq!(create_task.eval_interval, Some(3600)); + assert_eq!(create_task.eval_offset, Some(120)); + let show_create = create_task.to_string(); + assert!( + show_create.contains("EVAL OFFSET '120 s'"), + "unexpected display:\n{show_create}" + ); + let recreated = parse_create_flow(&show_create); + assert_eq!(recreated, create_task, "input sql is:\n{show_create}"); + + let sql = r#" +create flow task_2 +sink to schema_1.table_1 +eval interval '1h' +eval offset '2m' +as +select max(c1), min(c2) from schema_2.table_2;"#; + let create_task = parse_create_flow(sql); + assert_eq!(create_task.eval_interval, Some(3600)); + assert_eq!(create_task.eval_offset, Some(120)); + + // zero offset is canonicalized to `None` and omitted on display + let sql = r#" +CREATE FLOW task_3 +SINK TO schema_1.table_1 +EVAL INTERVAL '1 hour' +EVAL OFFSET '0 seconds' +AS +SELECT max(c1), min(c2) FROM schema_2.table_2;"#; + let create_task = parse_create_flow(sql); + assert_eq!(create_task.eval_interval, Some(3600)); + assert_eq!(create_task.eval_offset, None); + assert!( + !create_task.to_string().contains("EVAL OFFSET"), + "zero offset should be omitted on display" + ); + + let sql = r#" +CREATE FLOW task_4 +SINK TO schema_1.table_1 +EVAL OFFSET '2 minutes' +AS +SELECT max(c1), min(c2) FROM schema_2.table_2;"#; + let err = + ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default()) + .unwrap_err() + .to_string(); + assert!( + err.contains("EVAL OFFSET requires EVAL INTERVAL"), + "unexpected error: {err}" + ); + + for (offset, interval) in [ + ("-1 seconds", "1 hour"), + ("1 hour", "1 hour"), + ("2 hours", "1 hour"), + ] { + let sql = format!( + r#" +CREATE FLOW task_invalid +SINK TO schema_1.table_1 +EVAL INTERVAL '{interval}' +EVAL OFFSET '{offset}' +AS +SELECT max(c1), min(c2) FROM schema_2.table_2;"# + ); + let err = ParserContext::create_with_dialect( + &sql, + &GreptimeDbDialect {}, + ParseOptions::default(), + ) + .unwrap_err() + .to_string(); + assert!( + err.contains("EVAL OFFSET must be in range"), + "unexpected error for offset {offset}: {err}" + ); + } + + let sql = r#" +CREATE FLOW task_fractional_offset +SINK TO schema_1.table_1 +EVAL INTERVAL '1 hour' +EVAL OFFSET '1.5 seconds' +AS +SELECT max(c1), min(c2) FROM schema_2.table_2;"#; + let err = + ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default()) + .unwrap_err() + .to_string(); + assert!( + err.contains("EVAL OFFSET must be a whole number of seconds"), + "unexpected error: {err}" + ); + + let sql = r#" +CREATE FLOW task_fractional_interval +SINK TO schema_1.table_1 +EVAL INTERVAL '1.5 seconds' +EVAL OFFSET '1 second' +AS +SELECT max(c1), min(c2) FROM schema_2.table_2;"#; + let err = + ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default()) + .unwrap_err() + .to_string(); + assert!( + err.contains("EVAL INTERVAL must be a whole number of seconds"), + "unexpected error: {err}" + ); + } + #[test] fn test_parse_create_flow_with_tql_cte_query() { let sql = r#" diff --git a/src/sql/src/statements/create.rs b/src/sql/src/statements/create.rs index 0db002c672..aa39d42c29 100644 --- a/src/sql/src/statements/create.rs +++ b/src/sql/src/statements/create.rs @@ -695,6 +695,12 @@ pub struct CreateFlow { /// Duration in seconds as `i64` /// If not set, flow will be evaluated based on time window size and other args. pub eval_interval: Option, + /// Phase offset of the flow evaluation schedule within `eval_interval`. + /// Duration in seconds as `i64`. + /// Must be in range `[0, eval_interval)`. Only legal together with + /// `eval_interval`. A value of zero (the default) means the schedule is + /// anchored to the Unix epoch, i.e. phases at `k * eval_interval`. + pub eval_offset: Option, /// Comment string pub comment: Option, /// Flow creation options from `WITH (...)` @@ -753,6 +759,13 @@ impl Display for CreateFlow { if let Some(eval_interval) = &self.eval_interval { writeln!(f, "EVAL INTERVAL '{} s'", eval_interval)?; } + // Canonical display: omit a zero offset (equivalent to the default + // epoch-anchored schedule). Non-zero offsets are always emitted. + if let Some(eval_offset) = &self.eval_offset + && *eval_offset != 0 + { + writeln!(f, "EVAL OFFSET '{} s'", eval_offset)?; + } if let Some(comment) = &self.comment { writeln!(f, "COMMENT '{}'", comment)?; } diff --git a/tests/cases/standalone/common/flow/flow_advance_ttl.result b/tests/cases/standalone/common/flow/flow_advance_ttl.result index fd109043a0..6b085390aa 100644 --- a/tests/cases/standalone/common/flow/flow_advance_ttl.result +++ b/tests/cases/standalone/common/flow/flow_advance_ttl.result @@ -8,9 +8,9 @@ CREATE TABLE distinct_basic ( Affected Rows: 0 --- should fallback to streaming mode +-- should fallback to streaming mode when there is no EVAL INTERVAL -- SQLNESS REPLACE id=\d+ id=REDACTED -CREATE FLOW test_distinct_basic SINK TO out_distinct_basic EVAL INTERVAL '1m' AS +CREATE FLOW test_distinct_basic SINK TO out_distinct_basic AS SELECT DISTINCT number as dis FROM @@ -157,6 +157,39 @@ DROP TABLE out_distinct_basic; Affected Rows: 0 +-- test ttl = instant with EVAL INTERVAL must be rejected +-- since the batching scheduler cannot read instant-TTL source tables and the +-- streaming engine cannot honor the schedule +CREATE TABLE distinct_basic ( + "number" INT, + ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY(number), + TIME INDEX(ts) +)WITH ('ttl' = 'instant'); + +Affected Rows: 0 + +-- SQLNESS REPLACE id=\d+ id=REDACTED +CREATE FLOW test_distinct_basic SINK TO out_distinct_basic EVAL INTERVAL '1m' AS +SELECT + DISTINCT number as dis +FROM + distinct_basic; + +Error: 1004(InvalidArguments), Invalid SQL, error: flow 'test_distinct_basic' with EVAL INTERVAL requires the batching scheduler, but source tables with ttl=instant are not supported under batching mode; use a TTL longer than the flush interval + +SELECT count(*) FROM INFORMATION_SCHEMA.FLOWS WHERE flow_name = 'test_distinct_basic'; + ++----------+ +| count(*) | ++----------+ +| 0 | ++----------+ + +DROP TABLE distinct_basic; + +Affected Rows: 0 + -- test ttl = 5s CREATE TABLE distinct_basic ( "number" INT, diff --git a/tests/cases/standalone/common/flow/flow_advance_ttl.sql b/tests/cases/standalone/common/flow/flow_advance_ttl.sql index 7693ce672d..4ebb8873a0 100644 --- a/tests/cases/standalone/common/flow/flow_advance_ttl.sql +++ b/tests/cases/standalone/common/flow/flow_advance_ttl.sql @@ -6,9 +6,9 @@ CREATE TABLE distinct_basic ( TIME INDEX(ts) )WITH ('ttl' = 'instant'); --- should fallback to streaming mode +-- should fallback to streaming mode when there is no EVAL INTERVAL -- SQLNESS REPLACE id=\d+ id=REDACTED -CREATE FLOW test_distinct_basic SINK TO out_distinct_basic EVAL INTERVAL '1m' AS +CREATE FLOW test_distinct_basic SINK TO out_distinct_basic AS SELECT DISTINCT number as dis FROM @@ -63,6 +63,27 @@ DROP FLOW test_distinct_basic; DROP TABLE distinct_basic; DROP TABLE out_distinct_basic; +-- test ttl = instant with EVAL INTERVAL must be rejected +-- since the batching scheduler cannot read instant-TTL source tables and the +-- streaming engine cannot honor the schedule +CREATE TABLE distinct_basic ( + "number" INT, + ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY(number), + TIME INDEX(ts) +)WITH ('ttl' = 'instant'); + +-- SQLNESS REPLACE id=\d+ id=REDACTED +CREATE FLOW test_distinct_basic SINK TO out_distinct_basic EVAL INTERVAL '1m' AS +SELECT + DISTINCT number as dis +FROM + distinct_basic; + +SELECT count(*) FROM INFORMATION_SCHEMA.FLOWS WHERE flow_name = 'test_distinct_basic'; + +DROP TABLE distinct_basic; + -- test ttl = 5s CREATE TABLE distinct_basic ( "number" INT, diff --git a/tests/cases/standalone/common/flow/flow_eval_offset.result b/tests/cases/standalone/common/flow/flow_eval_offset.result new file mode 100644 index 0000000000..5c25a1d25f --- /dev/null +++ b/tests/cases/standalone/common/flow/flow_eval_offset.result @@ -0,0 +1,110 @@ +-- EVAL OFFSET: fixed UTC epoch phase (offset + k * interval), deterministic. +CREATE TABLE eval_offset_input ( + ts TIMESTAMP(3) TIME INDEX, + series STRING, + v DOUBLE, + PRIMARY KEY(series) +); + +Affected Rows: 0 + +-- Scheduled at odd seconds (1 + 2k): phase anchored to the Unix epoch. +CREATE FLOW eval_offset_phase_flow +SINK TO eval_offset_phase_sink +EVAL INTERVAL '2s' +EVAL OFFSET '1s' +AS +SELECT date_trunc('second', now()) AS ts, count(v) AS c +FROM eval_offset_input +GROUP BY date_trunc('second', now()); + +Affected Rows: 0 + +SHOW CREATE FLOW eval_offset_phase_flow; + ++------------------------+------------------------------------------------------------------------------------------------------------------------+ +| Flow | Create Flow | ++------------------------+------------------------------------------------------------------------------------------------------------------------+ +| eval_offset_phase_flow | CREATE FLOW IF NOT EXISTS eval_offset_phase_flow | +| | SINK TO public.eval_offset_phase_sink | +| | EVAL INTERVAL '2 s' | +| | EVAL OFFSET '1 s' | +| | AS SELECT date_trunc('second', now()) AS ts, count(v) AS c FROM eval_offset_input GROUP BY date_trunc('second', now()) | ++------------------------+------------------------------------------------------------------------------------------------------------------------+ + +SELECT flow_definition FROM information_schema.flows WHERE flow_name = 'eval_offset_phase_flow'; + ++------------------------------------------------------------------------------------------------------------------------+ +| flow_definition | ++------------------------------------------------------------------------------------------------------------------------+ +| CREATE FLOW IF NOT EXISTS eval_offset_phase_flow | +| SINK TO public.eval_offset_phase_sink | +| EVAL INTERVAL '2 s' | +| EVAL OFFSET '1 s' | +| AS SELECT date_trunc('second', now()) AS ts, count(v) AS c FROM eval_offset_input GROUP BY date_trunc('second', now()) | ++------------------------------------------------------------------------------------------------------------------------+ + +INSERT INTO eval_offset_input VALUES + (now(), 'a', 1.0), + (now(), 'b', 2.0); + +Affected Rows: 2 + +-- SQLNESS SLEEP 5s +-- Prove the offset phase: every row the offset flow wrote has an odd +-- second-of-minute (1 + 2k is always odd). This is deterministic regardless +-- of the exact wall-clock alignment of the evaluations. +SELECT + count(*) > 0 AS offset_flow_ran, + bool_and(date_part('second', ts)::BIGINT % 2 = 1) AS offset_flow_at_odd_seconds +FROM eval_offset_phase_sink; + ++-----------------+----------------------------+ +| offset_flow_ran | offset_flow_at_odd_seconds | ++-----------------+----------------------------+ +| true | true | ++-----------------+----------------------------+ + +CREATE FLOW invalid_eval_offset_no_interval +SINK TO invalid_eval_offset_sink +EVAL OFFSET '1s' +AS SELECT 1; + +Error: 1004(InvalidArguments), Invalid interval provided: EVAL OFFSET requires EVAL INTERVAL to be specified + +CREATE FLOW invalid_eval_offset_too_large +SINK TO invalid_eval_offset_sink +EVAL INTERVAL '1s' +EVAL OFFSET '2s' +AS SELECT 1; + +Error: 1004(InvalidArguments), Invalid interval provided: EVAL OFFSET must be in range [0, EVAL INTERVAL), got 2 seconds with EVAL INTERVAL 1 seconds + +CREATE FLOW invalid_eval_offset_fractional +SINK TO invalid_eval_offset_sink +EVAL INTERVAL '1s' +EVAL OFFSET '0.5s' +AS SELECT 1; + +Error: 1004(InvalidArguments), Invalid interval provided: EVAL OFFSET must be a whole number of seconds + +CREATE FLOW invalid_eval_offset_fractional_interval +SINK TO invalid_eval_offset_sink +EVAL INTERVAL '1.5s' +EVAL OFFSET '1s' +AS SELECT 1; + +Error: 1004(InvalidArguments), Invalid interval provided: EVAL INTERVAL must be a whole number of seconds when EVAL OFFSET is specified + +DROP FLOW eval_offset_phase_flow; + +Affected Rows: 0 + +DROP TABLE eval_offset_phase_sink; + +Affected Rows: 0 + +DROP TABLE eval_offset_input; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/flow/flow_eval_offset.sql b/tests/cases/standalone/common/flow/flow_eval_offset.sql new file mode 100644 index 0000000000..4bb4f61d07 --- /dev/null +++ b/tests/cases/standalone/common/flow/flow_eval_offset.sql @@ -0,0 +1,63 @@ +-- EVAL OFFSET: fixed UTC epoch phase (offset + k * interval), deterministic. + +CREATE TABLE eval_offset_input ( + ts TIMESTAMP(3) TIME INDEX, + series STRING, + v DOUBLE, + PRIMARY KEY(series) +); + +-- Scheduled at odd seconds (1 + 2k): phase anchored to the Unix epoch. +CREATE FLOW eval_offset_phase_flow +SINK TO eval_offset_phase_sink +EVAL INTERVAL '2s' +EVAL OFFSET '1s' +AS +SELECT date_trunc('second', now()) AS ts, count(v) AS c +FROM eval_offset_input +GROUP BY date_trunc('second', now()); + +SHOW CREATE FLOW eval_offset_phase_flow; + +SELECT flow_definition FROM information_schema.flows WHERE flow_name = 'eval_offset_phase_flow'; + +INSERT INTO eval_offset_input VALUES + (now(), 'a', 1.0), + (now(), 'b', 2.0); + +-- SQLNESS SLEEP 5s + +-- Prove the offset phase: every row the offset flow wrote has an odd +-- second-of-minute (1 + 2k is always odd). This is deterministic regardless +-- of the exact wall-clock alignment of the evaluations. +SELECT + count(*) > 0 AS offset_flow_ran, + bool_and(date_part('second', ts)::BIGINT % 2 = 1) AS offset_flow_at_odd_seconds +FROM eval_offset_phase_sink; + +CREATE FLOW invalid_eval_offset_no_interval +SINK TO invalid_eval_offset_sink +EVAL OFFSET '1s' +AS SELECT 1; + +CREATE FLOW invalid_eval_offset_too_large +SINK TO invalid_eval_offset_sink +EVAL INTERVAL '1s' +EVAL OFFSET '2s' +AS SELECT 1; + +CREATE FLOW invalid_eval_offset_fractional +SINK TO invalid_eval_offset_sink +EVAL INTERVAL '1s' +EVAL OFFSET '0.5s' +AS SELECT 1; + +CREATE FLOW invalid_eval_offset_fractional_interval +SINK TO invalid_eval_offset_sink +EVAL INTERVAL '1.5s' +EVAL OFFSET '1s' +AS SELECT 1; + +DROP FLOW eval_offset_phase_flow; +DROP TABLE eval_offset_phase_sink; +DROP TABLE eval_offset_input; diff --git a/tests/compatibility/cases/flow_eval_offset_persistence/case.toml b/tests/compatibility/cases/flow_eval_offset_persistence/case.toml new file mode 100644 index 0000000000..823fe322d4 --- /dev/null +++ b/tests/compatibility/cases/flow_eval_offset_persistence/case.toml @@ -0,0 +1,8 @@ +name = "flow_eval_offset_persistence" +reason = "Verify a non-zero EVAL OFFSET created by a v1.3+ old stage persists in flow schedule metadata and remains visible after a distributed full restart on a v1.3+ new stage." +introduced_by = "#8878 thread r3911676153" +topologies = ["distributed"] +from_range = [">=v1.3.0"] +to_range = [">=v1.3.0"] +features = ["flow", "table"] +owner = "flow" diff --git a/tests/compatibility/cases/flow_eval_offset_persistence/setup.sql b/tests/compatibility/cases/flow_eval_offset_persistence/setup.sql new file mode 100644 index 0000000000..dd38b9940d --- /dev/null +++ b/tests/compatibility/cases/flow_eval_offset_persistence/setup.sql @@ -0,0 +1,15 @@ +CREATE TABLE compat_eval_offset_input ( + ts TIMESTAMP(3) TIME INDEX, + v DOUBLE +); + +CREATE FLOW compat_eval_offset_flow +SINK TO compat_eval_offset_sink +EVAL INTERVAL '10s' +EVAL OFFSET '3s' +AS +SELECT + date_trunc('second', now()) AS ts, + count(v) AS value_count +FROM compat_eval_offset_input +GROUP BY date_trunc('second', now()); diff --git a/tests/compatibility/cases/flow_eval_offset_persistence/verify.result b/tests/compatibility/cases/flow_eval_offset_persistence/verify.result new file mode 100644 index 0000000000..72828301bd --- /dev/null +++ b/tests/compatibility/cases/flow_eval_offset_persistence/verify.result @@ -0,0 +1,11 @@ +SHOW CREATE FLOW compat_eval_offset_flow; + ++-------------------------+-----------------------------------------------------------------------------------------------------------------------------------------+ +| Flow | Create Flow | ++-------------------------+-----------------------------------------------------------------------------------------------------------------------------------------+ +| compat_eval_offset_flow | CREATE FLOW IF NOT EXISTS compat_eval_offset_flow | +| | SINK TO flow_eval_offset_persistence.compat_eval_offset_sink | +| | EVAL INTERVAL '10 s' | +| | EVAL OFFSET '3 s' | +| | AS SELECT date_trunc('second', now()) AS ts, count(v) AS value_count FROM compat_eval_offset_input GROUP BY date_trunc('second', now()) | ++-------------------------+-----------------------------------------------------------------------------------------------------------------------------------------+ diff --git a/tests/compatibility/cases/flow_eval_offset_persistence/verify.sql b/tests/compatibility/cases/flow_eval_offset_persistence/verify.sql new file mode 100644 index 0000000000..d287dcf396 --- /dev/null +++ b/tests/compatibility/cases/flow_eval_offset_persistence/verify.sql @@ -0,0 +1 @@ +SHOW CREATE FLOW compat_eval_offset_flow;