fix

Author: zhyass

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

@@ -267,11 +267,11 @@ impl ReclusterTableInterpreter {
                         acc
                     },
                 );
-                
+
                 let query_str = self.ctx.get_query_str();
                 let write_progress = self.ctx.get_write_progress();
                 let write_progress_value = write_progress.as_ref().get_values();
-                
+
                 let subquery_executor = Arc::new(ServiceQueryExecutor::new(
                     QueryContext::create_from(self.ctx.as_ref()),
                 ));
@@ -408,7 +408,6 @@ impl ReclusterTableInterpreter {
                     plan_id: 0,
                     input: plan,
                     table_info: table_info.clone(),
-                    range_id: bind_context.columns.last().unwrap().index,
                     num_partitions: total_partitions,
                     rows_per_block,
                 }));

src/query/service/src/pipelines/processors/transforms/transform_partition_collect.rs

@@ -60,19 +60,24 @@ impl Exchange for HilbertPartitionExchange {
             .unwrap()
             .as_u_int64()
             .unwrap();
+        data_block.pop_columns(1);
 
         // Scatter the data block to different partitions.
         let indices = range_ids
             .iter()
             .map(|&id| (id % self.num_partitions as u64) as u16)
             .collect::<Vec<_>>();
-        data_block.pop_columns(1);
-        let scatter_blocks = DataBlock::scatter(&data_block, &indices, self.num_partitions)?;
-
+        let scatter_indices =
+            DataBlock::divide_indices_by_scatter_size(&indices, self.num_partitions);
         // Partition the data blocks to different processors.
         let mut output_data_blocks = vec![vec![]; n];
-        for (partition_id, data_block) in scatter_blocks.into_iter().enumerate() {
-            output_data_blocks[partition_id % n].push((partition_id, data_block));
+        for (partition_id, indices) in scatter_indices.iter().take(self.num_partitions).enumerate()
+        {
+            if indices.is_empty() {
+                continue;
+            }
+            let block = data_block.take_with_optimize_size(indices)?;
+            output_data_blocks[partition_id % n].push((partition_id, block));
         }
 
         // Union data blocks for each processor.

src/query/service/src/servers/flight/v1/exchange/exchange_injector.rs

@@ -93,7 +93,6 @@ impl ExchangeInjector for DefaultExchangeInjector {
                     .iter()
                     .position(|x| x == local_id)
                     .unwrap();
-                println!("local_pos {}", local_pos);
                 HashFlightScatter::try_create(
                     ctx.get_function_context()?,
                     exchange.shuffle_keys.clone(),

src/query/sql/src/executor/physical_plan_visitor.rs

@@ -133,7 +133,6 @@ pub trait PhysicalPlanReplacer {
             plan_id: plan.plan_id,
             input: Box::new(input),
             table_info: plan.table_info.clone(),
-            range_id: plan.range_id,
             num_partitions: plan.num_partitions,
             rows_per_block: plan.rows_per_block,
         })))

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

@@ -16,7 +16,6 @@ use databend_common_catalog::plan::ReclusterTask;
 use databend_common_meta_app::schema::TableInfo;
 
 use crate::executor::PhysicalPlan;
-use crate::IndexType;
 
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 pub struct Recluster {
@@ -30,7 +29,6 @@ pub struct HilbertPartition {
     pub plan_id: u32,
     pub input: Box<PhysicalPlan>,
     pub table_info: TableInfo,
-    pub range_id: IndexType,
     pub num_partitions: usize,
     pub rows_per_block: usize,
 }