chore: simplify alter and drop cluster key logic

Author: zhyass

src/common/base/src/runtime/metrics/family_metrics/counter.rs

@@ -42,7 +42,7 @@ impl<Labels: FamilyLabels> FamilyCounter<Labels> {
         FamilyCounter {
             index,
             labels,
-            value: Arc::new(Default::default()),
+            value: Default::default(),
         }
     }
 

src/query/catalog/src/table.rs

@@ -14,6 +14,7 @@
 
 use std::any::Any;
 use std::collections::BTreeMap;
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use chrono::DateTime;
@@ -22,12 +23,13 @@ use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::BlockThresholds;
 use databend_common_expression::ColumnId;
-use databend_common_expression::RemoteExpr;
 use databend_common_expression::Scalar;
 use databend_common_expression::TableSchema;
 use databend_common_io::constants::DEFAULT_BLOCK_BUFFER_SIZE;
 use databend_common_io::constants::DEFAULT_BLOCK_MAX_ROWS;
 use databend_common_io::constants::DEFAULT_BLOCK_MIN_ROWS;
+use databend_common_meta_app::app_error::AppError;
+use databend_common_meta_app::app_error::UnknownTableId;
 use databend_common_meta_app::schema::TableIdent;
 use databend_common_meta_app::schema::TableInfo;
 use databend_common_meta_app::schema::TableMeta;
@@ -37,9 +39,12 @@ use databend_common_meta_types::MetaId;
 use databend_common_pipeline_core::Pipeline;
 use databend_common_storage::Histogram;
 use databend_common_storage::StorageMetrics;
+use databend_storages_common_table_meta::meta::ClusterKey;
 use databend_storages_common_table_meta::meta::SnapshotId;
 use databend_storages_common_table_meta::meta::TableSnapshot;
 use databend_storages_common_table_meta::table::ChangeType;
+use databend_storages_common_table_meta::table::ClusterType;
+use databend_storages_common_table_meta::table::OPT_KEY_CLUSTER_TYPE;
 use databend_storages_common_table_meta::table::OPT_KEY_TEMP_PREFIX;
 use databend_storages_common_table_meta::table_id_ranges::is_temp_table_id;
 
@@ -117,8 +122,18 @@ pub trait Table: Sync + Send {
         false
     }
 
-    fn cluster_keys(&self, _ctx: Arc<dyn TableContext>) -> Vec<RemoteExpr<String>> {
-        vec![]
+    fn cluster_key_meta(&self) -> Option<ClusterKey> {
+        None
+    }
+
+    fn cluster_type(&self) -> Option<ClusterType> {
+        self.cluster_key_meta()?;
+        let cluster_type = self
+            .options()
+            .get(OPT_KEY_CLUSTER_TYPE)
+            .and_then(|s| s.parse::<ClusterType>().ok())
+            .unwrap_or(ClusterType::Linear);
+        Some(cluster_type)
     }
 
     fn change_tracking_enabled(&self) -> bool {
@@ -159,31 +174,6 @@ pub trait Table: Sync + Send {
         false
     }
 
-    #[async_backtrace::framed]
-    async fn alter_table_cluster_keys(
-        &self,
-        ctx: Arc<dyn TableContext>,
-        cluster_key: String,
-        cluster_type: String,
-    ) -> Result<()> {
-        let (_, _, _) = (ctx, cluster_key, cluster_type);
-
-        Err(ErrorCode::UnsupportedEngineParams(format!(
-            "Altering table cluster keys is not supported for the '{}' engine.",
-            self.engine()
-        )))
-    }
-
-    #[async_backtrace::framed]
-    async fn drop_table_cluster_keys(&self, ctx: Arc<dyn TableContext>) -> Result<()> {
-        let _ = ctx;
-
-        Err(ErrorCode::UnsupportedEngineParams(format!(
-            "Dropping table cluster keys is not supported for the '{}' engine.",
-            self.engine()
-        )))
-    }
-
     /// Gather partitions to be scanned according to the push_downs
     #[async_backtrace::framed]
     async fn read_partitions(
@@ -584,11 +574,6 @@ pub struct NavigationDescriptor {
     pub point: NavigationPoint,
 }
 
-use std::collections::HashMap;
-
-use databend_common_meta_app::app_error::AppError;
-use databend_common_meta_app::app_error::UnknownTableId;
-
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default)]
 pub struct ParquetTableColumnStatisticsProvider {
     column_stats: HashMap<ColumnId, Option<BasicColumnStatistics>>,

src/query/service/src/interpreters/hook/compact_hook.rs

@@ -194,7 +194,7 @@ async fn compact_table(
 
     {
         // do recluster.
-        if !table.cluster_keys(ctx.clone()).is_empty() {
+        if table.cluster_key_meta().is_some() {
             let recluster = RelOperator::Recluster(Recluster {
                 catalog: compact_target.catalog,
                 database: compact_target.database,

src/query/service/src/interpreters/interpreter_cluster_key_alter.rs

@@ -14,8 +14,14 @@
 
 use std::sync::Arc;
 
+use databend_common_catalog::table::Table;
+use databend_common_catalog::table::TableExt;
 use databend_common_exception::Result;
+use databend_common_meta_app::schema::UpdateTableMetaReq;
+use databend_common_meta_types::MatchSeq;
 use databend_common_sql::plans::AlterTableClusterKeyPlan;
+use databend_common_storages_fuse::FuseTable;
+use databend_storages_common_table_meta::table::OPT_KEY_CLUSTER_TYPE;
 
 use super::Interpreter;
 use crate::pipelines::PipelineBuildResult;
@@ -52,12 +58,35 @@ impl Interpreter for AlterTableClusterKeyInterpreter {
         let table = catalog
             .get_table(&tenant, &plan.database, &plan.table)
             .await?;
+        // check mutability
+        table.check_mutable()?;
 
+        let fuse_table = FuseTable::try_from_table(table.as_ref())?;
         let cluster_key_str = format!("({})", plan.cluster_keys.join(", "));
+        // if new cluster_key_str is the same with old one,
+        // no need to change
+        if let Some(old_cluster_key_str) = fuse_table.cluster_key_str()
+            && *old_cluster_key_str == cluster_key_str
+        {
+            let old_cluster_type = fuse_table.cluster_type();
+            if old_cluster_type.is_some_and(|v| v.to_string().to_lowercase() == plan.cluster_type) {
+                return Ok(PipelineBuildResult::create());
+            }
+        }
 
-        table
-            .alter_table_cluster_keys(self.ctx.clone(), cluster_key_str, plan.cluster_type.clone())
-            .await?;
+        let table_info = fuse_table.get_table_info();
+        let mut new_table_meta = table_info.meta.clone();
+        new_table_meta
+            .options
+            .insert(OPT_KEY_CLUSTER_TYPE.to_owned(), plan.cluster_type.clone());
+        new_table_meta = new_table_meta.push_cluster_key(cluster_key_str);
+
+        let req = UpdateTableMetaReq {
+            table_id: table_info.ident.table_id,
+            seq: MatchSeq::Exact(table_info.ident.seq),
+            new_table_meta,
+        };
+        catalog.update_single_table_meta(req, table_info).await?;
 
         Ok(PipelineBuildResult::create())
     }

src/query/service/src/interpreters/interpreter_cluster_key_drop.rs

@@ -14,8 +14,14 @@
 
 use std::sync::Arc;
 
+use databend_common_catalog::table::Table;
+use databend_common_catalog::table::TableExt;
 use databend_common_exception::Result;
+use databend_common_meta_app::schema::UpdateTableMetaReq;
+use databend_common_meta_types::MatchSeq;
 use databend_common_sql::plans::DropTableClusterKeyPlan;
+use databend_common_storages_fuse::FuseTable;
+use databend_storages_common_table_meta::table::OPT_KEY_CLUSTER_TYPE;
 
 use super::Interpreter;
 use crate::pipelines::PipelineBuildResult;
@@ -52,8 +58,25 @@ impl Interpreter for DropTableClusterKeyInterpreter {
         let table = catalog
             .get_table(&tenant, &plan.database, &plan.table)
             .await?;
+        if table.cluster_key_meta().is_none() {
+            return Ok(PipelineBuildResult::create());
+        }
+        // check mutability
+        table.check_mutable()?;
 
-        table.drop_table_cluster_keys(self.ctx.clone()).await?;
+        let fuse_table = FuseTable::try_from_table(table.as_ref())?;
+        let table_info = fuse_table.get_table_info();
+        let mut new_table_meta = table_info.meta.clone();
+        new_table_meta.default_cluster_key = None;
+        new_table_meta.default_cluster_key_id = None;
+        new_table_meta.options.remove(OPT_KEY_CLUSTER_TYPE);
+
+        let req = UpdateTableMetaReq {
+            table_id: table_info.ident.table_id,
+            seq: MatchSeq::Exact(table_info.ident.seq),
+            new_table_meta,
+        };
+        catalog.update_single_table_meta(req, table_info).await?;
 
         Ok(PipelineBuildResult::create())
     }

src/query/service/src/interpreters/interpreter_replace.rs

@@ -290,7 +290,7 @@ impl ReplaceInterpreter {
             .ctx
             .get_settings()
             .get_replace_into_bloom_pruning_max_column_number()?;
-        let bloom_filter_column_indexes = if !table.cluster_keys(self.ctx.clone()).is_empty() {
+        let bloom_filter_column_indexes = if table.cluster_key_meta().is_some() {
             fuse_table
                 .choose_bloom_filter_columns(
                     self.ctx.clone(),

src/query/service/src/servers/flight_sql/flight_sql_service/mod.rs

@@ -59,7 +59,7 @@ impl FlightSqlServiceImpl {
     pub fn create() -> Self {
         FlightSqlServiceImpl {
             sessions: Mutex::new(Default::default()),
-            statements: Arc::new(Default::default()),
+            statements: Default::default(),
         }
     }
 }

src/query/service/src/sessions/query_ctx.rs

@@ -180,7 +180,7 @@ impl QueryContext {
             fragment_id: Arc::new(AtomicUsize::new(0)),
             inserted_segment_locs: Arc::new(RwLock::new(HashSet::new())),
             block_threshold: Arc::new(RwLock::new(BlockThresholds::default())),
-            m_cte_temp_table: Arc::new(Default::default()),
+            m_cte_temp_table: Default::default(),
         })
     }
 

src/query/service/src/sessions/query_ctx_shared.rs

@@ -181,8 +181,8 @@ impl QueryContextShared {
             finish_time: Default::default(),
             on_error_map: Arc::new(RwLock::new(None)),
             on_error_mode: Arc::new(RwLock::new(None)),
-            copy_status: Arc::new(Default::default()),
-            mutation_status: Arc::new(Default::default()),
+            copy_status: Default::default(),
+            mutation_status: Default::default(),
             partitions_shas: Arc::new(RwLock::new(vec![])),
             cacheable: Arc::new(AtomicBool::new(true)),
             can_scan_from_agg_index: Arc::new(AtomicBool::new(true)),

src/query/service/src/sessions/session.rs

@@ -63,7 +63,7 @@ impl Session {
         session_ctx: Box<SessionContext>,
         mysql_connection_id: Option<u32>,
     ) -> Result<Session> {
-        let status = Arc::new(Default::default());
+        let status = Default::default();
         Ok(Session {
             id,
             typ: RwLock::new(typ),