feat: Implement SHOW CREATE FLOW

src/common/meta/src/key/flow.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub(crate) mod flow_info;
+pub mod flow_info;
 pub(crate) mod flow_name;
 pub(crate) mod flownode_flow;
 pub(crate) mod table_flow;

src/common/meta/src/key/flow/flow_info.rs

@@ -141,6 +141,26 @@ impl FlowInfoValue {
     pub fn source_table_ids(&self) -> &[TableId] {
         &self.source_table_ids
     }
+
+    pub fn flow_name(&self) -> &String {
+        &self.flow_name
+    }
+
+    pub fn sink_table_name(&self) -> &TableName {
+        &self.sink_table_name
+    }
+
+    pub fn raw_sql(&self) -> &String {
+        &self.raw_sql
+    }
+
+    pub fn expire_after(&self) -> Option<i64> {
+        self.expire_after
+    }
+
+    pub fn comment(&self) -> &String {
+        &self.comment
+    }
 }
 
 pub type FlowInfoManagerRef = Arc<FlowInfoManager>;

src/frontend/src/instance.rs

@@ -514,6 +514,9 @@ pub fn check_permission(
         Statement::ShowCreateTable(stmt) => {
             validate_param(&stmt.table_name, query_ctx)?;
         }
+        Statement::ShowCreateFlow(stmt) => {
+            validate_param(&stmt.flow_name, query_ctx)?;
+        }
         Statement::CreateExternalTable(stmt) => {
             validate_param(&stmt.name, query_ctx)?;
         }

src/operator/src/statement.rs

@@ -236,6 +236,44 @@ impl StatementExecutor {
                 self.show_create_table(table_name, table_ref, query_ctx)
                     .await
             }
+            Statement::ShowCreateFlow(show) => {
+                let obj_name = &show.flow_name;
+                let (catalog_name, flow_name) = match &obj_name.0[..] {
+                    [table] => (query_ctx.current_catalog().to_string(), table.value.clone()),
+                    [catalog, table] => (catalog.value.clone(), table.value.clone()),
+                    _ => {
+                        return InvalidSqlSnafu {
+                            err_msg: format!(
+                "expect flow name to be <schema>.<table> or <table>, actual: {obj_name}",
+            ),
+                        }
+                        .fail()
+                    }
+                };
+
+                let flow_name_val = self
+                    .flow_metadata_manager
+                    .flow_name_manager()
+                    .get(&catalog_name, &flow_name)
+                    .await
+                    .context(error::TableMetadataManagerSnafu)?
+                    .context(error::FlowNotFoundSnafu {
+                        flow_name: &flow_name,
+                    })?;
+
+                let flow_val = self
+                    .flow_metadata_manager
+                    .flow_info_manager()
+                    .get(flow_name_val.flow_id())
+                    .await
+                    .context(error::TableMetadataManagerSnafu)?
+                    .context(error::FlowNotFoundSnafu {
+                        flow_name: &flow_name,
+                    })?;
+
+                self.show_create_flow(obj_name.clone(), flow_val, query_ctx)
+                    .await
+            }
             Statement::SetVariables(set_var) => {
                 let var_name = set_var.variable.to_string().to_uppercase();
                 match var_name.as_str() {

src/operator/src/statement/show.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use common_meta::key::flow::flow_info::FlowInfoValue;
 use common_query::Output;
 use common_telemetry::tracing;
 use partition::manager::PartitionInfo;
@@ -23,6 +24,7 @@ use sql::statements::create::Partitions;
 use sql::statements::show::{
     ShowColumns, ShowDatabases, ShowIndex, ShowKind, ShowTables, ShowVariables,
 };
+use sqlparser::ast::ObjectName;
 use table::table_name::TableName;
 use table::TableRef;
 
@@ -95,6 +97,17 @@ impl StatementExecutor {
             .context(error::ExecuteStatementSnafu)
     }
 
+    #[tracing::instrument(skip_all)]
+    pub async fn show_create_flow(
+        &self,
+        flow_name: ObjectName,
+        flow_val: FlowInfoValue,
+        query_ctx: QueryContextRef,
+    ) -> Result<Output> {
+        query::sql::show_create_flow(flow_name, flow_val, query_ctx)
+            .context(error::ExecuteStatementSnafu)
+    }
+
     #[tracing::instrument(skip_all)]
     pub fn show_variable(&self, stmt: ShowVariables, query_ctx: QueryContextRef) -> Result<Output> {
         query::sql::show_variable(stmt, query_ctx).context(error::ExecuteStatementSnafu)

src/query/Cargo.toml

@@ -58,6 +58,7 @@ regex.workspace = true
 session.workspace = true
 snafu.workspace = true
 sql.workspace = true
+sqlparser.workspace = true
 store-api.workspace = true
 substrait.workspace = true
 table.workspace = true

src/query/src/sql.rs

@@ -31,6 +31,7 @@ use common_datasource::file_format::{infer_schemas, FileFormat, Format};
 use common_datasource::lister::{Lister, Source};
 use common_datasource::object_store::build_backend;
 use common_datasource::util::find_dir_and_filename;
+use common_meta::key::flow::flow_info::FlowInfoValue;
 use common_query::prelude::GREPTIME_TIMESTAMP;
 use common_query::Output;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
@@ -49,10 +50,14 @@ use regex::Regex;
 use session::context::QueryContextRef;
 pub use show_create_table::create_table_stmt;
 use snafu::{ensure, OptionExt, ResultExt};
-use sql::statements::create::Partitions;
+use sql::ast::Ident;
+use sql::error::SyntaxSnafu;
+use sql::statements::create::{CreateFlow, Partitions};
 use sql::statements::show::{
     ShowColumns, ShowDatabases, ShowIndex, ShowKind, ShowTables, ShowVariables,
 };
+use sqlparser::ast::ObjectName;
+use sqlparser::parser::{Parser, ParserOptions};
 use table::requests::{FILE_TABLE_LOCATION_KEY, FILE_TABLE_PATTERN_KEY};
 use table::TableRef;
 
@@ -134,6 +139,13 @@ static SHOW_CREATE_TABLE_OUTPUT_SCHEMA: Lazy<Arc<Schema>> = Lazy::new(|| {
     ]))
 });
 
+static SHOW_CREATE_FLOW_OUTPUT_SCHEMA: Lazy<Arc<Schema>> = Lazy::new(|| {
+    Arc::new(Schema::new(vec![
+        ColumnSchema::new("Flow", ConcreteDataType::string_datatype(), false),
+        ColumnSchema::new("Create Flow", ConcreteDataType::string_datatype(), false),
+    ]))
+});
+
 fn null() -> Expr {
     lit(ScalarValue::Null)
 }
@@ -606,6 +618,54 @@ pub fn show_create_table(
     Ok(Output::new_with_record_batches(records))
 }
 
+pub fn show_create_flow(
+    flow_name: ObjectName,
+    flow_val: FlowInfoValue,
+    query_ctx: QueryContextRef,
+) -> Result<Output> {
+    let mut parser = Parser::new(query_ctx.sql_dialect())
+        .with_options(ParserOptions::new().with_trailing_commas(true))
+        .try_with_sql(flow_val.raw_sql())
+        .context(SyntaxSnafu)
+        .context(error::SqlSnafu)?;
+
+    let query = Box::new(
+        parser
+            .parse_query()
+            .context(SyntaxSnafu)
+            .context(error::SqlSnafu)?,
+    );
+
+    let comment = if flow_val.comment().is_empty() {
+        None
+    } else {
+        Some(flow_val.comment().clone())
+    };
+
+    let stmt = CreateFlow {
+        flow_name,
+        sink_table_name: ObjectName(vec![Ident {
+            value: flow_val.sink_table_name().table_name.clone(),
+            quote_style: None,
+        }]),
+        or_replace: true,
+        if_not_exists: true,
+        expire_after: flow_val.expire_after(),
+        comment,
+        query,
+    };
+
+    let sql = format!("{}", stmt);
+    let columns = vec![
+        Arc::new(StringVector::from(vec![flow_val.flow_name().clone()])) as _,
+        Arc::new(StringVector::from(vec![sql])) as _,
+    ];
+    let records = RecordBatches::try_from_columns(SHOW_CREATE_FLOW_OUTPUT_SCHEMA.clone(), columns)
+        .context(error::CreateRecordBatchSnafu)?;
+
+    Ok(Output::new_with_record_batches(records))
+}
+
 pub fn describe_table(table: TableRef) -> Result<Output> {
     let table_info = table.table_info();
     let columns_schemas = table_info.meta.schema.column_schemas();

src/sql/src/error.rs

@@ -141,6 +141,9 @@ pub enum Error {
     #[snafu(display("Invalid table name: {}", name))]
     InvalidTableName { name: String },
 
+    #[snafu(display("Invalid flow name: {}", name))]
+    InvalidFlowName { name: String },
+
     #[snafu(display("Invalid default constraint, column: {}", column))]
     InvalidDefault {
         column: String,
@@ -274,6 +277,7 @@ impl ErrorExt for Error {
             | InvalidDatabaseOption { .. }
             | ColumnTypeMismatch { .. }
             | InvalidTableName { .. }
+            | InvalidFlowName { .. }
             | InvalidSqlValue { .. }
             | TimestampOverflow { .. }
             | InvalidTableOption { .. }

src/sql/src/parsers/show_parser.rs

@@ -16,11 +16,13 @@ use snafu::{ensure, ResultExt};
 use sqlparser::keywords::Keyword;
 use sqlparser::tokenizer::Token;
 
-use crate::error::{self, InvalidDatabaseNameSnafu, InvalidTableNameSnafu, Result};
+use crate::error::{
+    self, InvalidDatabaseNameSnafu, InvalidFlowNameSnafu, InvalidTableNameSnafu, Result,
+};
 use crate::parser::ParserContext;
 use crate::statements::show::{
-    ShowColumns, ShowCreateTable, ShowDatabases, ShowIndex, ShowKind, ShowStatus, ShowTables,
-    ShowVariables,
+    ShowColumns, ShowCreateFlow, ShowCreateTable, ShowDatabases, ShowIndex, ShowKind, ShowStatus,
+    ShowTables, ShowVariables,
 };
 use crate::statements::statement::Statement;
 
@@ -62,6 +64,8 @@ impl<'a> ParserContext<'a> {
         } else if self.consume_token("CREATE") {
             if self.consume_token("TABLE") {
                 self.parse_show_create_table()
+            } else if self.consume_token("FLOW") {
+                self.parse_show_create_flow()
             } else {
                 self.unsupported(self.peek_token_as_string())
             }
@@ -109,6 +113,24 @@ impl<'a> ParserContext<'a> {
         Ok(Statement::ShowCreateTable(ShowCreateTable { table_name }))
     }
 
+    fn parse_show_create_flow(&mut self) -> Result<Statement> {
+        let raw_flow_name = self
+            .parse_object_name()
+            .with_context(|_| error::UnexpectedSnafu {
+                sql: self.sql,
+                expected: "a flow name",
+                actual: self.peek_token_as_string(),
+            })?;
+        let flow_name = Self::canonicalize_object_name(raw_flow_name);
+        ensure!(
+            !flow_name.0.is_empty(),
+            InvalidFlowNameSnafu {
+                name: flow_name.to_string(),
+            }
+        );
+        Ok(Statement::ShowCreateFlow(ShowCreateFlow { flow_name }))
+    }
+
     fn parse_show_table_name(&mut self) -> Result<String> {
         self.parser.next_token();
         let table_name = self

src/sql/src/statements/show.rs

@@ -132,6 +132,19 @@ impl Display for ShowCreateTable {
     }
 }
 
+/// SQL structure for `SHOW CREATE FLOW`.
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
+pub struct ShowCreateFlow {
+    pub flow_name: ObjectName,
+}
+
+impl Display for ShowCreateFlow {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let flow_name = &self.flow_name;
+        write!(f, r#"SHOW CREATE FLOW {flow_name}"#)
+    }
+}
+
 /// SQL structure for `SHOW VARIABLES xxx`.
 #[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct ShowVariables {
@@ -241,6 +254,35 @@ mod tests {
         .is_err());
     }
 
+    #[test]
+    pub fn test_show_create_flow() {
+        let sql = "SHOW CREATE FLOW test";
+        let stmts: Vec<Statement> =
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
+                .unwrap();
+        assert_eq!(1, stmts.len());
+        assert_matches!(&stmts[0], Statement::ShowCreateFlow { .. });
+        match &stmts[0] {
+            Statement::ShowCreateFlow(show) => {
+                let flow_name = show.flow_name.to_string();
+                assert_eq!(flow_name, "test");
+            }
+            _ => {
+                unreachable!();
+            }
+        }
+    }
+    #[test]
+    pub fn test_show_create_missing_flow() {
+        let sql = "SHOW CREATE FLOW";
+        assert!(ParserContext::create_with_dialect(
+            sql,
+            &GreptimeDbDialect {},
+            ParseOptions::default()
+        )
+        .is_err());
+    }
+
     #[test]
     fn test_display_show_variables() {
         let sql = r"show variables v1;";

src/sql/src/statements/statement.rs

@@ -31,8 +31,8 @@ use crate::statements::insert::Insert;
 use crate::statements::query::Query;
 use crate::statements::set_variables::SetVariables;
 use crate::statements::show::{
-    ShowColumns, ShowCreateTable, ShowDatabases, ShowIndex, ShowKind, ShowStatus, ShowTables,
-    ShowVariables,
+    ShowColumns, ShowCreateFlow, ShowCreateTable, ShowDatabases, ShowIndex, ShowKind, ShowStatus,
+    ShowTables, ShowVariables,
 };
 use crate::statements::tql::Tql;
 use crate::statements::truncate::TruncateTable;
@@ -81,6 +81,8 @@ pub enum Statement {
     ShowIndex(ShowIndex),
     // SHOW CREATE TABLE
     ShowCreateTable(ShowCreateTable),
+    // SHOW CREATE FLOW
+    ShowCreateFlow(ShowCreateFlow),
     // SHOW STATUS
     ShowStatus(ShowStatus),
     // DESCRIBE TABLE
@@ -118,6 +120,7 @@ impl Display for Statement {
             Statement::ShowColumns(s) => s.fmt(f),
             Statement::ShowIndex(s) => s.fmt(f),
             Statement::ShowCreateTable(s) => s.fmt(f),
+            Statement::ShowCreateFlow(s) => s.fmt(f),
             Statement::ShowStatus(s) => s.fmt(f),
             Statement::DescribeTable(s) => s.fmt(f),
             Statement::Explain(s) => s.fmt(f),