feat(flow): add flow statistics and SHOW FLOW STATUS (release slice)

Signed-off-by: discord9 <discord9@163.com>
(cherry picked from commit 523c60c53bdc21fd1b1fa43b857e78de710f0d58)

src/sql/src/parsers/show_parser.rs

@@ -26,8 +26,8 @@ use crate::error::{
 use crate::parser::ParserContext;
 use crate::statements::show::{
     ShowColumns, ShowCreateDatabase, ShowCreateFlow, ShowCreateTable, ShowCreateTableVariant,
-    ShowCreateView, ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion,
-    ShowSearchPath, ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews,
+    ShowCreateView, ShowDatabases, ShowFlowStatus, ShowFlows, ShowIndex, ShowKind, ShowProcessList,
+    ShowRegion, ShowSearchPath, ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews,
 };
 use crate::statements::statement::Statement;
 
@@ -57,6 +57,12 @@ impl ParserContext<'_> {
             self.parse_show_views()
         } else if self.consume_token("FLOWS") {
             self.parse_show_flows()
+        } else if self.consume_token("FLOW") {
+            if self.consume_token("STATUS") {
+                self.parse_show_flow_status()
+            } else {
+                self.unsupported(self.peek_token_as_string())
+            }
         } else if self.matches_keyword(Keyword::CHARSET) {
             self.parser.next_token();
             Ok(Statement::ShowCharset(self.parse_show_kind()?))
@@ -587,6 +593,12 @@ impl ParserContext<'_> {
         Ok(Statement::ShowFlows(ShowFlows { kind, database }))
     }
 
+    fn parse_show_flow_status(&mut self) -> Result<Statement> {
+        let kind = self.parse_show_kind()?;
+
+        Ok(Statement::ShowFlowStatus(ShowFlowStatus { kind }))
+    }
+
     fn parse_show_processlist(&mut self, full: bool) -> Result<Statement> {
         match self.parser.next_token().token {
             Token::EOF | Token::SemiColon => {
@@ -1250,6 +1262,22 @@ mod tests {
         assert_eq!(sql, stmts[0].to_string());
     }
 
+    #[test]
+    pub fn test_show_flow_status() {
+        let sql = "SHOW FLOW STATUS";
+        let result =
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
+        let stmts = result.unwrap();
+        assert_eq!(1, stmts.len());
+        assert_eq!(
+            stmts[0],
+            Statement::ShowFlowStatus(ShowFlowStatus {
+                kind: ShowKind::All,
+            })
+        );
+        assert_eq!(sql, stmts[0].to_string());
+    }
+
     #[test]
     pub fn test_show_processlist() {
         let sql = "SHOW PROCESSLIST";

src/sql/src/statements/show.rs

@@ -256,6 +256,21 @@ impl Display for ShowFlows {
     }
 }
 
+/// SQL structure for `SHOW FLOW STATUS`.
+#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut, Serialize)]
+pub struct ShowFlowStatus {
+    pub kind: ShowKind,
+}
+
+impl Display for ShowFlowStatus {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "SHOW FLOW STATUS")?;
+        format_kind!(self, f);
+
+        Ok(())
+    }
+}
+
 /// SQL structure for `SHOW CREATE VIEW`.
 #[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut, Serialize)]
 pub struct ShowCreateView {

src/sql/src/statements/statement.rs

@@ -40,8 +40,8 @@ use crate::statements::query::Query;
 use crate::statements::set_variables::SetVariables;
 use crate::statements::show::{
     ShowColumns, ShowCreateDatabase, ShowCreateFlow, ShowCreateTable, ShowCreateView,
-    ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion, ShowSearchPath,
-    ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews,
+    ShowDatabases, ShowFlowStatus, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion,
+    ShowSearchPath, ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews,
 };
 use crate::statements::tql::Tql;
 use crate::statements::truncate::TruncateTable;
@@ -118,6 +118,8 @@ pub enum Statement {
     ShowCreateTrigger(crate::statements::show::trigger::ShowCreateTrigger),
     /// SHOW FLOWS
     ShowFlows(ShowFlows),
+    /// SHOW FLOW STATUS
+    ShowFlowStatus(ShowFlowStatus),
     // SHOW TRIGGERS
     #[cfg(feature = "enterprise")]
     ShowTriggers(crate::statements::show::trigger::ShowTriggers),
@@ -178,6 +180,7 @@ impl Statement {
             | Statement::ShowCreateTable(_)
             | Statement::ShowCreateFlow(_)
             | Statement::ShowFlows(_)
+            | Statement::ShowFlowStatus(_)
             | Statement::ShowCreateView(_)
             | Statement::ShowStatus(_)
             | Statement::ShowSearchPath(_)
@@ -268,6 +271,7 @@ impl Display for Statement {
             #[cfg(feature = "enterprise")]
             Statement::ShowCreateTrigger(s) => s.fmt(f),
             Statement::ShowFlows(s) => s.fmt(f),
+            Statement::ShowFlowStatus(s) => s.fmt(f),
             #[cfg(feature = "enterprise")]
             Statement::ShowTriggers(s) => s.fmt(f),
             Statement::ShowCreateDatabase(s) => s.fmt(f),

src/sql/src/util.rs

@@ -335,6 +335,7 @@ fn extract_tables_from_statement(stmt: &Statement, names: &mut HashSet<ObjectNam
         | Statement::ShowCreateDatabase(_)
         | Statement::ShowCreateFlow(_)
         | Statement::ShowFlows(_)
+        | Statement::ShowFlowStatus(_)
         | Statement::ShowStatus(_)
         | Statement::ShowSearchPath(_)
         | Statement::ShowViews(_)