fix

Author: zhyass

Cargo.lock

@@ -424,7 +424,6 @@ build_exceptions! {
     InvalidSessionState(4004),
 
     // recluster error codes
-    NoNeedToRecluster(4011),
     NoNeedToCompact(4012),
     UnsupportedClusterType(4013),
 

src/common/exception/src/exception_code.rs

@@ -107,6 +107,7 @@ databend-enterprise-aggregating-index = { workspace = true }
 databend-enterprise-attach-table = { workspace = true }
 databend-enterprise-background-service = { workspace = true }
 databend-enterprise-data-mask-feature = { workspace = true }
+databend-enterprise-hilbert-clustering = { workspace = true }
 databend-enterprise-inverted-index = { workspace = true }
 databend-enterprise-resources-management = { workspace = true }
 databend-enterprise-stream-handler = { workspace = true }

src/query/service/Cargo.toml

@@ -38,7 +38,6 @@ use databend_common_sql::plans::InsertInputSource;
 use databend_common_sql::plans::Mutation;
 use databend_common_sql::plans::OptimizeCompactBlock;
 use databend_common_sql::plans::PresignAction;
-use databend_common_sql::plans::Recluster;
 use databend_common_sql::plans::RewriteKind;
 use databend_common_sql::Planner;
 use databend_common_users::RoleCacheManager;
@@ -965,9 +964,7 @@ impl AccessChecker for PrivilegeAccess {
             Plan::DropTableClusterKey(plan) => {
                 self.validate_table_access(&plan.catalog, &plan.database, &plan.table, UserPrivilegeType::Drop, false, false).await?
             }
-            Plan::ReclusterTable{s_expr, ..} => {
-                let plan: Recluster = s_expr.plan().clone().try_into()?;
-                // UDF has been disabled in recluster, no need to check udf privilege access.
+            Plan::ReclusterTable(plan) => {
                 self.validate_table_access(&plan.catalog, &plan.database, &plan.table, UserPrivilegeType::Alter, false, false).await?
             }
             Plan::TruncateTable(plan) => {

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

@@ -26,7 +26,7 @@ use databend_common_pipeline_core::Pipeline;
 use databend_common_sql::executor::physical_plans::MutationKind;
 use databend_common_sql::optimizer::SExpr;
 use databend_common_sql::plans::OptimizeCompactBlock;
-use databend_common_sql::plans::Recluster;
+use databend_common_sql::plans::ReclusterPlan;
 use databend_common_sql::plans::RelOperator;
 use databend_storages_common_table_meta::table::ClusterType;
 use log::info;
@@ -218,21 +218,16 @@ async fn compact_table(
                     &compact_target.table,
                 )?;
                 ctx.set_enable_sort_spill(false);
-                let recluster = RelOperator::Recluster(Recluster {
+                let recluster = ReclusterPlan {
                     catalog: compact_target.catalog,
                     database: compact_target.database,
                     table: compact_target.table,
                     limit: Some(settings.get_auto_compaction_segments_limit()? as usize),
-                    filters: None,
-                });
-                let s_expr = SExpr::create_leaf(Arc::new(recluster));
-                let recluster_interpreter = ReclusterTableInterpreter::try_create(
-                    ctx.clone(),
-                    s_expr,
-                    None,
-                    lock_opt,
-                    false,
-                )?;
+                    selection: None,
+                    is_final: false,
+                };
+                let recluster_interpreter =
+                    ReclusterTableInterpreter::try_create(ctx.clone(), recluster, lock_opt)?;
                 // Recluster will be done in `ReclusterTableInterpreter::execute2` directly,
                 // we do not need to use `PipelineCompleteExecutor` to execute it.
                 let build_res = recluster_interpreter.execute2().await?;

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

@@ -339,16 +339,10 @@ impl InterpreterFactory {
             Plan::DropTableClusterKey(drop_table_cluster_key) => Ok(Arc::new(
                 DropTableClusterKeyInterpreter::try_create(ctx, *drop_table_cluster_key.clone())?,
             )),
-            Plan::ReclusterTable {
-                s_expr,
-                hilbert_query,
-                is_final,
-            } => Ok(Arc::new(ReclusterTableInterpreter::try_create(
+            Plan::ReclusterTable(recluster) => Ok(Arc::new(ReclusterTableInterpreter::try_create(
                 ctx,
-                *s_expr.clone(),
-                hilbert_query.clone(),
+                *recluster.clone(),
                 LockTableOption::LockWithRetry,
-                *is_final,
             )?)),
             Plan::TruncateTable(truncate_table) => Ok(Arc::new(
                 TruncateTableInterpreter::try_create(ctx, *truncate_table.clone())?,

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

@@ -337,7 +337,10 @@ fn not(expr: ScalarExpr) -> ScalarExpr {
     })
 }
 
-fn scalar_expr_to_remote_expr(expr: &ScalarExpr, block_schema: &DataSchema) -> Result<RemoteExpr> {
+pub(crate) fn scalar_expr_to_remote_expr(
+    expr: &ScalarExpr,
+    block_schema: &DataSchema,
+) -> Result<RemoteExpr> {
     let expr = expr
         .as_expr()?
         .project_column_ref(|col| block_schema.index_of(&col.index.to_string()).unwrap());