feat(query): support alter table/database refresh cache

src/query/ast/src/ast/statements/database.rs

@@ -166,6 +166,9 @@ impl Display for AlterDatabaseStmt {
             AlterDatabaseAction::RenameDatabase { new_db } => {
                 write!(f, " RENAME TO {new_db}")?;
             }
+            AlterDatabaseAction::RefreshDatabaseCache => {
+                write!(f, " REFRESH CACHE")?;
+            }
         }
 
         Ok(())
@@ -175,6 +178,7 @@ impl Display for AlterDatabaseStmt {
 #[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut)]
 pub enum AlterDatabaseAction {
     RenameDatabase { new_db: Identifier },
+    RefreshDatabaseCache,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut)]

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

@@ -465,6 +465,7 @@ pub enum AlterTableAction {
     UnsetOptions {
         targets: Vec<Identifier>,
     },
+    RefreshTableCache,
 }
 
 impl Display for AlterTableAction {
@@ -534,6 +535,9 @@ impl Display for AlterTableAction {
                     write!(f, ")")?;
                 }
             }
+            AlterTableAction::RefreshTableCache => {
+                write!(f, "REFRESH CACHE")?;
+            }
         };
         Ok(())
     }

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

@@ -3702,15 +3702,23 @@ pub fn create_table_source(i: Input) -> IResult<CreateTableSource> {
 }
 
 pub fn alter_database_action(i: Input) -> IResult<AlterDatabaseAction> {
-    let mut rename_database = map(
+    let rename_database = map(
         rule! {
             RENAME ~ TO ~ #ident
         },
         |(_, _, new_db)| AlterDatabaseAction::RenameDatabase { new_db },
     );
 
+    let refresh_cache = map(
+        rule! {
+            REFRESH ~ CACHE
+        },
+        |(_, _)| AlterDatabaseAction::RefreshDatabaseCache,
+    );
+
     rule!(
         #rename_database
+        | #refresh_cache
     )(i)
 }
 
@@ -3922,6 +3930,13 @@ pub fn alter_table_action(i: Input) -> IResult<AlterTableAction> {
         |(_, _, targets)| AlterTableAction::UnsetOptions { targets },
     );
 
+    let refresh_cache = map(
+        rule! {
+            REFRESH ~ CACHE
+        },
+        |(_, _)| AlterTableAction::RefreshTableCache,
+    );
+
     rule!(
         #alter_table_cluster_key
         | #drop_table_cluster_key
@@ -3935,6 +3950,7 @@ pub fn alter_table_action(i: Input) -> IResult<AlterTableAction> {
         | #revert_table
         | #set_table_options
         | #unset_table_options
+        | #refresh_cache
     )(i)
 }
 

src/query/ast/src/parser/token.rs

@@ -459,6 +459,8 @@ pub enum TokenKind {
     CHAR,
     #[token("COLUMN", ignore(ascii_case))]
     COLUMN,
+    #[token("CACHE", ignore(ascii_case))]
+    CACHE,
     #[token("COLUMN_MATCH_MODE", ignore(ascii_case))]
     COLUMN_MATCH_MODE,
     #[token("COLUMNS", ignore(ascii_case))]

src/query/ast/tests/it/parser.rs

@@ -269,6 +269,7 @@ fn test_statement() {
         r#"OPTIMIZE TABLE t PURGE BEFORE (SNAPSHOT => '9828b23f74664ff3806f44bbc1925ea5') LIMIT 10;"#,
         r#"OPTIMIZE TABLE t PURGE BEFORE (TIMESTAMP => '2023-06-26 09:49:02.038483'::TIMESTAMP) LIMIT 10;"#,
         r#"ALTER TABLE t CLUSTER BY(c1);"#,
+        r#"ALTER TABLE t refresh cache;"#,
         r#"ALTER TABLE t COMMENT='t1-commnet';"#,
         r#"ALTER TABLE t DROP CLUSTER KEY;"#,
         r#"ALTER TABLE t RECLUSTER FINAL WHERE c1 > 0 LIMIT 10;"#,
@@ -290,6 +291,7 @@ fn test_statement() {
         r#"ALTER DATABASE IF EXISTS ctl.c RENAME TO a;"#,
         r#"ALTER DATABASE c RENAME TO a;"#,
         r#"ALTER DATABASE ctl.c RENAME TO a;"#,
+        r#"ALTER DATABASE ctl.c refresh cache;"#,
         r#"VACUUM TABLE t;"#,
         r#"VACUUM TABLE t DRY RUN;"#,
         r#"VACUUM TABLE t DRY RUN SUMMARY;"#,

src/query/ast/tests/it/testdata/stmt-error.txt

@@ -188,7 +188,7 @@ error:
   --> SQL:1:23
   |
 1 | alter database system x rename to db
-  | -----                 ^ unexpected `x`, expecting `RENAME` or `.`
+  | -----                 ^ unexpected `x`, expecting `RENAME`, `REFRESH`, or `.`
   | |                      
   | while parsing `ALTER DATABASE [IF EXISTS] <action>`
 

src/query/ast/tests/it/testdata/stmt.txt

@@ -12553,6 +12553,40 @@ AlterTable(
 )
 
 
+---------- Input ----------
+ALTER TABLE t refresh cache;
+---------- Output ---------
+ALTER TABLE t REFRESH CACHE
+---------- AST ------------
+AlterTable(
+    AlterTableStmt {
+        if_exists: false,
+        table_reference: Table {
+            span: Some(
+                12..13,
+            ),
+            catalog: None,
+            database: None,
+            table: Identifier {
+                span: Some(
+                    12..13,
+                ),
+                name: "t",
+                quote: None,
+                ident_type: None,
+            },
+            alias: None,
+            temporal: None,
+            with_options: None,
+            pivot: None,
+            unpivot: None,
+            sample: None,
+        },
+        action: RefreshTableCache,
+    },
+)
+
+
 ---------- Input ----------
 ALTER TABLE t COMMENT='t1-commnet';
 ---------- Output ---------
@@ -13617,6 +13651,37 @@ AlterDatabase(
 )
 
 
+---------- Input ----------
+ALTER DATABASE ctl.c refresh cache;
+---------- Output ---------
+ALTER DATABASE ctl.c REFRESH CACHE
+---------- AST ------------
+AlterDatabase(
+    AlterDatabaseStmt {
+        if_exists: false,
+        catalog: Some(
+            Identifier {
+                span: Some(
+                    15..18,
+                ),
+                name: "ctl",
+                quote: None,
+                ident_type: None,
+            },
+        ),
+        database: Identifier {
+            span: Some(
+                19..20,
+            ),
+            name: "c",
+            quote: None,
+            ident_type: None,
+        },
+        action: RefreshDatabaseCache,
+    },
+)
+
+
 ---------- Input ----------
 VACUUM TABLE t;
 ---------- Output ---------

src/query/catalog/src/database.rs

@@ -113,6 +113,16 @@ pub trait Database: DynClone + Sync + Send {
         Ok(())
     }
 
+    #[async_backtrace::framed]
+    async fn refresh_table(&self, _table_name: &str) -> Result<()> {
+        Ok(())
+    }
+
+    #[async_backtrace::framed]
+    async fn refresh_database(&self) -> Result<()> {
+        Ok(())
+    }
+
     #[async_backtrace::framed]
     async fn list_tables_names(&self) -> Result<Vec<String>> {
         Ok(vec![])

src/query/service/src/interpreters/access/privilege_access.rs

@@ -1021,6 +1021,10 @@ impl AccessChecker for PrivilegeAccess {
             Plan::DropTableClusterKey(plan) => {
                 self.validate_table_access(&plan.catalog, &plan.database, &plan.table, UserPrivilegeType::Drop, false, false).await?
             }
+            Plan::RefreshTableCache(_) | Plan::RefreshDatabaseCache(_) => {
+                // Only Iceberg support this plan
+                return Ok(())
+            }
             Plan::ReclusterTable(plan) => {
                 self.validate_table_access(&plan.catalog, &plan.database, &plan.table, UserPrivilegeType::Alter, false, false).await?
             }

src/query/service/src/interpreters/interpreter_factory.rs

@@ -62,6 +62,8 @@ use crate::interpreters::interpreter_presign::PresignInterpreter;
 use crate::interpreters::interpreter_procedure_call::CallProcedureInterpreter;
 use crate::interpreters::interpreter_procedure_create::CreateProcedureInterpreter;
 use crate::interpreters::interpreter_procedure_drop::DropProcedureInterpreter;
+use crate::interpreters::interpreter_refresh_database_cache::RefreshDatabaseCacheInterpreter;
+use crate::interpreters::interpreter_refresh_table_cache::RefreshTableCacheInterpreter;
 use crate::interpreters::interpreter_rename_warehouse::RenameWarehouseInterpreter;
 use crate::interpreters::interpreter_rename_warehouse_cluster::RenameWarehouseClusterInterpreter;
 use crate::interpreters::interpreter_resume_warehouse::ResumeWarehouseInterpreter;
@@ -338,6 +340,9 @@ impl InterpreterFactory {
             Plan::DropTableClusterKey(drop_table_cluster_key) => Ok(Arc::new(
                 DropTableClusterKeyInterpreter::try_create(ctx, *drop_table_cluster_key.clone())?,
             )),
+            Plan::RefreshTableCache(refresh_table_cache) => Ok(Arc::new(
+                RefreshTableCacheInterpreter::try_create(ctx, *refresh_table_cache.clone())?,
+            )),
             Plan::ReclusterTable(recluster) => Ok(Arc::new(ReclusterTableInterpreter::try_create(
                 ctx,
                 *recluster.clone(),
@@ -565,6 +570,9 @@ impl InterpreterFactory {
                 ctx,
                 *p.clone(),
             )?)),
+            Plan::RefreshDatabaseCache(refresh_database_cache) => Ok(Arc::new(
+                RefreshDatabaseCacheInterpreter::try_create(ctx, *refresh_database_cache.clone())?,
+            )),
             Plan::Kill(p) => Ok(Arc::new(KillInterpreter::try_create(ctx, *p.clone())?)),
 
             Plan::RevertTable(p) => Ok(Arc::new(RevertTableInterpreter::try_create(

src/query/service/src/interpreters/interpreter_refresh_database_cache.rs

@@ -0,0 +1,59 @@
+// 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::sync::Arc;
+
+use databend_common_exception::Result;
+use databend_common_sql::plans::RefreshDatabaseCachePlan;
+
+use super::Interpreter;
+use crate::pipelines::PipelineBuildResult;
+use crate::sessions::QueryContext;
+use crate::sessions::TableContext;
+
+pub struct RefreshDatabaseCacheInterpreter {
+    ctx: Arc<QueryContext>,
+    plan: RefreshDatabaseCachePlan,
+}
+
+impl RefreshDatabaseCacheInterpreter {
+    pub fn try_create(ctx: Arc<QueryContext>, plan: RefreshDatabaseCachePlan) -> Result<Self> {
+        Ok(RefreshDatabaseCacheInterpreter { ctx, plan })
+    }
+}
+
+#[async_trait::async_trait]
+impl Interpreter for RefreshDatabaseCacheInterpreter {
+    fn name(&self) -> &str {
+        "RefreshDatabaseCacheInterpreter"
+    }
+
+    fn is_ddl(&self) -> bool {
+        true
+    }
+
+    #[async_backtrace::framed]
+    async fn execute2(&self) -> Result<PipelineBuildResult> {
+        let plan = &self.plan;
+        let catalog = self.ctx.get_catalog(&plan.catalog).await?;
+
+        let _ = catalog
+            .get_database(&plan.tenant, &plan.database)
+            .await?
+            .refresh_database()
+            .await?;
+
+        Ok(PipelineBuildResult::create())
+    }
+}

src/query/service/src/interpreters/interpreter_refresh_table_cache.rs

@@ -0,0 +1,59 @@
+// 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::sync::Arc;
+
+use databend_common_exception::Result;
+use databend_common_sql::plans::RefreshTableCachePlan;
+
+use super::Interpreter;
+use crate::pipelines::PipelineBuildResult;
+use crate::sessions::QueryContext;
+use crate::sessions::TableContext;
+
+pub struct RefreshTableCacheInterpreter {
+    ctx: Arc<QueryContext>,
+    plan: RefreshTableCachePlan,
+}
+
+impl RefreshTableCacheInterpreter {
+    pub fn try_create(ctx: Arc<QueryContext>, plan: RefreshTableCachePlan) -> Result<Self> {
+        Ok(RefreshTableCacheInterpreter { ctx, plan })
+    }
+}
+
+#[async_trait::async_trait]
+impl Interpreter for RefreshTableCacheInterpreter {
+    fn name(&self) -> &str {
+        "RefreshTableCacheInterpreter"
+    }
+
+    fn is_ddl(&self) -> bool {
+        true
+    }
+
+    #[async_backtrace::framed]
+    async fn execute2(&self) -> Result<PipelineBuildResult> {
+        let plan = &self.plan;
+        let catalog = self.ctx.get_catalog(&plan.catalog).await?;
+
+        let _ = catalog
+            .get_database(&plan.tenant, &plan.database)
+            .await?
+            .refresh_table(&plan.table)
+            .await?;
+
+        Ok(PipelineBuildResult::create())
+    }
+}

src/query/service/src/interpreters/mod.rs

@@ -84,6 +84,8 @@ mod interpreter_procedure_call;
 mod interpreter_procedure_create;
 mod interpreter_procedure_desc;
 mod interpreter_procedure_drop;
+mod interpreter_refresh_database_cache;
+mod interpreter_refresh_table_cache;
 mod interpreter_rename_warehouse;
 mod interpreter_rename_warehouse_cluster;
 mod interpreter_replace;