fix: unquote flow_name in create flow expr

src/operator/src/error.rs

@@ -704,6 +704,13 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Invalid flow name: {name}"))]
+    InvalidFlowName {
+        name: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Empty {} expr", name))]
     EmptyDdlExpr {
         name: String,
@@ -821,6 +828,7 @@ impl ErrorExt for Error {
             | Error::UnsupportedRegionRequest { .. }
             | Error::InvalidTableName { .. }
             | Error::InvalidViewName { .. }
+            | Error::InvalidFlowName { .. }
             | Error::InvalidView { .. }
             | Error::InvalidExpr { .. }
             | Error::AdminFunctionNotFound { .. }

src/operator/src/expr_factory.rs

@@ -38,7 +38,7 @@ use query::sql::{
 use session::context::QueryContextRef;
 use session::table_name::table_idents_to_full_name;
 use snafu::{ensure, OptionExt, ResultExt};
-use sql::ast::ColumnOption;
+use sql::ast::{ColumnOption, ObjectName};
 use sql::statements::alter::{
     AlterDatabase, AlterDatabaseOperation, AlterTable, AlterTableOperation,
 };
@@ -55,8 +55,9 @@ use table::table_reference::TableReference;
 use crate::error::{
     BuildCreateExprOnInsertionSnafu, ColumnDataTypeSnafu, ConvertColumnDefaultConstraintSnafu,
     ConvertIdentifierSnafu, EncodeJsonSnafu, ExternalSnafu, IllegalPrimaryKeysDefSnafu,
-    InferFileTableSchemaSnafu, InvalidSqlSnafu, NotSupportedSnafu, ParseSqlSnafu,
-    PrepareFileTableSnafu, Result, SchemaIncompatibleSnafu, UnrecognizedTableOptionSnafu,
+    InferFileTableSchemaSnafu, InvalidFlowNameSnafu, InvalidSqlSnafu, NotSupportedSnafu,
+    ParseSqlSnafu, PrepareFileTableSnafu, Result, SchemaIncompatibleSnafu,
+    UnrecognizedTableOptionSnafu,
 };
 
 #[derive(Debug, Copy, Clone)]
@@ -731,7 +732,7 @@ pub fn to_create_flow_task_expr(
 
     Ok(CreateFlowExpr {
         catalog_name: query_ctx.current_catalog().to_string(),
-        flow_name: create_flow.flow_name.to_string(),
+        flow_name: sanitize_flow_name(create_flow.flow_name)?,
         source_table_names,
         sink_table_name: Some(sink_table_name),
         or_replace: create_flow.or_replace,
@@ -743,6 +744,14 @@ pub fn to_create_flow_task_expr(
     })
 }
 
+/// sanitize the flow name, remove possible quotes
+fn sanitize_flow_name(flow_name: ObjectName) -> Result<String> {
+    let ident = flow_name.0.first().context(InvalidFlowNameSnafu {
+        name: flow_name.to_string(),
+    })?;
+    Ok(ident.value.clone())
+}
+
 #[cfg(test)]
 mod tests {
     use api::v1::{SetDatabaseOptions, UnsetDatabaseOptions};
@@ -755,6 +764,40 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn test_create_flow_expr() {
+        let sql = r"
+CREATE FLOW `task_2`
+SINK TO schema_1.table_1
+AS
+SELECT max(c1), min(c2) FROM schema_2.table_2;";
+        let stmt =
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
+                .unwrap()
+                .pop()
+                .unwrap();
+
+        let Statement::CreateFlow(create_flow) = stmt else {
+            unreachable!()
+        };
+        let expr = to_create_flow_task_expr(create_flow, &QueryContext::arc()).unwrap();
+
+        let to_dot_sep =
+            |c: TableName| format!("{}.{}.{}", c.catalog_name, c.schema_name, c.table_name);
+        assert_eq!("task_2", expr.flow_name);
+        assert_eq!("greptime", expr.catalog_name);
+        assert_eq!(
+            "greptime.schema_1.table_1",
+            expr.sink_table_name.map(to_dot_sep).unwrap()
+        );
+        assert_eq!(1, expr.source_table_names.len());
+        assert_eq!(
+            "greptime.schema_2.table_2",
+            to_dot_sep(expr.source_table_names[0].clone())
+        );
+        assert_eq!("SELECT max(c1), min(c2) FROM schema_2.table_2", expr.sql);
+    }
+
     #[test]
     fn test_create_to_expr() {
         let sql = "CREATE TABLE monitor (host STRING,ts TIMESTAMP,TIME INDEX (ts),PRIMARY KEY(host)) ENGINE=mito WITH(ttl='3days', write_buffer_size='1024KB');";

src/sql/src/parsers/create_parser.rs

@@ -1306,7 +1306,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;";
 
         // create flow without `OR REPLACE`, `IF NOT EXISTS`, `EXPIRE AFTER` and `COMMENT`
         let sql = r"
-CREATE FLOW task_2
+CREATE FLOW `task_2`
 SINK TO schema_1.table_1
 AS
 SELECT max(c1), min(c2) FROM schema_2.table_2;";
@@ -1322,6 +1322,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;";
         assert!(!create_task.if_not_exists);
         assert!(create_task.expire_after.is_none());
         assert!(create_task.comment.is_none());
+        assert_eq!(create_task.flow_name.to_string(), "`task_2`");
     }
 
     #[test]