feat(flow): stabilize eval interval scheduling (#8360) (#8378)

* feat(flow): stabilize eval interval scheduling



* fix(flow): satisfy eval schedule clippy



* test(flow): trim eval schedule coverage



* test(flow): cover stable eval scheduling



* fix(flow): reserve scheduled runtime hint



* test(flow): trim sqlness result eof



* fix(flow): harden eval schedule edges



* fix(flow): address scheduled flow review



* fix(flow): clean scheduled config handling



* test(flow): add eval interval compat case



* test(flow): cover show create flow in compat



* fix(flow): drop scheduled time from flow context



* test(flow): assert scheduled now binding



---------

Signed-off-by: discord9 <discord9@163.com>
Signed-off-by: evenyag <realevenyag@gmail.com>
Co-authored-by: discord9 <discord9@163.com>

src/query/src/datafusion.rs

@@ -543,6 +543,19 @@ impl QueryEngine for DatafusionQueryEngine {
             .config_options
             .insert(config_options);
 
+        // Apply scheduled time from query context if present, so that `now()` /
+        // `current_timestamp()` functions evaluate against the logical scheduled time
+        // rather than wall-clock.
+        match crate::options::parse_scheduled_time_datetime(&query_ctx.extensions()) {
+            Ok(Some(scheduled_rt)) => {
+                state.execution_props_mut().query_execution_start_time = Some(scheduled_rt);
+            }
+            Ok(None) => {}
+            Err(err) => {
+                common_telemetry::warn!(err; "Ignoring invalid scheduled time query extension");
+            }
+        }
+
         QueryEngineContext::new(state, query_ctx)
     }
 

src/query/src/options.rs

@@ -14,8 +14,10 @@
 
 use std::collections::HashMap;
 
+use chrono::{DateTime, Utc};
 use common_base::memory_limit::MemoryLimit;
 use serde::{Deserialize, Serialize};
+use session::context::QueryContextRef;
 use store_api::storage::RegionId;
 use table::metadata::TableId;
 
@@ -25,6 +27,10 @@ pub const FLOW_INCREMENTAL_AFTER_SEQS: &str = "flow.incremental_after_seqs";
 pub const FLOW_INCREMENTAL_MODE: &str = "flow.incremental_mode";
 pub const FLOW_RETURN_REGION_SEQ: &str = "flow.return_region_seq";
 pub const FLOW_SINK_TABLE_ID: &str = "flow.sink_table_id";
+/// Flow scheduler binding for the logical time of one scheduled attempt.
+/// Query planning, SQL/TQL parsing, range-select rewrite and DataFusion
+/// execution read this extension so `now()` is stable for the whole attempt.
+pub const FLOW_SCHEDULED_TIME_MILLIS: &str = "flow.scheduled_time_millis";
 /// Enable by default, set to false to explicitly disable.
 pub const QUERY_ENABLE_REMOTE_DYNAMIC_FILTER_PUSHDOWN: &str =
     "query.enable_remote_dynamic_filter_pushdown";
@@ -279,6 +285,52 @@ fn parse_bool(option_name: &str, value: &str) -> Result<bool> {
     }
 }
 
+/// Parse the scheduled time (in milliseconds since Unix epoch) from extensions.
+///
+/// Returns `Ok(None)` if the extension key is absent, `Ok(Some(millis))` on success,
+/// or `Err` if the value is malformed.
+pub fn parse_scheduled_time_millis(extensions: &HashMap<String, String>) -> Result<Option<i64>> {
+    match extensions.get(FLOW_SCHEDULED_TIME_MILLIS) {
+        Some(val) => val.parse::<i64>().map(Some).map_err(|_| {
+            invalid_query_context_extension(format!(
+                "Invalid value for {}: {}",
+                FLOW_SCHEDULED_TIME_MILLIS, val
+            ))
+        }),
+        None => Ok(None),
+    }
+}
+
+/// Parse the scheduled time from extensions into a [`DateTime<Utc>`].
+///
+/// Returns `Ok(None)` if the extension key is absent, `Ok(Some(datetime))` on success,
+/// or `Err` if the value is malformed.  Millis that produce an out-of-range timestamp are
+/// also rejected.
+pub fn parse_scheduled_time_datetime(
+    extensions: &HashMap<String, String>,
+) -> Result<Option<DateTime<Utc>>> {
+    match parse_scheduled_time_millis(extensions)? {
+        Some(millis) => DateTime::from_timestamp_millis(millis)
+            .map(Some)
+            .ok_or_else(|| {
+                invalid_query_context_extension(format!(
+                    "Out-of-range timestamp for {}: {} ms",
+                    FLOW_SCHEDULED_TIME_MILLIS, millis
+                ))
+            }),
+        None => Ok(None),
+    }
+}
+
+/// Convenience helper: extract scheduled time from a [`QueryContextRef`] as a [`DateTime<Utc>`].
+///
+/// Errors are silently swallowed, returning `None`.
+/// Use [`parse_scheduled_time_datetime`] when a `Result` is needed.
+pub fn scheduled_time_from_ctx(query_ctx: &QueryContextRef) -> Option<DateTime<Utc>> {
+    let extensions = query_ctx.extensions();
+    parse_scheduled_time_datetime(&extensions).ok().flatten()
+}
+
 fn invalid_query_context_extension(reason: String) -> Error {
     InvalidQueryContextExtensionSnafu { reason }.build()
 }
@@ -551,4 +603,59 @@ mod flow_extension_tests {
             Some(HashMap::from([(1, 10)]))
         );
     }
+
+    // --- scheduled time helper tests ---
+
+    #[test]
+    fn test_parse_scheduled_time_millis_absent() {
+        let exts = HashMap::new();
+        assert_eq!(parse_scheduled_time_millis(&exts).unwrap(), None);
+    }
+
+    #[test]
+    fn test_parse_scheduled_time_millis_valid() {
+        let exts = HashMap::from([(
+            FLOW_SCHEDULED_TIME_MILLIS.to_string(),
+            "1700000000000".to_string(),
+        )]);
+        assert_eq!(
+            parse_scheduled_time_millis(&exts).unwrap(),
+            Some(1700000000000)
+        );
+    }
+
+    #[test]
+    fn test_parse_scheduled_time_millis_malformed() {
+        let exts = HashMap::from([(
+            FLOW_SCHEDULED_TIME_MILLIS.to_string(),
+            "not-a-number".to_string(),
+        )]);
+        let err = parse_scheduled_time_millis(&exts).unwrap_err();
+        assert!(format!("{err}").contains(FLOW_SCHEDULED_TIME_MILLIS));
+    }
+
+    #[test]
+    fn test_parse_scheduled_time_datetime_valid() {
+        let exts = HashMap::from([(
+            FLOW_SCHEDULED_TIME_MILLIS.to_string(),
+            "1700000000000".to_string(),
+        )]);
+        let dt = parse_scheduled_time_datetime(&exts).unwrap().unwrap();
+        assert_eq!(dt.timestamp_millis(), 1700000000000);
+    }
+
+    #[test]
+    fn test_parse_scheduled_time_datetime_negative_millis() {
+        let exts = HashMap::from([(FLOW_SCHEDULED_TIME_MILLIS.to_string(), "-1".to_string())]);
+        let dt = parse_scheduled_time_datetime(&exts).unwrap().unwrap();
+        assert_eq!(dt.timestamp_millis(), -1);
+    }
+
+    #[test]
+    fn test_parse_scheduled_time_datetime_out_of_range() {
+        // i64::MAX millis is well beyond representable chrono::DateTime range
+        let exts = HashMap::from([(FLOW_SCHEDULED_TIME_MILLIS.to_string(), i64::MAX.to_string())]);
+        let err = parse_scheduled_time_datetime(&exts).unwrap_err();
+        assert!(format!("{err}").contains("Out-of-range"));
+    }
 }

src/query/src/parser.rs

@@ -125,12 +125,17 @@ pub struct QueryLanguageParser {}
 
 impl QueryLanguageParser {
     /// Try to parse SQL with GreptimeDB dialect, return the statement when success.
-    pub fn parse_sql(sql: &str, _query_ctx: &QueryContextRef) -> Result<QueryStatement> {
+    pub fn parse_sql(sql: &str, query_ctx: &QueryContextRef) -> Result<QueryStatement> {
         let _timer = PARSE_SQL_ELAPSED.start_timer();
-        let mut statement =
-            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
-                .map_err(BoxedError::new)
-                .context(QueryParseSnafu { query: sql })?;
+        let scheduled_time =
+            crate::options::parse_scheduled_time_datetime(&query_ctx.extensions())?;
+        let mut statement = ParserContext::create_with_dialect(
+            sql,
+            &GreptimeDbDialect {},
+            ParseOptions { scheduled_time },
+        )
+        .map_err(BoxedError::new)
+        .context(QueryParseSnafu { query: sql })?;
         if statement.len() != 1 {
             MultipleStatementsSnafu {
                 query: sql.to_string(),
@@ -323,7 +328,7 @@ impl Alias {
 
 #[cfg(test)]
 mod test {
-    use session::context::QueryContext;
+    use session::context::{QueryContext, QueryContextBuilder};
 
     use super::*;
 
@@ -338,6 +343,32 @@ mod test {
         assert_eq!("SELECT * FROM t1", sql_stmt.to_string());
     }
 
+    #[test]
+    fn parse_sql_tql_uses_scheduled_time_extension() {
+        let ctx = Arc::new(
+            QueryContextBuilder::default()
+                .set_extension(
+                    crate::options::FLOW_SCHEDULED_TIME_MILLIS.to_string(),
+                    "1700000000000".to_string(),
+                )
+                .build(),
+        );
+        let query = "TQL EVAL (now() - '10 minutes'::interval, now(), '1m') http_requests_total";
+        let stmt = QueryLanguageParser::parse_sql(query, &ctx).unwrap();
+
+        match stmt {
+            QueryStatement::Sql(sql::statements::statement::Statement::Tql(
+                sql::statements::tql::Tql::Eval(eval),
+            )) => {
+                assert_eq!(eval.start, "1699999400");
+                assert_eq!(eval.end, "1700000000");
+                assert_eq!(eval.step, "1m");
+                assert_eq!(eval.query, "http_requests_total");
+            }
+            _ => panic!("Expected TQL eval statement, got {stmt:?}"),
+        }
+    }
+
     #[test]
     fn parse_promql_timestamp() {
         let cases = vec![

src/query/src/planner.rs

@@ -24,6 +24,7 @@ use catalog::table_source::DfTableSourceProvider;
 use common_error::ext::BoxedError;
 use common_telemetry::tracing;
 use datafusion::common::{DFSchema, plan_err};
+use datafusion::execution::SessionStateBuilder;
 use datafusion::execution::context::SessionState;
 use datafusion::sql::planner::PlannerContext;
 use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
@@ -85,6 +86,31 @@ impl DfLogicalPlanner {
         }
     }
 
+    /// Derive a [`SessionState`] whose [`ExecutionProps`] includes
+    /// `query_execution_start_time` if a scheduled time extension is present
+    /// in the query context.
+    fn derive_session_state_with_scheduled_time(
+        &self,
+        query_ctx: &QueryContextRef,
+    ) -> Result<SessionState> {
+        let extensions = query_ctx.extensions();
+        match crate::options::parse_scheduled_time_datetime(&extensions)? {
+            Some(dt) => {
+                let execution_props = self
+                    .session_state
+                    .execution_props()
+                    .clone()
+                    .with_query_execution_start_time(dt);
+                Ok(
+                    SessionStateBuilder::new_from_existing(self.session_state.clone())
+                        .with_execution_props(execution_props)
+                        .build(),
+                )
+            }
+            None => Ok(self.session_state.clone()),
+        }
+    }
+
     /// Basically the same with `explain_to_plan` in DataFusion, but adapted to Greptime's
     /// `plan_sql` to support Greptime Statements.
     async fn explain_to_plan(
@@ -178,15 +204,16 @@ impl DfLogicalPlanner {
             .fail()?;
         }
 
+        let scheduled_state = self.derive_session_state_with_scheduled_time(&query_ctx)?;
         let table_provider = DfTableSourceProvider::new(
             self.engine_state.catalog_manager().clone(),
             self.engine_state.disallow_cross_catalog_query(),
             query_ctx.clone(),
             Arc::new(DefaultPlanDecoder::new(
-                self.session_state.clone(),
+                scheduled_state.clone(),
                 &query_ctx,
             )?),
-            self.session_state
+            scheduled_state
                 .config_options()
                 .sql_parser
                 .enable_ident_normalization,
@@ -194,7 +221,7 @@ impl DfLogicalPlanner {
 
         let context_provider = DfContextProviderAdapter::try_new(
             self.engine_state.clone(),
-            self.session_state.clone(),
+            scheduled_state.clone(),
             Some(&df_stmt),
             query_ctx.clone(),
         )
@@ -230,7 +257,7 @@ impl DfLogicalPlanner {
             .await?;
 
         // Optimize logical plan by extension rules
-        let context = QueryEngineContext::new(self.session_state.clone(), query_ctx);
+        let context = QueryEngineContext::new(scheduled_state, query_ctx);
         let plan = self
             .engine_state
             .optimize_by_extension_rules(plan, &context)?;
@@ -248,9 +275,10 @@ impl DfLogicalPlanner {
         normalize_ident: bool,
         query_ctx: QueryContextRef,
     ) -> Result<DfExpr> {
+        let scheduled_state = self.derive_session_state_with_scheduled_time(&query_ctx)?;
         let context_provider = DfContextProviderAdapter::try_new(
             self.engine_state.clone(),
-            self.session_state.clone(),
+            scheduled_state,
             None,
             query_ctx,
         )
@@ -271,8 +299,9 @@ impl DfLogicalPlanner {
 
     #[tracing::instrument(skip_all)]
     async fn plan_pql(&self, stmt: &EvalStmt, query_ctx: QueryContextRef) -> Result<LogicalPlan> {
+        let scheduled_state = self.derive_session_state_with_scheduled_time(&query_ctx)?;
         let plan_decoder = Arc::new(DefaultPlanDecoder::new(
-            self.session_state.clone(),
+            scheduled_state.clone(),
             &query_ctx,
         )?);
         let table_provider = DfTableSourceProvider::new(
@@ -280,7 +309,7 @@ impl DfLogicalPlanner {
             self.engine_state.disallow_cross_catalog_query(),
             query_ctx.clone(),
             plan_decoder,
-            self.session_state
+            scheduled_state
                 .config_options()
                 .sql_parser
                 .enable_ident_normalization,
@@ -290,7 +319,7 @@ impl DfLogicalPlanner {
             .map_err(BoxedError::new)
             .context(QueryPlanSnafu)?;
 
-        let context = QueryEngineContext::new(self.session_state.clone(), query_ctx);
+        let context = QueryEngineContext::new(scheduled_state, query_ctx);
         Ok(self
             .engine_state
             .optimize_by_extension_rules(plan, &context)?)

src/query/src/range_select/plan_rewrite.rs

@@ -19,6 +19,7 @@ use std::time::Duration;
 use arrow_schema::DataType;
 use async_recursion::async_recursion;
 use catalog::table_source::DfTableSourceProvider;
+use chrono::{DateTime, Utc};
 use common_time::interval::{MS_PER_DAY, NANOS_PER_MILLI};
 use common_time::timestamp::TimeUnit;
 use common_time::{IntervalDayTime, IntervalMonthDayNano, IntervalYearMonth, Timestamp, Timezone};
@@ -121,7 +122,7 @@ fn parse_duration_expr(args: &[Expr], i: usize) -> DFResult<Duration> {
             parse_duration(str).map_err(DataFusionError::Plan)
         }
         Some(expr) => {
-            let ms = evaluate_expr_to_millisecond(args, i, true)?;
+            let ms = evaluate_expr_to_millisecond(args, i, true, None)?;
             if ms <= 0 {
                 return Err(dispose_parse_error(Some(expr)));
             }
@@ -138,14 +139,22 @@ fn parse_duration_expr(args: &[Expr], i: usize) -> DFResult<Duration> {
 /// Output a millisecond timestamp
 ///
 /// if `interval_only==true`, only accept expr with all interval type (case 2 will return a error)
-fn evaluate_expr_to_millisecond(args: &[Expr], i: usize, interval_only: bool) -> DFResult<i64> {
+fn evaluate_expr_to_millisecond(
+    args: &[Expr],
+    i: usize,
+    interval_only: bool,
+    scheduled_time: Option<DateTime<Utc>>,
+) -> DFResult<i64> {
     let Some(expr) = args.get(i) else {
         return Err(dispose_parse_error(None));
     };
     if interval_only && !interval_only_in_expr(expr) {
         return Err(dispose_parse_error(Some(expr)));
     }
-    let info = SimplifyContext::default().with_current_time();
+    let info = match scheduled_time {
+        Some(dt) => SimplifyContext::default().with_query_execution_start_time(Some(dt)),
+        None => SimplifyContext::default().with_current_time(),
+    };
     let simplify_expr = ExprSimplifier::new(info).simplify(expr.clone())?;
     match simplify_expr {
         Expr::Literal(ScalarValue::TimestampNanosecond(ts_nanos, _), _)
@@ -207,13 +216,22 @@ fn evaluate_expr_to_millisecond(args: &[Expr], i: usize, interval_only: bool) ->
 /// 2. Timestamp string: align to specific timestamp
 /// 3. An expr can be evaluated at the logical plan stage (e.g. `now() - INTERVAL '1' day`)
 /// 4. leave empty (as Default Option): align to unix epoch 0 (timezone aware)
-fn parse_align_to(args: &[Expr], i: usize, timezone: Option<&Timezone>) -> DFResult<i64> {
+fn parse_align_to(
+    args: &[Expr],
+    i: usize,
+    timezone: Option<&Timezone>,
+    scheduled_time: Option<DateTime<Utc>>,
+) -> DFResult<i64> {
     let Ok(s) = parse_str_expr(args, i) else {
-        return evaluate_expr_to_millisecond(args, i, false);
+        return evaluate_expr_to_millisecond(args, i, false, scheduled_time);
     };
     let upper = s.to_uppercase();
     match upper.as_str() {
-        "NOW" => return Ok(Timestamp::current_millis().value()),
+        "NOW" => {
+            return Ok(scheduled_time
+                .map(|dt| dt.timestamp_millis())
+                .unwrap_or_else(|| Timestamp::current_millis().value()));
+        }
         // default align to unix epoch 0 (timezone aware)
         "" => return Ok(timezone.map(|tz| tz.local_minus_utc() * 1000).unwrap_or(0)),
         _ => (),
@@ -285,7 +303,15 @@ impl TreeNodeRewriter for RangeExprRewriter<'_> {
                 .map_err(|e| DataFusionError::Plan(e.to_string()))?;
             let by = parse_expr_list(&func.args, 4, byc)?;
             let align = parse_duration_expr(&func.args, byc + 4)?;
-            let align_to = parse_align_to(&func.args, byc + 5, Some(&self.query_ctx.timezone()))?;
+            let scheduled_time =
+                crate::options::parse_scheduled_time_datetime(&self.query_ctx.extensions())
+                    .map_err(|err| DataFusionError::Plan(err.to_string()))?;
+            let align_to = parse_align_to(
+                &func.args,
+                byc + 5,
+                Some(&self.query_ctx.timezone()),
+                scheduled_time,
+            )?;
             let mut data_type = range_expr.get_type(self.input_plan.schema())?;
             let mut need_cast = false;
             let fill = Fill::try_from_str(parse_str_expr(&func.args, 2)?, &data_type)?;
@@ -665,7 +691,7 @@ mod test {
     use datafusion_expr::{BinaryExpr, Literal, Operator};
     use datatypes::prelude::ConcreteDataType;
     use datatypes::schema::{ColumnSchema, Schema};
-    use session::context::QueryContext;
+    use session::context::{QueryContext, QueryContextBuilder};
     use table::metadata::{TableInfoBuilder, TableMetaBuilder};
     use table::table::TableRef;
     use table::test_util::EmptyTable;
@@ -771,6 +797,12 @@ mod test {
         engine.planner().plan(&stmt, QueryContext::arc()).await
     }
 
+    async fn do_query_with_ctx(sql: &str, query_ctx: QueryContextRef) -> Result<LogicalPlan> {
+        let stmt = QueryLanguageParser::parse_sql(sql, &query_ctx).unwrap();
+        let engine = create_test_engine().await;
+        engine.planner().plan(&stmt, query_ctx).await
+    }
+
     async fn do_union_query(sql: &str) -> Result<LogicalPlan> {
         let stmt = QueryLanguageParser::parse_sql(sql, &QueryContext::arc()).unwrap();
         let engine = create_union_test_engine().await;
@@ -782,6 +814,26 @@ mod test {
         assert_eq!(plan.display_indent_schema().to_string(), expected);
     }
 
+    #[tokio::test]
+    async fn range_align_to_now_uses_scheduled_time_extension() {
+        let query_ctx = Arc::new(
+            QueryContextBuilder::default()
+                .set_extension(
+                    crate::options::FLOW_SCHEDULED_TIME_MILLIS.to_string(),
+                    "1700000000123".to_string(),
+                )
+                .build(),
+        );
+        let query = r#"SELECT timestamp, tag_0, tag_1, avg(field_0) RANGE '5m' FROM test ALIGN '1h' TO NOW by (tag_0,tag_1);"#;
+        let plan = do_query_with_ctx(query, query_ctx).await.unwrap();
+
+        assert!(
+            plan.display_indent_schema()
+                .to_string()
+                .contains("align_to=1700000000123ms")
+        );
+    }
+
     #[tokio::test]
     async fn range_no_project() {
         let query = r#"SELECT timestamp, tag_0, tag_1, avg(field_0 + field_1) RANGE '5m' FROM test ALIGN '1h' by (tag_0,tag_1);"#;
@@ -1077,37 +1129,54 @@ mod test {
     fn test_parse_align_to() {
         // test NOW
         let args = vec!["NOW".lit()];
-        let epsinon = parse_align_to(&args, 0, None).unwrap() - Timestamp::current_millis().value();
+        let epsinon =
+            parse_align_to(&args, 0, None, None).unwrap() - Timestamp::current_millis().value();
         assert!(epsinon.abs() < 100);
+        let scheduled_time = DateTime::from_timestamp(1_700_000_000, 123_000_000).unwrap();
+        assert_eq!(
+            scheduled_time.timestamp_millis(),
+            parse_align_to(&args, 0, None, Some(scheduled_time)).unwrap()
+        );
         // test default
         let args = vec!["".lit()];
-        assert_eq!(0, parse_align_to(&args, 0, None).unwrap());
+        assert_eq!(0, parse_align_to(&args, 0, None, None).unwrap());
         // test default with timezone
         let args = vec!["".lit()];
         assert_eq!(
             -36000 * 1000,
-            parse_align_to(&args, 0, Some(&Timezone::from_tz_string("HST").unwrap())).unwrap()
+            parse_align_to(
+                &args,
+                0,
+                Some(&Timezone::from_tz_string("HST").unwrap()),
+                None
+            )
+            .unwrap()
         );
         assert_eq!(
             28800 * 1000,
             parse_align_to(
                 &args,
                 0,
-                Some(&Timezone::from_tz_string("Asia/Shanghai").unwrap())
+                Some(&Timezone::from_tz_string("Asia/Shanghai").unwrap()),
+                None
             )
             .unwrap()
         );
 
         // test Timestamp
         let args = vec!["1970-01-01T00:00:00+08:00".lit()];
-        assert_eq!(parse_align_to(&args, 0, None).unwrap(), -8 * 60 * 60 * 1000);
+        assert_eq!(
+            parse_align_to(&args, 0, None, None).unwrap(),
+            -8 * 60 * 60 * 1000
+        );
         // timezone
         let args = vec!["1970-01-01T00:00:00".lit()];
         assert_eq!(
             parse_align_to(
                 &args,
                 0,
-                Some(&Timezone::from_tz_string("Asia/Shanghai").unwrap())
+                Some(&Timezone::from_tz_string("Asia/Shanghai").unwrap()),
+                None
             )
             .unwrap(),
             -8 * 60 * 60 * 1000
@@ -1122,7 +1191,7 @@ mod test {
                 ScalarValue::IntervalDayTime(Some(IntervalDayTime::new(0, 10).into())).lit(),
             ),
         })];
-        assert_eq!(parse_align_to(&args, 0, None).unwrap(), 20);
+        assert_eq!(parse_align_to(&args, 0, None, None).unwrap(), 20);
     }
 
     #[test]