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/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,66 @@
+// 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::ErrorCode;
+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?;
+
+        if !catalog.support_partition() {
+            return Err(ErrorCode::TableEngineNotSupported(
+                "Only Iceberg catalog support execute ALTER DATABASE <database_name> REFRESH CACHE",
+            ));
+        }
+
+        let _ = catalog
+            .get_database(&plan.tenant, &plan.database)
+            .await?
+            .trigger_use()
+            .await?;
+
+        Ok(PipelineBuildResult::create())
+    }
+}

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

@@ -0,0 +1,64 @@
+// 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::ErrorCode;
+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?;
+
+        if !catalog.support_partition() {
+            return Err(ErrorCode::TableEngineNotSupported(
+                "Only Iceberg catalog support execute ALTER TABLE <table_name> REFRESH CACHE",
+            ));
+        }
+
+        let _ = catalog
+            .get_table(&plan.tenant, &plan.database, &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;

src/query/sql/src/planner/binder/ddl/database.rs

@@ -38,6 +38,7 @@ use crate::planner::semantic::normalize_identifier;
 use crate::plans::CreateDatabasePlan;
 use crate::plans::DropDatabasePlan;
 use crate::plans::Plan;
+use crate::plans::RefreshDatabaseCachePlan;
 use crate::plans::RenameDatabaseEntity;
 use crate::plans::RenameDatabasePlan;
 use crate::plans::RewriteKind;
@@ -199,6 +200,14 @@ impl Binder {
                     entities: vec![entry],
                 })))
             }
+
+            AlterDatabaseAction::RefreshDatabaseCache => Ok(Plan::RefreshDatabaseCache(Box::new(
+                RefreshDatabaseCachePlan {
+                    tenant,
+                    catalog,
+                    database,
+                },
+            ))),
         }
     }
 

src/query/sql/src/planner/binder/ddl/table.rs

@@ -129,6 +129,7 @@ use crate::plans::OptimizeCompactSegmentPlan;
 use crate::plans::OptimizePurgePlan;
 use crate::plans::Plan;
 use crate::plans::ReclusterPlan;
+use crate::plans::RefreshTableCachePlan;
 use crate::plans::RelOperator;
 use crate::plans::RenameTableColumnPlan;
 use crate::plans::RenameTablePlan;
@@ -1130,6 +1131,14 @@ impl Binder {
                     table,
                 })))
             }
+            AlterTableAction::RefreshTableCache => {
+                Ok(Plan::RefreshTableCache(Box::new(RefreshTableCachePlan {
+                    tenant,
+                    catalog,
+                    database,
+                    table,
+                })))
+            }
         }
     }
 

src/query/sql/src/planner/format/display_plan.rs

@@ -69,6 +69,7 @@ impl Plan {
             Plan::DropDatabase(_) => Ok("DropDatabase".to_string()),
             Plan::UndropDatabase(_) => Ok("UndropDatabase".to_string()),
             Plan::RenameDatabase(_) => Ok("RenameDatabase".to_string()),
+            Plan::RefreshDatabaseCache(_) => Ok("RefreshDatabaseCache".to_string()),
 
             // Tables
             Plan::CreateTable(create_table) => format_create_table(create_table, options),
@@ -86,6 +87,7 @@ impl Plan {
             Plan::DropTableColumn(_) => Ok("DropTableColumn".to_string()),
             Plan::AlterTableClusterKey(_) => Ok("AlterTableClusterKey".to_string()),
             Plan::DropTableClusterKey(_) => Ok("DropTableClusterKey".to_string()),
+            Plan::RefreshTableCache(_) => Ok("RefreshTableCache".to_string()),
             Plan::ReclusterTable(_) => Ok("ReclusterTable".to_string()),
             Plan::TruncateTable(_) => Ok("TruncateTable".to_string()),
             Plan::OptimizePurge(_) => Ok("OptimizePurge".to_string()),

src/query/sql/src/planner/plans/ddl/database.rs

@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
+use databend_common_expression::DataSchema;
 use databend_common_expression::DataSchemaRef;
 use databend_common_meta_app::schema::database_name_ident::DatabaseNameIdent;
 use databend_common_meta_app::schema::CreateDatabaseReq;
@@ -128,3 +131,16 @@ impl ShowCreateDatabasePlan {
         self.schema.clone()
     }
 }
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct RefreshDatabaseCachePlan {
+    pub tenant: Tenant,
+    pub catalog: String,
+    pub database: String,
+}
+
+impl RefreshDatabaseCachePlan {
+    pub fn schema(&self) -> DataSchemaRef {
+        Arc::new(DataSchema::empty())
+    }
+}

src/query/sql/src/planner/plans/ddl/table.rs

@@ -495,3 +495,17 @@ impl DropTableClusterKeyPlan {
         Arc::new(DataSchema::empty())
     }
 }
+
+#[derive(Clone, Debug)]
+pub struct RefreshTableCachePlan {
+    pub tenant: Tenant,
+    pub catalog: String,
+    pub database: String,
+    pub table: String,
+}
+
+impl RefreshTableCachePlan {
+    pub fn schema(&self) -> DataSchemaRef {
+        Arc::new(DataSchema::empty())
+    }
+}