feat(query): Materialized View support insert values with base table

src/query/ast/src/ast/statements/materialized_view.rs

@@ -0,0 +1,149 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fmt::Display;
+use std::fmt::Formatter;
+
+use databend_common_ast_visit_derive::Walk;
+use databend_common_ast_visit_derive::WalkMut;
+use derive_visitor::Drive;
+use derive_visitor::DriveMut;
+
+use crate::ast::CreateOption;
+use crate::ast::Identifier;
+use crate::ast::Query;
+use crate::ast::ShowLimit;
+use crate::ast::write_comma_separated_list;
+use crate::ast::write_dot_separated_list;
+
+#[derive(Debug, Clone, PartialEq, Drive, DriveMut, Walk, WalkMut)]
+pub struct CreateMaterializedViewStmt {
+    pub create_option: CreateOption,
+    pub catalog: Option<Identifier>,
+    pub database: Option<Identifier>,
+    pub view: Identifier,
+    pub columns: Vec<Identifier>,
+    pub query: Box<Query>,
+}
+
+impl Display for CreateMaterializedViewStmt {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "CREATE ")?;
+        if let CreateOption::CreateOrReplace = self.create_option {
+            write!(f, "OR REPLACE ")?;
+        }
+        write!(f, "MATERIALIZED VIEW ")?;
+        if let CreateOption::CreateIfNotExists = self.create_option {
+            write!(f, "IF NOT EXISTS ")?;
+        }
+        write_dot_separated_list(
+            f,
+            self.catalog
+                .iter()
+                .chain(&self.database)
+                .chain(Some(&self.view)),
+        )?;
+        if !self.columns.is_empty() {
+            write!(f, " (")?;
+            write_comma_separated_list(f, &self.columns)?;
+            write!(f, ")")?;
+        }
+        write!(f, " AS {}", self.query)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut, Walk, WalkMut)]
+pub struct DropMaterializedViewStmt {
+    pub if_exists: bool,
+    pub catalog: Option<Identifier>,
+    pub database: Option<Identifier>,
+    pub view: Identifier,
+}
+
+impl Display for DropMaterializedViewStmt {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "DROP MATERIALIZED VIEW ")?;
+        if self.if_exists {
+            write!(f, "IF EXISTS ")?;
+        }
+        write_dot_separated_list(
+            f,
+            self.catalog
+                .iter()
+                .chain(&self.database)
+                .chain(Some(&self.view)),
+        )
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut, Walk, WalkMut)]
+pub struct RefreshMaterializedViewStmt {
+    pub catalog: Option<Identifier>,
+    pub database: Option<Identifier>,
+    pub view: Identifier,
+}
+
+impl Display for RefreshMaterializedViewStmt {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "REFRESH MATERIALIZED VIEW ")?;
+        write_dot_separated_list(
+            f,
+            self.catalog
+                .iter()
+                .chain(&self.database)
+                .chain(Some(&self.view)),
+        )
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut, Walk, WalkMut)]
+pub struct DescribeMaterializedViewStmt {
+    pub catalog: Option<Identifier>,
+    pub database: Option<Identifier>,
+    pub view: Identifier,
+}
+
+impl Display for DescribeMaterializedViewStmt {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "DESCRIBE MATERIALIZED VIEW ")?;
+        write_dot_separated_list(
+            f,
+            self.catalog
+                .iter()
+                .chain(&self.database)
+                .chain(Some(&self.view)),
+        )
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Drive, DriveMut, Walk, WalkMut)]
+pub struct ShowMaterializedViewsStmt {
+    pub catalog: Option<Identifier>,
+    pub database: Option<Identifier>,
+    pub limit: Option<ShowLimit>,
+}
+
+impl Display for ShowMaterializedViewsStmt {
+    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "SHOW MATERIALIZED VIEWS")?;
+        if self.catalog.is_some() || self.database.is_some() {
+            write!(f, " FROM ")?;
+            write_dot_separated_list(f, self.catalog.iter().chain(&self.database))?;
+        }
+        if let Some(limit) = &self.limit {
+            write!(f, " {limit}")?;
+        }
+        Ok(())
+    }
+}

src/query/ast/src/ast/statements/mod.rs

@@ -30,6 +30,7 @@ mod insert;
 mod insert_multi_table;
 mod kill;
 mod lock;
+mod materialized_view;
 mod merge_into;
 mod network_policy;
 mod notification;
@@ -81,6 +82,7 @@ pub use insert::*;
 pub use insert_multi_table::*;
 pub use kill::*;
 pub use lock::*;
+pub use materialized_view::*;
 pub use merge_into::*;
 pub use network_policy::*;
 pub use notification::*;

src/query/ast/src/ast/statements/statement.rs

@@ -219,6 +219,13 @@ pub enum Statement {
     ShowViews(ShowViewsStmt),
     DescribeView(DescribeViewStmt),
 
+    // Materialized Views
+    CreateMaterializedView(CreateMaterializedViewStmt),
+    DropMaterializedView(DropMaterializedViewStmt),
+    RefreshMaterializedView(RefreshMaterializedViewStmt),
+    DescribeMaterializedView(DescribeMaterializedViewStmt),
+    ShowMaterializedViews(ShowMaterializedViewsStmt),
+
     // Streams
     CreateStream(CreateStreamStmt),
     DropStream(DropStreamStmt),
@@ -554,6 +561,8 @@ impl Statement {
             | Statement::ShowColumns(..)
             | Statement::ShowViews(..)
             | Statement::DescribeView(..)
+            | Statement::ShowMaterializedViews(..)
+            | Statement::DescribeMaterializedView(..)
             | Statement::ShowStreams(..)
             | Statement::DescribeStream(..)
             | Statement::RefreshIndex(..)
@@ -619,6 +628,9 @@ impl Statement {
             | Statement::DropDatabase(..)
             | Statement::DropTable(..)
             | Statement::DropView(..)
+            | Statement::DropMaterializedView(..)
+            | Statement::CreateMaterializedView(..)
+            | Statement::RefreshMaterializedView(..)
             | Statement::DropIndex(..)
             | Statement::DropSequence(..)
             | Statement::DropDictionary(..)
@@ -932,6 +944,11 @@ impl Display for Statement {
             Statement::DropView(stmt) => write!(f, "{stmt}")?,
             Statement::ShowViews(stmt) => write!(f, "{stmt}")?,
             Statement::DescribeView(stmt) => write!(f, "{stmt}")?,
+            Statement::CreateMaterializedView(stmt) => write!(f, "{stmt}")?,
+            Statement::DropMaterializedView(stmt) => write!(f, "{stmt}")?,
+            Statement::RefreshMaterializedView(stmt) => write!(f, "{stmt}")?,
+            Statement::DescribeMaterializedView(stmt) => write!(f, "{stmt}")?,
+            Statement::ShowMaterializedViews(stmt) => write!(f, "{stmt}")?,
             Statement::CreateStream(stmt) => write!(f, "{stmt}")?,
             Statement::DropStream(stmt) => write!(f, "{stmt}")?,
             Statement::ShowStreams(stmt) => write!(f, "{stmt}")?,

src/query/ast/src/ast/statements/table.rs

@@ -879,6 +879,7 @@ pub enum Engine {
     Null,
     Memory,
     Fuse,
+    MaterializedView,
     View,
     Random,
     Iceberg,
@@ -893,6 +894,7 @@ impl Display for Engine {
             Engine::Null => write!(f, "NULL"),
             Engine::Memory => write!(f, "MEMORY"),
             Engine::Fuse => write!(f, "FUSE"),
+            Engine::MaterializedView => write!(f, "MATERIALIZED_VIEW"),
             Engine::View => write!(f, "VIEW"),
             Engine::Random => write!(f, "RANDOM"),
             Engine::Iceberg => write!(f, "ICEBERG"),
@@ -909,6 +911,7 @@ impl From<&str> for Engine {
             "null" => Engine::Null,
             "memory" => Engine::Memory,
             "fuse" => Engine::Fuse,
+            "materialized_view" => Engine::MaterializedView,
             "view" => Engine::View,
             "random" => Engine::Random,
             "iceberg" => Engine::Iceberg,

src/query/ast/src/parser/statement.rs

@@ -1613,6 +1613,95 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
         },
     );
 
+    let create_materialized_view = map_res(
+        rule! {
+            CREATE ~ ( OR ~ ^REPLACE )? ~ MATERIALIZED ~ ^VIEW ~ ( IF ~ ^NOT ~ ^EXISTS )?
+            ~ #dot_separated_idents_1_to_3
+            ~ ( "(" ~ #comma_separated_list1(ident) ~ ")" )?
+            ~ AS ~ #query
+        },
+        |(
+            _,
+            opt_or_replace,
+            _,
+            _,
+            opt_if_not_exists,
+            (catalog, database, view),
+            opt_columns,
+            _,
+            query,
+        )| {
+            let create_option =
+                parse_create_option(opt_or_replace.is_some(), opt_if_not_exists.is_some())?;
+            Ok(Statement::CreateMaterializedView(
+                CreateMaterializedViewStmt {
+                    create_option,
+                    catalog,
+                    database,
+                    view,
+                    columns: opt_columns
+                        .map(|(_, columns, _)| columns)
+                        .unwrap_or_default(),
+                    query: Box::new(query),
+                },
+            ))
+        },
+    );
+    let drop_materialized_view = map(
+        rule! {
+            DROP ~ MATERIALIZED ~ ^VIEW ~ ( IF ~ ^EXISTS )? ~ #dot_separated_idents_1_to_3
+        },
+        |(_, _, _, opt_if_exists, (catalog, database, view))| {
+            Statement::DropMaterializedView(DropMaterializedViewStmt {
+                if_exists: opt_if_exists.is_some(),
+                catalog,
+                database,
+                view,
+            })
+        },
+    );
+    let refresh_materialized_view = map(
+        rule! {
+            REFRESH ~ MATERIALIZED ~ ^VIEW ~ #dot_separated_idents_1_to_3
+        },
+        |(_, _, _, (catalog, database, view))| {
+            Statement::RefreshMaterializedView(RefreshMaterializedViewStmt {
+                catalog,
+                database,
+                view,
+            })
+        },
+    );
+    let describe_materialized_view = map(
+        rule! {
+            ( DESC | DESCRIBE ) ~ MATERIALIZED ~ ^VIEW ~ #dot_separated_idents_1_to_3
+        },
+        |(_, _, _, (catalog, database, view))| {
+            Statement::DescribeMaterializedView(DescribeMaterializedViewStmt {
+                catalog,
+                database,
+                view,
+            })
+        },
+    );
+    let show_materialized_views = map(
+        rule! {
+            SHOW ~ MATERIALIZED ~ ^VIEWS ~ ( ( FROM | IN ) ~ #dot_separated_idents_1_to_2 )? ~ #show_limit?
+        },
+        |(_, _, _, ctl_db, limit)| {
+            let (catalog, database) = match ctl_db {
+                Some((_, (Some(c), d))) => (Some(c), Some(d)),
+                Some((_, (None, d))) => (None, Some(d)),
+                _ => (None, None),
+            };
+            Statement::ShowMaterializedViews(ShowMaterializedViewsStmt {
+                catalog,
+                database,
+                limit,
+            })
+        },
+    );
+
     let create_index = map_res(
         rule! {
             CREATE
@@ -2913,6 +3002,7 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
                 | #show_tables_status : "`SHOW TABLES STATUS [FROM <database>] [<show_limit>]`"
                 | #show_drop_tables_status : "`SHOW DROP TABLES [FROM <database>]`"
                 | #show_views : "`SHOW [FULL] VIEWS [FROM <database>] [<show_limit>]`"
+                | #show_materialized_views : "`SHOW MATERIALIZED VIEWS [FROM [<catalog>.]<database>] [<show_limit>]`"
                 | #show_virtual_columns : "`SHOW VIRTUAL COLUMNS FROM <table> [FROM|IN <catalog>.<database>] [<show_limit>]`"
             )
             | (
@@ -2987,7 +3077,8 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
         ATTACH => rule!(#attach_table : "`ATTACH TABLE [<database>.]<table> <uri>`"
             ).parse(i),
         REFRESH => rule!(
-            #refresh_index: "`REFRESH <index_type> INDEX <index> [LIMIT <limit>]`"
+            #refresh_materialized_view: "`REFRESH MATERIALIZED VIEW [<database>.]<view>`"
+            | #refresh_index: "`REFRESH <index_type> INDEX <index> [LIMIT <limit>]`"
             | #refresh_table_index: "`REFRESH <index_type> INDEX <index> ON [<database>.]<table> [LIMIT <limit>]`"
             | #refresh_virtual_column: "`REFRESH VIRTUAL COLUMN FOR [<database>.]<table>`"
         ).parse(i),
@@ -3015,6 +3106,7 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
         DESC | DESCRIBE => rule!(
             #desc_task : "`DESC | DESCRIBE TASK <name>`"
             | #describe_view : "`DESCRIBE VIEW [<database>.]<view>`"
+            | #describe_materialized_view : "`DESCRIBE MATERIALIZED VIEW [<database>.]<view>`"
             | #describe_user: "`DESCRIBE USER <user_name>`"
             | #describe_row_access : "`DESC[RIBE] ROW ACCESS POLICY <name>`"
             | #desc_stage: "`DESC STAGE <stage_name>`"
@@ -3049,6 +3141,7 @@ AS
                 | #create_table : "`CREATE [OR REPLACE] TABLE [IF NOT EXISTS] [<database>.]<table> [<source>] [<table_options>]`"
                 | #create_dictionary : "`CREATE [OR REPLACE] DICTIONARY [IF NOT EXISTS] <dictionary_name> [(<column>, ...)] PRIMARY KEY [<primary_key>, ...] SOURCE (<source_name> ([<source_options>])) [COMMENT <comment>] `"
                 | #create_view : "`CREATE [OR REPLACE] VIEW [IF NOT EXISTS] [<database>.]<view> [(<column>, ...)] AS SELECT ...`"
+                | #create_materialized_view : "`CREATE [OR REPLACE] MATERIALIZED VIEW [IF NOT EXISTS] [<database>.]<view> [(<column>, ...)] AS SELECT ...`"
                 | #create_index: "`CREATE [OR REPLACE] AGGREGATING INDEX [IF NOT EXISTS] <index> AS SELECT ...`"
                 | #create_table_index: "`CREATE [OR REPLACE] <index_type> INDEX [IF NOT EXISTS] <index> ON [<database>.]<table>(<column>, ...)`"
             )
@@ -3097,6 +3190,7 @@ AS
                 | #drop_table : "`DROP TABLE [IF EXISTS] [<database>.]<table>`"
                 | #drop_dictionary : "`DROP DICTIONARY [IF EXISTS] <dictionary_name>`"
                 | #drop_view : "`DROP VIEW [IF EXISTS] [<database>.]<view>`"
+                | #drop_materialized_view : "`DROP MATERIALIZED VIEW [IF EXISTS] [<database>.]<view>`"
                 | #drop_index: "`DROP <index_type> INDEX [IF EXISTS] <index>`"
                 | #drop_table_index: "`DROP <index_type> INDEX [IF EXISTS] <index> ON [<database>.]<table>`"
             )