fix

Author: zhyass

src/query/service/src/interpreters/interpreter_table_recluster.rs

@@ -255,14 +255,17 @@ impl ReclusterTableInterpreter {
                 let block_thresholds = tbl.get_block_thresholds();
                 let total_bytes = recluster_info.removed_statistics.uncompressed_byte_size as usize;
                 let total_rows = recluster_info.removed_statistics.row_count as usize;
-                let rows_per_block = block_thresholds.calc_rows_per_block(total_bytes, total_rows);
-                let total_partitions = total_rows / rows_per_block;
+                let block_size = settings.get_max_block_size()?;
+                let rows_per_block = block_thresholds
+                    .calc_rows_per_block(total_bytes, total_rows)
+                    .max(block_size as usize);
+                let total_partitions = std::cmp::max(total_rows / rows_per_block, 1);
 
                 let subquery_executor = Arc::new(ServiceQueryExecutor::new(
                     QueryContext::create_from(self.ctx.as_ref()),
                 ));
+                let partitions = settings.get_hilbert_num_range_ids()? as usize;
                 if hilbert_info.is_none() {
-                    let partitions = settings.get_hilbert_num_range_ids()? as usize;
                     let sample_size = settings.get_hilbert_sample_size_per_block()?;
 
                     let name_resolution_ctx = NameResolutionContext::try_from(settings.as_ref())?;
@@ -360,18 +363,20 @@ impl ReclusterTableInterpreter {
                 let mut variables = VecDeque::new();
 
                 let Plan::Query {
-                    s_expr,
+                    mut s_expr,
                     metadata,
                     bind_context,
                     ..
-                } = keys_bound
+                } = keys_bound.clone()
                 else {
                     unreachable!()
                 };
+                if total_partitions > partitions {
+                    *s_expr = replace_with_constant(&s_expr, &variables, total_partitions as u16)
+                }
                 metadata.write().replace_all_tables(tbl.clone());
-                let mut builder =
-                    PhysicalPlanBuilder::new(metadata.clone(), self.ctx.clone(), false);
-                let plan = Box::new(builder.build(s_expr, bind_context.column_set()).await?);
+                let mut builder = PhysicalPlanBuilder::new(metadata, self.ctx.clone(), false);
+                let plan = Box::new(builder.build(&s_expr, bind_context.column_set()).await?);
                 let data_blocks = subquery_executor
                     .execute_query_with_physical_plan(&plan)
                     .await?;

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

@@ -93,11 +93,7 @@ pub async fn plan_hilbert_sql(
     optimize(opt_ctx, plan).await
 }
 
-pub fn replace_with_constant(
-    expr: &SExpr,
-    variables: &VecDeque<Scalar>,
-    total_partitions: u16,
-) -> SExpr {
+pub fn replace_with_constant(expr: &SExpr, variables: &VecDeque<Scalar>, partitions: u16) -> SExpr {
     #[recursive::recursive]
     fn visit_expr_column(expr: &mut ScalarExpr, variables: &mut VecDeque<Scalar>) {
         match expr {
@@ -124,11 +120,11 @@ pub fn replace_with_constant(
     fn replace_with_constant_into_child(
         s_expr: &SExpr,
         variables: &mut VecDeque<Scalar>,
-        total_partitions: u16,
+        partitions: u16,
     ) -> SExpr {
         let mut s_expr = s_expr.clone();
         s_expr.plan = match s_expr.plan.as_ref() {
-            RelOperator::EvalScalar(expr) => {
+            RelOperator::EvalScalar(expr) if !variables.is_empty() => {
                 let mut expr = expr.clone();
                 for item in &mut expr.items {
                     visit_expr_column(&mut item.scalar, variables);
@@ -137,12 +133,10 @@ pub fn replace_with_constant(
             }
             RelOperator::Aggregate(aggr) => {
                 let mut aggr = aggr.clone();
-                if aggr.aggregate_functions.len() == 1 {
-                    let mut agg_func = aggr.aggregate_functions[0].clone();
-                    if let ScalarExpr::AggregateFunction(func) = &mut agg_func.scalar {
+                for item in &mut aggr.aggregate_functions {
+                    if let ScalarExpr::AggregateFunction(func) = &mut item.scalar {
                         if func.func_name == "range_bound" {
-                            func.params[0] = Scalar::Number(NumberScalar::UInt16(total_partitions));
-                            aggr.aggregate_functions = vec![agg_func];
+                            func.params[0] = Scalar::Number(NumberScalar::UInt16(partitions));
                         }
                     }
                 }
@@ -157,9 +151,7 @@ pub fn replace_with_constant(
             let mut children = Vec::with_capacity(s_expr.children.len());
             for child in s_expr.children.iter() {
                 children.push(Arc::new(replace_with_constant_into_child(
-                    child,
-                    variables,
-                    total_partitions,
+                    child, variables, partitions,
                 )));
             }
             s_expr.children = children;
@@ -168,5 +160,5 @@ pub fn replace_with_constant(
     }
 
     let mut variables = variables.clone();
-    replace_with_constant_into_child(expr, &mut variables, total_partitions)
+    replace_with_constant_into_child(expr, &mut variables, partitions)
 }