fix

src/query/sql/src/executor/physical_plans/physical_recluster.rs

@@ -30,9 +30,9 @@ use crate::executor::physical_plans::FragmentKind;
 use crate::executor::physical_plans::MutationKind;
 use crate::executor::PhysicalPlan;
 use crate::executor::PhysicalPlanBuilder;
+use crate::plans::plan_hilbert_stmt;
 use crate::plans::set_update_stream_columns;
 use crate::plans::Plan;
-use crate::Planner;
 
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct Recluster {
@@ -80,8 +80,7 @@ impl PhysicalPlanBuilder {
                     )));
                 };
 
-                let mut planner = Planner::new(self.ctx.clone());
-                let plan = planner.plan_stmt(stmt).await?;
+                let plan = plan_hilbert_stmt(self.ctx.clone(), stmt).await?;
                 let (mut s_expr, metadata, bind_context) = match plan {
                     Plan::Query {
                         s_expr,

src/query/sql/src/planner/plans/recluster.rs

@@ -15,13 +15,21 @@
 use std::sync::Arc;
 
 use databend_common_ast::ast::Statement;
+use databend_common_catalog::catalog::CatalogManager;
 use databend_common_catalog::plan::Filters;
+use databend_common_catalog::table_context::TableContext;
 use educe::Educe;
 
+use crate::optimizer::optimize;
+use crate::optimizer::OptimizerContext;
 use crate::optimizer::SExpr;
 use crate::plans::Operator;
+use crate::plans::Plan;
 use crate::plans::RelOp;
 use crate::plans::RelOperator;
+use crate::Binder;
+use crate::MetadataRef;
+use crate::NameResolutionContext;
 
 #[derive(Debug, PartialEq, Clone, Educe)]
 #[educe(Eq, Hash)]
@@ -62,3 +70,27 @@ pub(crate) fn set_update_stream_columns(
         }
     }
 }
+
+#[async_backtrace::framed]
+#[fastrace::trace]
+pub(crate) async fn plan_hilbert_stmt(
+    ctx: Arc<dyn TableContext>,
+    stmt: &Statement,
+) -> databend_common_exception::Result<Plan> {
+    let settings = ctx.get_settings();
+    let name_resolution_ctx = NameResolutionContext::try_from(settings.as_ref())?;
+    let metadata = MetadataRef::default();
+    let binder = Binder::new(
+        ctx.clone(),
+        CatalogManager::instance(),
+        name_resolution_ctx,
+        metadata.clone(),
+    );
+
+    let plan = binder.bind(stmt).await?;
+    let opt_ctx = OptimizerContext::new(ctx.clone(), metadata)
+        .with_enable_distributed_optimization(!ctx.get_cluster().is_empty())
+        .with_enable_join_reorder(unsafe { !settings.get_disable_join_reorder()? })
+        .with_enable_dphyp(settings.get_enable_dphyp()?);
+    optimize(opt_ctx, plan).await
+}