feat(flow): add eval interval option (#6623)

* feat: add flow eval interval Signed-off-by: discord9 <discord9@163.com> * feat: tql flow must have eval interval Signed-off-by: discord9 <discord9@163.com> * chore: clippy Signed-off-by: discord9 <discord9@163.com> * test: update sqlness Signed-off-by: discord9 <discord9@163.com> * wip Signed-off-by: discord9 <discord9@163.com> * wip Signed-off-by: discord9 <discord9@163.com> * feat: check for now func Signed-off-by: discord9 <discord9@163.com> * refactor: use ms instead Signed-off-by: discord9 <discord9@163.com> * fix: not panic&proper simplifier Signed-off-by: discord9 <discord9@163.com> * test: update to fix Signed-off-by: discord9 <discord9@163.com> * feat: not allow month in interval Signed-off-by: discord9 <discord9@163.com> * test: update remov months Signed-off-by: discord9 <discord9@163.com> * refactor: per review Signed-off-by: discord9 <discord9@163.com> * chore: after rebase fix Signed-off-by: discord9 <discord9@163.com> * feat: use seconds and add to field instead Signed-off-by: discord9 <discord9@163.com> * chore: aft rebase fix Signed-off-by: discord9 <discord9@163.com> * fix: add check for month Signed-off-by: discord9 <discord9@163.com> * chore: fmt Signed-off-by: discord9 <discord9@163.com> * refactor: per review Signed-off-by: discord9 <discord9@163.com> * refactor: rm clone per review Signed-off-by: discord9 <discord9@163.com> * chore: update proto Signed-off-by: discord9 <discord9@163.com> --------- Signed-off-by: discord9 <discord9@163.com>
2026-05-28 10:50:39 +00:00 · 2025-08-27 17:44:32 +08:00
parent 5ef4dd1743
commit 8452a9d579
28 changed files with 951 additions and 128 deletions
--- a/src/sql/src/parser.rs
+++ b/src/sql/src/parser.rs
@@ -170,7 +170,7 @@ impl ParserContext<'_> {
                Keyword::NoKeyword
                    if w.quote_style.is_none() && w.value.to_uppercase() == tql_parser::TQL =>
                {
-                    self.parse_tql()
+                    self.parse_tql(false)
                }

                Keyword::DECLARE => self.parse_declare_cursor(),
--- a/src/sql/src/parsers.rs
+++ b/src/sql/src/parsers.rs
@@ -31,5 +31,5 @@ pub(crate) mod set_var_parser;
 pub(crate) mod show_parser;
 pub(crate) mod tql_parser;
 pub(crate) mod truncate_parser;
-pub(crate) mod utils;
+pub mod utils;
 pub mod with_tql_parser;
--- a/src/sql/src/parsers/create_parser.rs
+++ b/src/sql/src/parsers/create_parser.rs
@@ -39,6 +39,7 @@ use crate::error::{
    SyntaxSnafu, UnexpectedSnafu, UnsupportedSnafu,
 };
 use crate::parser::{ParserContext, FLOW};
+use crate::parsers::tql_parser;
 use crate::parsers::utils::{
    self, validate_column_fulltext_create_option, validate_column_skipping_index_create_option,
 };
@@ -295,7 +296,16 @@ impl<'a> ParserContext<'a> {
            .parser
            .consume_tokens(&[Token::make_keyword(EXPIRE), Token::make_keyword(AFTER)])
        {
-            Some(self.parse_interval()?)
+            Some(self.parse_interval_no_month("EXPIRE AFTER")?)
+        } else {
+            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")?)
        } else {
            None
        };
@@ -321,10 +331,39 @@ impl<'a> ParserContext<'a> {
            .expect_keyword(Keyword::AS)
            .context(SyntaxSnafu)?;

+        let query = Box::new(self.parse_sql_or_tql(true)?);
+
+        Ok(Statement::CreateFlow(CreateFlow {
+            flow_name,
+            sink_table_name: output_table_name,
+            or_replace,
+            if_not_exists,
+            expire_after,
+            eval_interval,
+            comment,
+            query,
+        }))
+    }
+
+    fn parse_sql_or_tql(&mut self, require_now_expr: bool) -> Result<SqlOrTql> {
        let start_loc = self.parser.peek_token().span.start;
        let start_index = location_to_index(self.sql, &start_loc);

-        let query = self.parse_statement()?;
+        // only accept sql or tql
+        let query = match self.parser.peek_token().token {
+            Token::Word(w) => match w.keyword {
+                Keyword::SELECT => self.parse_query(),
+                Keyword::NoKeyword
+                    if w.quote_style.is_none() && w.value.to_uppercase() == tql_parser::TQL =>
+                {
+                    self.parse_tql(require_now_expr)
+                }
+
+                _ => self.unsupported(self.peek_token_as_string()),
+            },
+            _ => self.unsupported(self.peek_token_as_string()),
+        }?;
+
        let end_token = self.parser.peek_token();

        let raw_query = if end_token == Token::EOF {
@@ -335,23 +374,19 @@ impl<'a> ParserContext<'a> {
            &self.sql[start_index..end_index.min(self.sql.len())]
        };
        let raw_query = raw_query.trim_end_matches(";");
-
-        let query = Box::new(SqlOrTql::try_from_statement(query, raw_query)?);
-
-        Ok(Statement::CreateFlow(CreateFlow {
-            flow_name,
-            sink_table_name: output_table_name,
-            or_replace,
-            if_not_exists,
-            expire_after,
-            comment,
-            query,
-        }))
+        let query = SqlOrTql::try_from_statement(query, raw_query)?;
+        Ok(query)
    }

    /// Parse the interval expr to duration in seconds.
-    fn parse_interval(&mut self) -> Result<i64> {
+    fn parse_interval_no_month(&mut self, context: &str) -> Result<i64> {
        let interval = self.parse_interval_month_day_nano()?.0;
+        if interval.months != 0 {
+            return InvalidIntervalSnafu {
+                reason: format!("Interval with months is not allowed in {context}"),
+            }
+            .fail();
+        }
        Ok(
            interval.nanoseconds / 1_000_000_000
                + interval.days as i64 * 60 * 60 * 24
@@ -365,7 +400,7 @@ impl<'a> ParserContext<'a> {
    fn parse_interval_month_day_nano(&mut self) -> Result<(IntervalMonthDayNano, RawIntervalExpr)> {
        let interval_expr = self.parser.parse_expr().context(error::SyntaxSnafu)?;
        let raw_interval_expr = interval_expr.to_string();
-        let interval = utils::parser_expr_to_scalar_value_literal(interval_expr.clone())?
+        let interval = utils::parser_expr_to_scalar_value_literal(interval_expr.clone(), false)?
            .cast_to(&ArrowDataType::Interval(IntervalUnit::MonthDayNano))
            .ok()
            .with_context(|| InvalidIntervalSnafu {
@@ -1113,7 +1148,7 @@ mod tests {
    use common_catalog::consts::FILE_ENGINE;
    use common_error::ext::ErrorExt;
    use sqlparser::ast::ColumnOption::NotNull;
-    use sqlparser::ast::{BinaryOperator, Expr, ObjectName, Value};
+    use sqlparser::ast::{BinaryOperator, Expr, ObjectName, ObjectNamePart, Value};
    use sqlparser::dialect::GenericDialect;
    use sqlparser::tokenizer::Tokenizer;

@@ -1450,7 +1485,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;",
                r"
 CREATE FLOW `task_2`
 SINK TO schema_1.table_1
-EXPIRE AFTER '1 month 2 days 1h 2 min'
+EXPIRE AFTER '2 days 1h 2 min'
 AS
 SELECT max(c1), min(c2) FROM schema_2.table_2;",
                CreateFlowWoutQuery {
@@ -1461,7 +1496,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;",
                    ]),
                    or_replace: false,
                    if_not_exists: false,
-                    expire_after: Some(86400 * 3044 / 1000 + 2 * 86400 + 3600 + 2 * 60),
+                    expire_after: Some(2 * 86400 + 3600 + 2 * 60),
                    comment: None,
                },
            ),
@@ -1476,6 +1511,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;",
                or_replace: expected.or_replace,
                if_not_exists: expected.if_not_exists,
                expire_after: expected.expire_after,
+                eval_interval: None,
                comment: expected.comment,
                // ignore query parse result
                query: create_task.query.clone(),
@@ -1490,35 +1526,176 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;",

    #[test]
    fn test_parse_create_flow() {
-        let sql = r"
+        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]),
+            }
+        }
+        struct CreateFlowWoutQuery {
+            /// Flow name
+            pub flow_name: ObjectName,
+            /// Output (sink) table name
+            pub sink_table_name: ObjectName,
+            /// Whether to replace existing task
+            pub or_replace: bool,
+            /// Create if not exist
+            pub if_not_exists: bool,
+            /// `EXPIRE AFTER`
+            /// Duration in second as `i64`
+            pub expire_after: Option<i64>,
+            /// Duration for flow evaluation interval
+            /// 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>,
+            /// Comment string
+            pub comment: Option<String>,
+        }
+
+        // create flow without `OR REPLACE`, `IF NOT EXISTS`, `EXPIRE AFTER` and `COMMENT`
+        let testcases = vec![
+            (
+                r"
 CREATE OR REPLACE FLOW IF NOT EXISTS task_1
 SINK TO schema_1.table_1
 EXPIRE AFTER INTERVAL '5 minutes'
 COMMENT 'test comment'
 AS
-SELECT max(c1), min(c2) FROM schema_2.table_2;";
-        let stmts =
-            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
-                .unwrap();
-        assert_eq!(1, stmts.len());
-        let create_task = match &stmts[0] {
-            Statement::CreateFlow(c) => c,
-            _ => unreachable!(),
-        };
+SELECT max(c1), min(c2) FROM schema_2.table_2;",
+                CreateFlowWoutQuery {
+                    flow_name: ObjectName(vec![ObjectNamePart::Identifier(Ident::new("task_1"))]),
+                    sink_table_name: ObjectName(vec![
+                        ObjectNamePart::Identifier(Ident::new("schema_1")),
+                        ObjectNamePart::Identifier(Ident::new("table_1")),
+                    ]),
+                    or_replace: true,
+                    if_not_exists: true,
+                    expire_after: Some(300),
+                    eval_interval: None,
+                    comment: Some("test comment".to_string()),
+                },
+            ),
+            (
+                r"
+CREATE OR REPLACE FLOW IF NOT EXISTS task_1
+SINK TO schema_1.table_1
+EXPIRE AFTER INTERVAL '300 s'
+COMMENT 'test comment'
+AS
+SELECT max(c1), min(c2) FROM schema_2.table_2;",
+                CreateFlowWoutQuery {
+                    flow_name: ObjectName(vec![ObjectNamePart::Identifier(Ident::new("task_1"))]),
+                    sink_table_name: ObjectName(vec![
+                        ObjectNamePart::Identifier(Ident::new("schema_1")),
+                        ObjectNamePart::Identifier(Ident::new("table_1")),
+                    ]),
+                    or_replace: true,
+                    if_not_exists: true,
+                    expire_after: Some(300),
+                    eval_interval: None,
+                    comment: Some("test comment".to_string()),
+                },
+            ),
+            (
+                r"
+CREATE OR REPLACE FLOW IF NOT EXISTS task_1
+SINK TO schema_1.table_1
+EXPIRE AFTER '5 minutes'
+EVAL INTERVAL '10 seconds'
+COMMENT 'test comment'
+AS
+SELECT max(c1), min(c2) FROM schema_2.table_2;",
+                CreateFlowWoutQuery {
+                    flow_name: ObjectName(vec![ObjectNamePart::Identifier(Ident::new("task_1"))]),
+                    sink_table_name: ObjectName(vec![
+                        ObjectNamePart::Identifier(Ident::new("schema_1")),
+                        ObjectNamePart::Identifier(Ident::new("table_1")),
+                    ]),
+                    or_replace: true,
+                    if_not_exists: true,
+                    expire_after: Some(300),
+                    eval_interval: Some(10),
+                    comment: Some("test comment".to_string()),
+                },
+            ),
+            (
+                r"
+CREATE OR REPLACE FLOW IF NOT EXISTS task_1
+SINK TO schema_1.table_1
+EXPIRE AFTER '5 minutes'
+EVAL INTERVAL INTERVAL '10 seconds'
+COMMENT 'test comment'
+AS
+SELECT max(c1), min(c2) FROM schema_2.table_2;",
+                CreateFlowWoutQuery {
+                    flow_name: ObjectName(vec![ObjectNamePart::Identifier(Ident::new("task_1"))]),
+                    sink_table_name: ObjectName(vec![
+                        ObjectNamePart::Identifier(Ident::new("schema_1")),
+                        ObjectNamePart::Identifier(Ident::new("table_1")),
+                    ]),
+                    or_replace: true,
+                    if_not_exists: true,
+                    expire_after: Some(300),
+                    eval_interval: Some(10),
+                    comment: Some("test comment".to_string()),
+                },
+            ),
+            (
+                r"
+CREATE FLOW `task_2`
+SINK TO schema_1.table_1
+EXPIRE AFTER '2 days 1h 2 min'
+AS
+SELECT max(c1), min(c2) FROM schema_2.table_2;",
+                CreateFlowWoutQuery {
+                    flow_name: ObjectName(vec![ObjectNamePart::Identifier(Ident::with_quote(
+                        '`', "task_2",
+                    ))]),
+                    sink_table_name: ObjectName(vec![
+                        ObjectNamePart::Identifier(Ident::new("schema_1")),
+                        ObjectNamePart::Identifier(Ident::new("table_1")),
+                    ]),
+                    or_replace: false,
+                    if_not_exists: false,
+                    expire_after: Some(2 * 86400 + 3600 + 2 * 60),
+                    eval_interval: None,
+                    comment: None,
+                },
+            ),
+        ];

-        let expected = CreateFlow {
-            flow_name: vec![Ident::new("task_1")].into(),
-            sink_table_name: vec![Ident::new("schema_1"), Ident::new("table_1")].into(),
-            or_replace: true,
-            if_not_exists: true,
-            expire_after: Some(300),
-            comment: Some("test comment".to_string()),
-            // ignore query parse result
-            query: create_task.query.clone(),
-        };
-        assert_eq!(create_task, &expected);
+        for (sql, expected) in testcases {
+            let create_task = parse_create_flow(sql);

-        // create flow without `OR REPLACE`, `IF NOT EXISTS`, `EXPIRE AFTER` and `COMMENT`
+            let expected = CreateFlow {
+                flow_name: expected.flow_name,
+                sink_table_name: expected.sink_table_name,
+                or_replace: expected.or_replace,
+                if_not_exists: expected.if_not_exists,
+                expire_after: expected.expire_after,
+                eval_interval: expected.eval_interval,
+                comment: expected.comment,
+                // ignore query parse result
+                query: create_task.query.clone(),
+            };
+
+            assert_eq!(create_task, expected, "input sql is:\n{sql}");
+            let show_create = create_task.to_string();
+            let recreated = parse_create_flow(&show_create);
+            assert_eq!(recreated, expected, "input sql is:\n{show_create}");
+        }
+    }
+
+    #[test]
+    fn test_create_flow_no_month() {
        let sql = r"
 CREATE FLOW `task_2`
 SINK TO schema_1.table_1
@@ -1526,21 +1703,15 @@ EXPIRE AFTER '1 month 2 days 1h 2 min'
 AS
 SELECT max(c1), min(c2) FROM schema_2.table_2;";
        let stmts =
-            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
-                .unwrap();
-        assert_eq!(1, stmts.len());
-        let create_task = match &stmts[0] {
-            Statement::CreateFlow(c) => c,
-            _ => unreachable!(),
-        };
-        assert!(!create_task.or_replace);
-        assert!(!create_task.if_not_exists);
-        assert_eq!(
-            create_task.expire_after,
-            Some(86400 * 3044 / 1000 + 2 * 86400 + 3600 + 2 * 60)
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
+
+        assert!(
+            stmts.is_err()
+                && stmts
+                    .unwrap_err()
+                    .to_string()
+                    .contains("Interval with months is not allowed")
        );
-        assert!(create_task.comment.is_none());
-        assert_eq!(create_task.flow_name.to_string(), "`task_2`");
    }

    #[test]
--- a/src/sql/src/parsers/tql_parser.rs
+++ b/src/sql/src/parsers/tql_parser.rs
@@ -44,7 +44,7 @@ use crate::parsers::error::{
 /// - `TQL EXPLAIN [VERBOSE] [FORMAT format] <query>`
 /// - `TQL ANALYZE [VERBOSE] [FORMAT format] <query>`
 impl ParserContext<'_> {
-    pub(crate) fn parse_tql(&mut self) -> Result<Statement> {
+    pub(crate) fn parse_tql(&mut self, require_now_expr: bool) -> Result<Statement> {
        let _ = self.parser.next_token();

        match self.parser.peek_token().token {
@@ -56,7 +56,7 @@ impl ParserContext<'_> {
                        if (uppercase == EVAL || uppercase == EVALUATE)
                            && w.quote_style.is_none() =>
                    {
-                        self.parse_tql_params()
+                        self.parse_tql_params(require_now_expr)
                            .map(|params| Statement::Tql(Tql::Eval(TqlEval::from(params))))
                            .context(error::TQLSyntaxSnafu)
                    }
@@ -67,7 +67,7 @@ impl ParserContext<'_> {
                            let _consume_verbose_token = self.parser.next_token();
                        }
                        let format = self.parse_format_option();
-                        self.parse_tql_params()
+                        self.parse_tql_params(require_now_expr)
                            .map(|mut params| {
                                params.is_verbose = is_verbose;
                                params.format = format;
@@ -82,7 +82,7 @@ impl ParserContext<'_> {
                            let _consume_verbose_token = self.parser.next_token();
                        }
                        let format = self.parse_format_option();
-                        self.parse_tql_params()
+                        self.parse_tql_params(require_now_expr)
                            .map(|mut params| {
                                params.is_verbose = is_verbose;
                                params.format = format;
@@ -97,18 +97,35 @@ impl ParserContext<'_> {
        }
    }

-    fn parse_tql_params(&mut self) -> std::result::Result<TqlParameters, TQLError> {
+    /// `require_now_expr` indicates whether the start&end must contain a `now()` function.
+    fn parse_tql_params(
+        &mut self,
+        require_now_expr: bool,
+    ) -> std::result::Result<TqlParameters, TQLError> {
        let parser = &mut self.parser;
        let (start, end, step, lookback) = match parser.peek_token().token {
            Token::LParen => {
                let _consume_lparen_token = parser.next_token();
-                let start = Self::parse_string_or_number_or_word(parser, &[Token::Comma])?.0;
-                let end = Self::parse_string_or_number_or_word(parser, &[Token::Comma])?.0;
+                let start = Self::parse_string_or_number_or_word(
+                    parser,
+                    &[Token::Comma],
+                    require_now_expr,
+                )?
+                .0;
+                let end = Self::parse_string_or_number_or_word(
+                    parser,
+                    &[Token::Comma],
+                    require_now_expr,
+                )?
+                .0;

-                let (step, delimiter) =
-                    Self::parse_string_or_number_or_word(parser, &[Token::Comma, Token::RParen])?;
+                let (step, delimiter) = Self::parse_string_or_number_or_word(
+                    parser,
+                    &[Token::Comma, Token::RParen],
+                    false,
+                )?;
                let lookback = if delimiter == Token::Comma {
-                    Self::parse_string_or_number_or_word(parser, &[Token::RParen])
+                    Self::parse_string_or_number_or_word(parser, &[Token::RParen], false)
                        .ok()
                        .map(|t| t.0)
                } else {
@@ -168,6 +185,7 @@ impl ParserContext<'_> {
    fn parse_string_or_number_or_word(
        parser: &mut Parser,
        delimiter_tokens: &[Token],
+        require_now_expr: bool,
    ) -> std::result::Result<(String, Token), TQLError> {
        let mut tokens = vec![];

@@ -185,19 +203,30 @@ impl ParserContext<'_> {
            .context(ParserSnafu),
            1 => {
                let value = match tokens[0].clone() {
-                    Token::Number(n, _) => n,
-                    Token::DoubleQuotedString(s) | Token::SingleQuotedString(s) => s,
-                    Token::Word(_) => Self::parse_tokens_to_ts(tokens)?,
+                    Token::Number(n, _) if !require_now_expr => n,
+                    Token::DoubleQuotedString(s) | Token::SingleQuotedString(s)
+                        if !require_now_expr =>
+                    {
+                        s
+                    }
+                    Token::Word(_) => Self::parse_tokens_to_ts(tokens, require_now_expr)?,
                    unexpected => {
-                        return Err(ParserError::ParserError(format!(
-                            "Expected number, string or word, but have {unexpected:?}"
-                        )))
-                        .context(ParserSnafu);
+                        if !require_now_expr {
+                            return Err(ParserError::ParserError(format!(
+                                "Expected number, string or word, but have {unexpected:?}"
+                            )))
+                            .context(ParserSnafu);
+                        } else {
+                            return Err(ParserError::ParserError(format!(
+                                "Expected expression containing `now()`, but have {unexpected:?}"
+                            )))
+                            .context(ParserSnafu);
+                        }
                    }
                };
                Ok(value)
            }
-            _ => Self::parse_tokens_to_ts(tokens),
+            _ => Self::parse_tokens_to_ts(tokens, require_now_expr),
        };
        for token in delimiter_tokens {
            if parser.consume_token(token) {
@@ -211,9 +240,12 @@ impl ParserContext<'_> {
    }

    /// Parse the tokens to seconds and convert to string.
-    fn parse_tokens_to_ts(tokens: Vec<Token>) -> std::result::Result<String, TQLError> {
+    fn parse_tokens_to_ts(
+        tokens: Vec<Token>,
+        require_now_expr: bool,
+    ) -> std::result::Result<String, TQLError> {
        let parser_expr = Self::parse_to_expr(tokens)?;
-        let lit = utils::parser_expr_to_scalar_value_literal(parser_expr)
+        let lit = utils::parser_expr_to_scalar_value_literal(parser_expr, require_now_expr)
            .map_err(Box::new)
            .context(ConvertToLogicalExpressionSnafu)?;

@@ -281,6 +313,65 @@ mod tests {
        result.remove(0)
    }

+    #[test]
+    fn test_require_now_expr() {
+        let sql = "TQL EVAL (now() - now(), now() -  (now() - '10 seconds'::interval), '1s') http_requests_total{environment=~'staging|testing|development',method!='GET'} @ 1609746000 offset 5m";
+
+        let mut parser = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap();
+        let statement = parser.parse_tql(true).unwrap();
+        match statement {
+            Statement::Tql(Tql::Eval(eval)) => {
+                assert_eq!(eval.start, "0");
+                assert_eq!(eval.end, "10");
+                assert_eq!(eval.step, "1s");
+                assert_eq!(eval.lookback, None);
+                assert_eq!(eval.query, "http_requests_total{environment=~'staging|testing|development',method!='GET'} @ 1609746000 offset 5m");
+            }
+            _ => unreachable!(),
+        };
+
+        let sql = "TQL EVAL (0, 15, '1s') http_requests_total{environment=~'staging|testing|development',method!='GET'} @ 1609746000 offset 5m";
+
+        let mut parser = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap();
+        let statement = parser.parse_tql(true);
+        assert!(
+            statement.is_err()
+                && format!("{:?}", statement)
+                    .contains("Expected expression containing `now()`, but have "),
+            "statement: {:?}",
+            statement
+        );
+
+        let sql = "TQL EVAL (now() - now(), now() -  (now() - '10 seconds'::interval), '1s') http_requests_total{environment=~'staging|testing|development',method!='GET'} @ 1609746000 offset 5m";
+
+        let mut parser = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap();
+        let statement = parser.parse_tql(false).unwrap();
+        match statement {
+            Statement::Tql(Tql::Eval(eval)) => {
+                assert_eq!(eval.start, "0");
+                assert_eq!(eval.end, "10");
+                assert_eq!(eval.step, "1s");
+                assert_eq!(eval.lookback, None);
+                assert_eq!(eval.query, "http_requests_total{environment=~'staging|testing|development',method!='GET'} @ 1609746000 offset 5m");
+            }
+            _ => unreachable!(),
+        };
+
+        let sql = "TQL EVAL (0, 15, '1s') http_requests_total{environment=~'staging|testing|development',method!='GET'} @ 1609746000 offset 5m";
+        let mut parser = ParserContext::new(&GreptimeDbDialect {}, sql).unwrap();
+        let statement = parser.parse_tql(false).unwrap();
+        match statement {
+            Statement::Tql(Tql::Eval(eval)) => {
+                assert_eq!(eval.start, "0");
+                assert_eq!(eval.end, "15");
+                assert_eq!(eval.step, "1s");
+                assert_eq!(eval.lookback, None);
+                assert_eq!(eval.query, "http_requests_total{environment=~'staging|testing|development',method!='GET'} @ 1609746000 offset 5m");
+            }
+            _ => unreachable!(),
+        };
+    }
+
    #[test]
    fn test_parse_tql_eval_with_functions() {
        let sql = "TQL EVAL (now() - now(), now() -  (now() - '10 seconds'::interval), '1s') http_requests_total{environment=~'staging|testing|development',method!='GET'} @ 1609746000 offset 5m";
--- a/src/sql/src/parsers/utils.rs
+++ b/src/sql/src/parsers/utils.rs
@@ -20,10 +20,11 @@ use datafusion::error::Result as DfResult;
 use datafusion::execution::context::SessionState;
 use datafusion::execution::SessionStateBuilder;
 use datafusion::optimizer::simplify_expressions::ExprSimplifier;
+use datafusion_common::tree_node::{TreeNode, TreeNodeVisitor};
 use datafusion_common::{DFSchema, ScalarValue};
 use datafusion_expr::execution_props::ExecutionProps;
 use datafusion_expr::simplify::SimplifyContext;
-use datafusion_expr::{AggregateUDF, ScalarUDF, TableSource, WindowUDF};
+use datafusion_expr::{AggregateUDF, Expr, ScalarUDF, TableSource, WindowUDF};
 use datafusion_sql::planner::{ContextProvider, SqlToRel};
 use datafusion_sql::TableReference;
 use datatypes::arrow::datatypes::DataType;
@@ -33,26 +34,105 @@ use datatypes::schema::{
    COLUMN_FULLTEXT_OPT_KEY_GRANULARITY, COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE,
    COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY, COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE,
 };
-use snafu::ResultExt;
+use snafu::{ensure, ResultExt};
+use sqlparser::dialect::Dialect;

 use crate::error::{
-    ConvertToLogicalExpressionSnafu, ParseSqlValueSnafu, Result, SimplificationSnafu,
+    ConvertToLogicalExpressionSnafu, InvalidSqlSnafu, ParseSqlValueSnafu, Result,
+    SimplificationSnafu,
 };
+use crate::parser::{ParseOptions, ParserContext};
+use crate::statements::statement::Statement;
+
+/// Check if the given SQL query is a TQL statement.
+pub fn is_tql(dialect: &dyn Dialect, sql: &str) -> Result<bool> {
+    let stmts = ParserContext::create_with_dialect(sql, dialect, ParseOptions::default())?;
+
+    ensure!(
+        stmts.len() == 1,
+        InvalidSqlSnafu {
+            msg: format!("Expect only one statement, found {}", stmts.len())
+        }
+    );
+    let stmt = &stmts[0];
+    match stmt {
+        Statement::Tql(_) => Ok(true),
+        _ => Ok(false),
+    }
+}

 /// Convert a parser expression to a scalar value. This function will try the
 /// best to resolve and reduce constants. Exprs like `1 + 1` or `now()` can be
 /// handled properly.
-pub fn parser_expr_to_scalar_value_literal(expr: sqlparser::ast::Expr) -> Result<ScalarValue> {
+///
+/// if `require_now_expr` is true, it will ensure that the expression contains a `now()` function.
+/// If the expression does not contain `now()`, it will return an error.
+///
+pub fn parser_expr_to_scalar_value_literal(
+    expr: sqlparser::ast::Expr,
+    require_now_expr: bool,
+) -> Result<ScalarValue> {
    // 1. convert parser expr to logical expr
    let empty_df_schema = DFSchema::empty();
    let logical_expr = SqlToRel::new(&StubContextProvider::default())
        .sql_to_expr(expr.into(), &empty_df_schema, &mut Default::default())
        .context(ConvertToLogicalExpressionSnafu)?;

+    struct FindNow {
+        found: bool,
+    }
+
+    impl TreeNodeVisitor<'_> for FindNow {
+        type Node = Expr;
+        fn f_down(
+            &mut self,
+            node: &Self::Node,
+        ) -> DfResult<datafusion_common::tree_node::TreeNodeRecursion> {
+            if let Expr::ScalarFunction(func) = node {
+                if func.name().to_lowercase() == "now" {
+                    if !func.args.is_empty() {
+                        return Err(datafusion_common::DataFusionError::Plan(
+                            "now() function should not have arguments".to_string(),
+                        ));
+                    }
+                    self.found = true;
+                    return Ok(datafusion_common::tree_node::TreeNodeRecursion::Stop);
+                }
+            }
+            Ok(datafusion_common::tree_node::TreeNodeRecursion::Continue)
+        }
+    }
+
+    if require_now_expr {
+        let have_now = {
+            let mut visitor = FindNow { found: false };
+            logical_expr.visit(&mut visitor).unwrap();
+            visitor.found
+        };
+        if !have_now {
+            return ParseSqlValueSnafu {
+                msg: format!(
+                    "expected now() expression, but not found in {}",
+                    logical_expr
+                ),
+            }
+            .fail();
+        }
+    }
+
    // 2. simplify logical expr
    let execution_props = ExecutionProps::new().with_query_execution_start_time(Utc::now());
-    let info = SimplifyContext::new(&execution_props).with_schema(Arc::new(empty_df_schema));
-    let simplified_expr = ExprSimplifier::new(info)
+    let info =
+        SimplifyContext::new(&execution_props).with_schema(Arc::new(empty_df_schema.clone()));
+
+    let simplifier = ExprSimplifier::new(info);
+
+    // Coerce the logical expression so simplifier can handle it correctly. This is necessary for const eval with possible type mismatch. i.e.: `now() - now() + '15s'::interval` which is `TimestampNanosecond - TimestampNanosecond + IntervalMonthDayNano`.
+    let logical_expr = simplifier
+        .coerce(logical_expr, &empty_df_schema)
+        .context(SimplificationSnafu)?;
+
+    let simplified_expr = simplifier
        .simplify(logical_expr)
        .context(SimplificationSnafu)?;

@@ -177,3 +257,72 @@ pub fn convert_month_day_nano_to_duration(

    Ok(std::time::Duration::new(adjusted_seconds, nanos_remainder))
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use chrono::DateTime;
+    use datafusion::functions::datetime::expr_fn::now;
+    use datafusion_expr::lit;
+    use datatypes::arrow::datatypes::TimestampNanosecondType;
+
+    use super::*;
+
+    /// Keep this test to make sure we are using datafusion's `ExprSimplifier` correctly.
+    #[test]
+    fn test_simplifier() {
+        let now_time = DateTime::from_timestamp(61, 0).unwrap();
+        let lit_now = lit(ScalarValue::new_timestamp::<TimestampNanosecondType>(
+            now_time.timestamp_nanos_opt(),
+            None,
+        ));
+        let testcases = vec![
+            (now(), lit_now),
+            (now() - now(), lit(ScalarValue::DurationNanosecond(Some(0)))),
+            (
+                now() + lit(ScalarValue::new_interval_dt(0, 1500)),
+                lit(ScalarValue::new_timestamp::<TimestampNanosecondType>(
+                    Some(62500000000),
+                    None,
+                )),
+            ),
+            (
+                now() - (now() + lit(ScalarValue::new_interval_dt(0, 1500))),
+                lit(ScalarValue::DurationNanosecond(Some(-1500000000))),
+            ),
+            // this one failed if type is not coerced
+            (
+                now() - now() + lit(ScalarValue::new_interval_dt(0, 1500)),
+                lit(ScalarValue::new_interval_mdn(0, 0, 1500000000)),
+            ),
+            (
+                lit(ScalarValue::new_interval_mdn(
+                    0,
+                    0,
+                    61 * 86400 * 1_000_000_000,
+                )),
+                lit(ScalarValue::new_interval_mdn(
+                    0,
+                    0,
+                    61 * 86400 * 1_000_000_000,
+                )),
+            ),
+        ];
+
+        let execution_props = ExecutionProps::new().with_query_execution_start_time(now_time);
+        let info = SimplifyContext::new(&execution_props).with_schema(Arc::new(DFSchema::empty()));
+
+        let simplifier = ExprSimplifier::new(info);
+        for (expr, expected) in testcases {
+            let expr_name = expr.schema_name().to_string();
+            let expr = simplifier.coerce(expr, &DFSchema::empty()).unwrap();
+
+            let simplified_expr = simplifier.simplify(expr).unwrap();
+            assert_eq!(
+                simplified_expr, expected,
+                "Failed to simplify expression: {expr_name}"
+            );
+        }
+    }
+}
--- a/src/sql/src/statements/create.rs
+++ b/src/sql/src/statements/create.rs
@@ -383,6 +383,10 @@ pub struct CreateFlow {
    /// `EXPIRE AFTER`
    /// Duration in second as `i64`
    pub expire_after: Option<i64>,
+    /// Duration for flow evaluation interval
+    /// 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>,
    /// Comment string
    pub comment: Option<String>,
    /// SQL statement
@@ -436,7 +440,10 @@ impl Display for CreateFlow {
        writeln!(f, "{}", &self.flow_name)?;
        writeln!(f, "SINK TO {}", &self.sink_table_name)?;
        if let Some(expire_after) = &self.expire_after {
-            writeln!(f, "EXPIRE AFTER '{} s' ", expire_after)?;
+            writeln!(f, "EXPIRE AFTER '{} s'", expire_after)?;
+        }
+        if let Some(eval_interval) = &self.eval_interval {
+            writeln!(f, "EVAL INTERVAL '{} s'", eval_interval)?;
        }
        if let Some(comment) = &self.comment {
            writeln!(f, "COMMENT '{}'", comment)?;