feat(flow): support eval schedule offsets (#8878)

* feat(flow): support eval schedule offsets

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(flow): remove redundant schedule assertion

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* refactor(flow): trim eval offset compatibility scope

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* test(flow): trim eval offset edge coverage

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* docs(flow): trim eval offset comment noise

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(flow): address eval offset review feedback

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* test(compat): cover Flow eval offset persistence

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

---------

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

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<i64>,
+        anchor_secs: i64,
+        options: HashMap<String, String>,
+    ) -> 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);
+    }
+}

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<FlowType
 /// The flow option key for creating pending flow metadata when source tables do not exist.
 pub const DEFER_ON_MISSING_SOURCE_KEY: &str = "defer_on_missing_source";
 
+/// Internal transient key used to pass the typed `EVAL OFFSET` (whole seconds)
+/// from the operator to meta through `CreateFlowExpr.flow_options`. Inserted by
+/// the operator only after user option validation; parsed and stripped by
+/// `CreateFlowTask::try_from`. Must never be accepted as a user-provided option
+/// and must never be persisted into `FlowInfoValue.options` or be visible in
+/// user runtime options / SHOW CREATE.
+pub const INTERNAL_EVAL_OFFSET_KEY: &str = "__greptime_internal_eval_offset_secs";
+
 /// Internal transient key used to pass the serialized `FlowScheduleConfig` from
 /// meta to flownode through `CreateRequest.flow_options`. This key must never
 /// be accepted as a user-provided option and must never be persisted into
@@ -515,6 +524,34 @@ pub fn validate_flow_options(flow_task: &CreateFlowTask) -> 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<i64> {
     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<Utc>) -> Result<i64> {
+    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<Utc>` 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<i64> {
+    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<FlowScheduleConfig> {
+) -> Result<Option<FlowScheduleConfig>> {
     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<String, String>) -> HashMap<String, String> {
     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<String, String> = 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();
 

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

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,

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<i64>,
     pub eval_interval_secs: Option<i64>,
+    /// 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<i64>,
     pub comment: String,
     pub sql: String,
     pub flow_options: HashMap<String, String>,
@@ -1321,11 +1328,27 @@ impl TryFrom<PbCreateFlowTask> 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::<i64>().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<PbCreateFlowTask> 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<CreateFlowTask> 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,

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

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<i64> {
         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<i64> {
+    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<i64> {
     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<DueScheduledTimes> {
-    if schedule.interval_secs <= 0 {
-        return None;
-    }
+) -> Result<DueScheduledTimes> {
+    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::<Vec<_>>();
+            // 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<i64> {
     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::<i64>::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]

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<i64, Error> {
+    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<u64, Error> {
+    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<i64, Error> {
+    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

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<Item = (u64, Option<u64>)>,
 ) -> OutputWithMetrics {

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

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<i64>,
+    eval_interval_secs: Option<i64>,
+) -> 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<String, String>,
     has_missing_source_table: bool,
     has_instant_ttl_source_table: bool,
+    force_batching: bool,
 ) -> Result<Option<FlowType>> {
+    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<Output> {
         // 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<Output> {
-        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<i64>,
         query_context: QueryContextRef,
     ) -> Result<SubmitDdlTaskResponse> {
         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<FlowType> {
+        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"));

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

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<i64> {
+        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 <keyword>` 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<i64>,
+            /// Phase offset of the flow evaluation schedule within `eval_interval`.
+            /// Duration in seconds as `i64`.
+            pub eval_offset: Option<i64>,
             /// Comment string
             pub comment: Option<String>,
             /// 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#"

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<i64>,
+    /// 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<i64>,
     /// Comment string
     pub comment: Option<String>,
     /// 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)?;
         }

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,

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,

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
+

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;

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"

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

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

tests/compatibility/cases/flow_eval_offset_persistence/verify.sql

@@ -0,0 +1 @@
+SHOW CREATE FLOW compat_eval_offset_flow;