remove unused codes

Author: zhyass

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

@@ -28,7 +28,6 @@ use databend_common_catalog::table::TableExt;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::type_check::check_function;
-use databend_common_expression::types::NumberScalar;
 use databend_common_expression::DataBlock;
 use databend_common_expression::Scalar;
 use databend_common_functions::BUILTIN_FUNCTIONS;
@@ -53,8 +52,6 @@ use databend_common_sql::plans::plan_hilbert_sql;
 use databend_common_sql::plans::replace_with_constant;
 use databend_common_sql::plans::set_update_stream_columns;
 use databend_common_sql::plans::BoundColumnRef;
-use databend_common_sql::plans::ConstantExpr;
-use databend_common_sql::plans::FunctionCall;
 use databend_common_sql::plans::Plan;
 use databend_common_sql::plans::ReclusterPlan;
 use databend_common_sql::IdentifierNormalizer;
@@ -433,44 +430,22 @@ impl ReclusterTableInterpreter {
 
         // For distributed execution, add an exchange operator to distribute work
         if is_distributed {
-            let nodes_num = cluster.nodes.len() as u64;
-            let scalar_expr = ScalarExpr::FunctionCall(FunctionCall {
-                span: None,
-                func_name: "div".to_string(),
-                params: vec![],
-                arguments: vec![
-                    ScalarExpr::FunctionCall(FunctionCall {
-                        span: None,
-                        func_name: "multiply".to_string(),
-                        params: vec![],
-                        arguments: vec![
-                            ScalarExpr::BoundColumnRef(BoundColumnRef {
-                                span: None,
-                                column: bind_context.columns.last().unwrap().clone(),
-                            }),
-                            ScalarExpr::ConstantExpr(ConstantExpr {
-                                span: None,
-                                value: Scalar::Number(NumberScalar::UInt64(nodes_num)),
-                            }),
-                        ],
-                    }),
-                    ScalarExpr::ConstantExpr(ConstantExpr {
-                        span: None,
-                        value: Scalar::Number(NumberScalar::UInt64(total_partitions as u64)),
-                    }),
-                ],
-            });
-
             // Create an expression for the partition column,
             // i.e.`range_partition_id(hilbert_range_index({hilbert_keys_str}), [...]) AS _predicate`
-            let expr = scalar_expr_to_remote_expr(&scalar_expr, plan.output_schema()?.as_ref())?;
+            let expr = scalar_expr_to_remote_expr(
+                &ScalarExpr::BoundColumnRef(BoundColumnRef {
+                    span: None,
+                    column: bind_context.columns.last().unwrap().clone(),
+                }),
+                plan.output_schema()?.as_ref(),
+            )?;
 
             // Add exchange operator for data distribution,
             // shuffling data based on the hash of range partition IDs derived from the Hilbert index.
             plan = Box::new(PhysicalPlan::Exchange(Exchange {
                 plan_id: 0,
                 input: plan,
-                kind: FragmentKind::Modulo,
+                kind: FragmentKind::Normal,
                 keys: vec![expr],
                 allow_adjust_parallelism: true,
                 ignore_exchange: false,
@@ -487,8 +462,7 @@ impl ReclusterTableInterpreter {
             plan_id: 0,
             input: plan,
             table_info: table_info.clone(),
-            range_start: 0,
-            range_width: total_partitions,
+            num_partitions: total_partitions,
             table_meta_timestamps,
             bytes_per_block,
             rows_per_block,

src/query/service/src/pipelines/builders/builder_hilbert_partition.rs

@@ -51,7 +51,7 @@ impl PipelineBuilder {
 
         self.main_pipeline.exchange(
             num_processors,
-            ReclusterPartitionExchange::create(partition.range_start, partition.range_width),
+            ReclusterPartitionExchange::create(partition.num_partitions),
         );
 
         let settings = self.ctx.get_settings();
@@ -85,7 +85,7 @@ impl PipelineBuilder {
                         &settings,
                         processor_id.fetch_add(1, atomic::Ordering::AcqRel),
                         num_processors,
-                        partition.range_width,
+                        partition.num_partitions,
                         window_spill_settings.clone(),
                         disk_spill.clone(),
                         ReclusterPartitionStrategy::new(properties.clone()),
@@ -112,7 +112,7 @@ impl PipelineBuilder {
                         &settings,
                         processor_id.fetch_add(1, atomic::Ordering::AcqRel),
                         num_processors,
-                        partition.range_width,
+                        partition.num_partitions,
                         window_spill_settings.clone(),
                         disk_spill.clone(),
                         CompactPartitionStrategy::new(

src/query/service/src/pipelines/builders/builder_recluster.rs

@@ -195,7 +195,7 @@ impl PipelineBuilder {
 
                     self.main_pipeline.exchange(
                         num_processors,
-                        ReclusterPartitionExchange::create(0, partitions),
+                        ReclusterPartitionExchange::create(partitions),
                     );
                     let processor_id = AtomicUsize::new(0);
 

src/query/service/src/pipelines/processors/transforms/aggregator/aggregate_exchange_injector.rs

@@ -233,7 +233,6 @@ impl ExchangeInjector for AggregateInjector {
         match exchange {
             DataExchange::Merge(_) => unreachable!(),
             DataExchange::Broadcast(_) => unreachable!(),
-            DataExchange::Modulo(_) => unreachable!(),
             DataExchange::ShuffleDataExchange(exchange) => {
                 Ok(Arc::new(Box::new(HashTableHashScatter {
                     buckets: exchange.destination_ids.len(),

src/query/service/src/pipelines/processors/transforms/recluster/recluster_partition_exchange.rs

@@ -22,16 +22,14 @@ use databend_common_pipeline_core::processors::Exchange;
 use crate::pipelines::processors::transforms::WindowPartitionMeta;
 
 pub struct ReclusterPartitionExchange {
-    start: u64,
-    width: usize,
+    num_partitions: usize,
     start_time: Instant,
 }
 
 impl ReclusterPartitionExchange {
-    pub fn create(start: u64, width: usize) -> Arc<ReclusterPartitionExchange> {
+    pub fn create(num_partitions: usize) -> Arc<ReclusterPartitionExchange> {
         Arc::new(ReclusterPartitionExchange {
-            start,
-            width,
+            num_partitions,
             start_time: Instant::now(),
         })
     }
@@ -50,18 +48,21 @@ impl Exchange for ReclusterPartitionExchange {
         // Scatter the data block to different partitions.
         let indices = range_ids
             .iter()
-            .map(|&id| (id - self.start) as u16)
+            .map(|&id| (id % self.num_partitions as u64) as u16)
             .collect::<Vec<_>>();
         data_block.pop_columns(1);
-
-        let scatter_indices = DataBlock::divide_indices_by_scatter_size(&indices, self.width);
+        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, indices) in scatter_indices.into_iter().take(self.width).enumerate() {
+        for (partition_id, indices) in scatter_indices
+            .into_iter()
+            .take(self.num_partitions)
+            .enumerate()
+        {
             if !indices.is_empty() {
-                let target = (partition_id * n) / self.width;
                 let block = data_block.take_with_optimize_size(&indices)?;
-                output_data_blocks[target].push((partition_id, block));
+                output_data_blocks[partition_id % n].push((partition_id, block));
             }
         }
         log::info!("Recluster range exchange: {:?}", self.start_time.elapsed());

src/query/service/src/pipelines/processors/transforms/recluster/recluster_partition_strategy.rs

@@ -37,17 +37,6 @@ impl ReclusterPartitionStrategy {
 impl PartitionProcessStrategy for ReclusterPartitionStrategy {
     const NAME: &'static str = "Recluster";
 
-    fn calc_partitions(
-        &self,
-        processor_id: usize,
-        num_processors: usize,
-        num_partitions: usize,
-    ) -> Vec<usize> {
-        (0..num_partitions)
-            .filter(|&partition| (partition * num_processors) / num_partitions == processor_id)
-            .collect()
-    }
-
     /// Stream write each block, and flush it conditionally based on builder status
     /// and input size estimation.
     fn process_data_blocks(&self, data_blocks: Vec<DataBlock>) -> Result<Vec<DataBlock>> {
@@ -127,17 +116,6 @@ impl CompactPartitionStrategy {
 impl PartitionProcessStrategy for CompactPartitionStrategy {
     const NAME: &'static str = "Compact";
 
-    fn calc_partitions(
-        &self,
-        processor_id: usize,
-        num_processors: usize,
-        num_partitions: usize,
-    ) -> Vec<usize> {
-        (0..num_partitions)
-            .filter(|&partition| (partition * num_processors) / num_partitions == processor_id)
-            .collect()
-    }
-
     /// Collects blocks into batches and merges them via `concat` when size threshold is reached.
     fn process_data_blocks(&self, data_blocks: Vec<DataBlock>) -> Result<Vec<DataBlock>> {
         let blocks_num = data_blocks.len();

src/query/service/src/pipelines/processors/transforms/window/partition/partition_process_strategy.rs

@@ -22,14 +22,6 @@ use databend_common_settings::Settings;
 pub trait PartitionProcessStrategy: Send + Sync + 'static {
     const NAME: &'static str;
 
-    /// Partition assignment: map partition index to processor via proportional mapping.
-    fn calc_partitions(
-        &self,
-        processor_id: usize,
-        num_processors: usize,
-        num_partitions: usize,
-    ) -> Vec<usize>;
-
     fn process_data_blocks(&self, data_blocks: Vec<DataBlock>) -> Result<Vec<DataBlock>>;
 }
 
@@ -66,17 +58,6 @@ impl WindowPartitionStrategy {
 impl PartitionProcessStrategy for WindowPartitionStrategy {
     const NAME: &'static str = "Window";
 
-    fn calc_partitions(
-        &self,
-        processor_id: usize,
-        num_processors: usize,
-        num_partitions: usize,
-    ) -> Vec<usize> {
-        (0..num_partitions)
-            .filter(|&partition| partition % num_processors == processor_id)
-            .collect()
-    }
-
     fn process_data_blocks(&self, data_blocks: Vec<DataBlock>) -> Result<Vec<DataBlock>> {
         let data_blocks = data_blocks
             .into_iter()

src/query/service/src/pipelines/processors/transforms/window/partition/transform_window_partition_collect.rs

@@ -92,7 +92,9 @@ impl<S: PartitionProcessStrategy> TransformPartitionCollect<S> {
         strategy: S,
     ) -> Result<Self> {
         // Calculate the partition ids collected by the processor.
-        let partitions = strategy.calc_partitions(processor_id, num_processors, num_partitions);
+        let partitions: Vec<usize> = (0..num_partitions)
+            .filter(|&partition| partition % num_processors == processor_id)
+            .collect();
 
         // Map each partition id to new partition id.
         let mut partition_id = vec![0; num_partitions];

src/query/service/src/schedulers/fragments/fragmenter.rs

@@ -15,7 +15,6 @@
 use std::sync::Arc;
 
 use databend_common_catalog::table_context::TableContext;
-use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_meta_types::NodeInfo;
 use databend_common_sql::executor::physical_plans::CompactSource;
@@ -27,7 +26,6 @@ use databend_common_sql::executor::physical_plans::ExchangeSink;
 use databend_common_sql::executor::physical_plans::ExchangeSource;
 use databend_common_sql::executor::physical_plans::FragmentKind;
 use databend_common_sql::executor::physical_plans::HashJoin;
-use databend_common_sql::executor::physical_plans::HilbertPartition;
 use databend_common_sql::executor::physical_plans::MutationSource;
 use databend_common_sql::executor::physical_plans::Recluster;
 use databend_common_sql::executor::physical_plans::ReplaceInto;
@@ -41,7 +39,6 @@ use crate::schedulers::PlanFragment;
 use crate::servers::flight::v1::exchange::BroadcastExchange;
 use crate::servers::flight::v1::exchange::DataExchange;
 use crate::servers::flight::v1::exchange::MergeExchange;
-use crate::servers::flight::v1::exchange::ModuloExchange;
 use crate::servers::flight::v1::exchange::ShuffleDataExchange;
 use crate::sessions::QueryContext;
 use crate::sql::executor::PhysicalPlan;
@@ -69,7 +66,6 @@ enum State {
     Compact,
     Recluster,
     Other,
-    HilbertRecluster,
 }
 
 impl Fragmenter {
@@ -118,15 +114,6 @@ impl Fragmenter {
                 FragmentKind::Expansive => {
                     Ok(Some(BroadcastExchange::create(Self::get_executors(ctx))))
                 }
-                FragmentKind::Modulo => {
-                    if plan.keys.len() != 1 {
-                        return Err(ErrorCode::Internal("Modulo exchange require one key"));
-                    }
-                    Ok(Some(ModuloExchange::create(
-                        Self::get_executors(ctx),
-                        plan.keys[0].clone(),
-                    )))
-                }
                 _ => Ok(None),
             },
             _ => Ok(None),
@@ -213,15 +200,6 @@ impl PhysicalPlanReplacer for Fragmenter {
         Ok(PhysicalPlan::Recluster(Box::new(plan.clone())))
     }
 
-    fn replace_hilbert_serialize(&mut self, plan: &HilbertPartition) -> Result<PhysicalPlan> {
-        let input = self.replace(&plan.input)?;
-        self.state = State::HilbertRecluster;
-        Ok(PhysicalPlan::HilbertPartition(Box::new(HilbertPartition {
-            input: Box::new(input),
-            ..plan.clone()
-        })))
-    }
-
     fn replace_compact_source(&mut self, plan: &CompactSource) -> Result<PhysicalPlan> {
         self.state = State::Compact;
         Ok(PhysicalPlan::CompactSource(Box::new(plan.clone())))
@@ -323,7 +301,6 @@ impl PhysicalPlanReplacer for Fragmenter {
             State::ReplaceInto => FragmentType::ReplaceInto,
             State::Compact => FragmentType::Compact,
             State::Recluster => FragmentType::Recluster,
-            State::HilbertRecluster => FragmentType::HilbertRecluster,
         };
         self.state = State::Other;
         let exchange = Self::get_exchange(self.ctx.clone(), &plan)?;