feat(parser): support emit_mode (#7783)

The parser part

Approved-By: BugenZhao
Approved-By: kwannoel

Author: TennyZhuang

src/frontend/src/handler/mod.rs

@@ -316,6 +316,7 @@ pub async fn handle(
 
             with_options: _, // It is put in OptimizerContext
             or_replace,      // not supported
+            emit_mode,
         } => {
             if or_replace {
                 return Err(ErrorCode::NotImplemented(
@@ -324,6 +325,13 @@ pub async fn handle(
                 )
                 .into());
             }
+            if emit_mode == Some(EmitMode::OnWindowClose) {
+                return Err(ErrorCode::NotImplemented(
+                    "CREATE MATERIALIZED VIEW EMIT ON WINDOW CLOSE".to_string(),
+                    None.into(),
+                )
+                .into());
+            }
             if materialized {
                 create_mv::handle_create_mv(handler_args, name, *query, columns).await
             } else {

src/sqlparser/src/ast/mod.rs

@@ -906,6 +906,7 @@ pub enum Statement {
         name: ObjectName,
         columns: Vec<Ident>,
         query: Box<Query>,
+        emit_mode: Option<EmitMode>,
         with_options: Vec<SqlOption>,
     },
     /// CREATE TABLE
@@ -1236,6 +1237,7 @@ impl fmt::Display for Statement {
                 query,
                 materialized,
                 with_options,
+                emit_mode,
             } => {
                 write!(
                     f,
@@ -1244,6 +1246,9 @@ impl fmt::Display for Statement {
                     materialized = if *materialized { "MATERIALIZED " } else { "" },
                     name = name
                 )?;
+                if let Some(emit_mode) = emit_mode {
+                    write!(f, " EMIT {}", emit_mode)?;
+                }
                 if !with_options.is_empty() {
                     write!(f, " WITH ({})", display_comma_separated(with_options))?;
                 }
@@ -1908,6 +1913,22 @@ impl fmt::Display for SqlOption {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum EmitMode {
+    Immediately,
+    OnWindowClose,
+}
+
+impl fmt::Display for EmitMode {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(match self {
+            EmitMode::Immediately => "IMMEDIATELY",
+            EmitMode::OnWindowClose => "ON WINDOW CLOSE",
+        })
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 pub enum TransactionMode {

src/sqlparser/src/keywords.rs

@@ -198,6 +198,7 @@ define_keywords!(
     EACH,
     ELEMENT,
     ELSE,
+    EMIT,
     ENCRYPTED,
     END,
     END_EXEC = "END-EXEC",
@@ -250,6 +251,7 @@ define_keywords!(
     IF,
     IGNORE,
     ILIKE,
+    IMMEDIATELY,
     IMMUTABLE,
     IN,
     INCLUDE,

src/sqlparser/src/parser.rs

@@ -1657,6 +1657,11 @@ impl Parser {
         // ANSI SQL and Postgres support RECURSIVE here, but we don't support it either.
         let name = self.parse_object_name()?;
         let columns = self.parse_parenthesized_column_list(Optional)?;
+        let emit_mode = if materialized {
+            self.parse_emit_mode()?
+        } else {
+            None
+        };
         let with_options = self.parse_options(Keyword::WITH)?;
         self.expect_keyword(Keyword::AS)?;
         let query = Box::new(self.parse_query()?);
@@ -1668,6 +1673,7 @@ impl Parser {
             materialized,
             or_replace,
             with_options,
+            emit_mode,
         })
     }
 
@@ -2204,6 +2210,25 @@ impl Parser {
         Ok(SqlOption { name, value })
     }
 
+    pub fn parse_emit_mode(&mut self) -> Result<Option<EmitMode>, ParserError> {
+        if self.parse_keyword(Keyword::EMIT) {
+            match self.parse_one_of_keywords(&[Keyword::IMMEDIATELY, Keyword::ON]) {
+                Some(Keyword::IMMEDIATELY) => Ok(Some(EmitMode::Immediately)),
+                Some(Keyword::ON) => {
+                    self.expect_keywords(&[Keyword::WINDOW, Keyword::CLOSE])?;
+                    Ok(Some(EmitMode::OnWindowClose))
+                }
+                Some(_) => unreachable!(),
+                None => self.expected(
+                    "IMMEDIATELY or ON WINDOW CLOSE after EMIT",
+                    self.peek_token(),
+                ),
+            }
+        } else {
+            Ok(None)
+        }
+    }
+
     pub fn parse_alter(&mut self) -> Result<Statement, ParserError> {
         if self.parse_keyword(Keyword::TABLE) {
             self.parse_alter_table()

src/sqlparser/tests/sqlparser_common.rs

@@ -2920,13 +2920,15 @@ fn parse_create_view() {
             or_replace,
             materialized,
             with_options,
+            emit_mode,
         } => {
             assert_eq!("myschema.myview", name.to_string());
             assert_eq!(Vec::<Ident>::new(), columns);
             assert_eq!("SELECT foo FROM bar", query.to_string());
             assert!(!materialized);
             assert!(!or_replace);
             assert_eq!(with_options, vec![]);
+            assert_eq!(emit_mode, None);
         }
         _ => unreachable!(),
     }
@@ -2966,13 +2968,15 @@ fn parse_create_view_with_columns() {
             with_options,
             query,
             materialized,
+            emit_mode,
         } => {
             assert_eq!("v", name.to_string());
             assert_eq!(columns, vec![Ident::new("has"), Ident::new("cols")]);
             assert_eq!(with_options, vec![]);
             assert_eq!("SELECT 1, 2", query.to_string());
             assert!(!materialized);
-            assert!(!or_replace)
+            assert!(!or_replace);
+            assert_eq!(emit_mode, None);
         }
         _ => unreachable!(),
     }
@@ -2988,13 +2992,15 @@ fn parse_create_or_replace_view() {
             with_options,
             query,
             materialized,
+            emit_mode,
         } => {
             assert_eq!("v", name.to_string());
             assert_eq!(columns, vec![]);
             assert_eq!(with_options, vec![]);
             assert_eq!("SELECT 1", query.to_string());
             assert!(!materialized);
-            assert!(or_replace)
+            assert!(or_replace);
+            assert_eq!(emit_mode, None);
         }
         _ => unreachable!(),
     }
@@ -3015,13 +3021,15 @@ fn parse_create_or_replace_materialized_view() {
             with_options,
             query,
             materialized,
+            emit_mode,
         } => {
             assert_eq!("v", name.to_string());
             assert_eq!(columns, vec![]);
             assert_eq!(with_options, vec![]);
             assert_eq!("SELECT 1", query.to_string());
             assert!(materialized);
-            assert!(or_replace)
+            assert!(or_replace);
+            assert_eq!(emit_mode, None);
         }
         _ => unreachable!(),
     }
@@ -3038,13 +3046,66 @@ fn parse_create_materialized_view() {
             query,
             materialized,
             with_options,
+            emit_mode,
+        } => {
+            assert_eq!("myschema.myview", name.to_string());
+            assert_eq!(Vec::<Ident>::new(), columns);
+            assert_eq!("SELECT foo FROM bar", query.to_string());
+            assert!(materialized);
+            assert_eq!(with_options, vec![]);
+            assert!(!or_replace);
+            assert_eq!(emit_mode, None);
+        }
+        _ => unreachable!(),
+    }
+}
+
+#[test]
+fn parse_create_materialized_view_emit_immediately() {
+    let sql = "CREATE MATERIALIZED VIEW myschema.myview EMIT IMMEDIATELY AS SELECT foo FROM bar";
+    match verified_stmt(sql) {
+        Statement::CreateView {
+            name,
+            or_replace,
+            columns,
+            query,
+            materialized,
+            with_options,
+            emit_mode,
+        } => {
+            assert_eq!("myschema.myview", name.to_string());
+            assert_eq!(Vec::<Ident>::new(), columns);
+            assert_eq!("SELECT foo FROM bar", query.to_string());
+            assert!(materialized);
+            assert_eq!(with_options, vec![]);
+            assert!(!or_replace);
+            assert_eq!(emit_mode, Some(EmitMode::Immediately));
+        }
+        _ => unreachable!(),
+    }
+}
+
+#[test]
+fn parse_create_materialized_view_emit_on_window_close() {
+    let sql =
+        "CREATE MATERIALIZED VIEW myschema.myview EMIT ON WINDOW CLOSE AS SELECT foo FROM bar";
+    match verified_stmt(sql) {
+        Statement::CreateView {
+            name,
+            or_replace,
+            columns,
+            query,
+            materialized,
+            with_options,
+            emit_mode,
         } => {
             assert_eq!("myschema.myview", name.to_string());
             assert_eq!(Vec::<Ident>::new(), columns);
             assert_eq!("SELECT foo FROM bar", query.to_string());
             assert!(materialized);
             assert_eq!(with_options, vec![]);
             assert!(!or_replace);
+            assert_eq!(emit_mode, Some(EmitMode::OnWindowClose));
         }
         _ => unreachable!(),
     }

src/tests/sqlsmith/src/sql_gen/mod.rs

@@ -203,6 +203,7 @@ impl<'a, R: Rng> SqlGenerator<'a, R> {
             columns: vec![],
             query,
             with_options: vec![],
+            emit_mode: None,
         };
         (mview, table)
     }