for test

Author: zhyass

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

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::sync::Arc;
+use std::time::Instant;
 
 use databend_common_exception::Result;
 use databend_common_expression::DataBlock;
@@ -34,6 +35,7 @@ impl ReclusterPartitionExchange {
 impl Exchange for ReclusterPartitionExchange {
     const NAME: &'static str = "Recluster";
     fn partition(&self, mut data_block: DataBlock, n: usize) -> Result<Vec<DataBlock>> {
+        let start = Instant::now();
         let range_ids = data_block
             .get_last_column()
             .as_number()
@@ -58,6 +60,7 @@ impl Exchange for ReclusterPartitionExchange {
                 output_data_blocks[target].push((partition_id, block));
             }
         }
+        log::info!("Recluster range exchange: {:?}", start.elapsed());
 
         // Union data blocks for each processor.
         Ok(output_data_blocks

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

@@ -15,6 +15,7 @@
 use std::any::Any;
 use std::collections::VecDeque;
 use std::sync::Arc;
+use std::time::Instant;
 
 use databend_common_exception::Result;
 use databend_common_expression::types::ArgType;
@@ -45,6 +46,8 @@ where T: ArgType
     output_data: VecDeque<DataBlock>,
     bounds: Vec<T::Scalar>,
     max_value: Option<T::Scalar>,
+
+    start: Instant,
 }
 
 impl<T> TransformRangePartitionIndexer<T>
@@ -65,6 +68,7 @@ where
             output_data: VecDeque::new(),
             bounds: vec![],
             max_value: None,
+            start: Instant::now(),
         })
     }
 }
@@ -121,6 +125,7 @@ where
             .expect("require a ReclusterSampleMeta");
         self.input_data = meta.blocks;
         self.state.merge_sample::<T>(meta.sample_values)?;
+        log::info!("Recluster range partition: {:?}", self.start.elapsed());
         Ok(Event::Async)
     }
 

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

@@ -15,6 +15,7 @@
 use std::any::Any;
 use std::collections::VecDeque;
 use std::sync::Arc;
+use std::time::Instant;
 
 use databend_common_exception::Result;
 use databend_common_expression::BlockMetaInfoDowncast;
@@ -68,6 +69,9 @@ pub struct TransformReclusterPartition {
     partition_data: Vec<PartitionData>,
     output_data: VecDeque<DataBlock>,
 
+    start: Instant,
+    cnt: usize,
+
     step: Step,
 }
 
@@ -98,6 +102,8 @@ impl TransformReclusterPartition {
                 partition_data,
                 output_data: VecDeque::new(),
                 step: Step::Consume,
+                start: Instant::now(),
+                cnt: 0,
             },
         )))
     }
@@ -133,6 +139,10 @@ impl Processor for TransformReclusterPartition {
 
         if self.input.is_finished() {
             if !self.partition_data.is_empty() {
+                if self.cnt == 0 {
+                    log::info!("Recluster: start flush: {:?}", self.start.elapsed());
+                }
+                self.cnt += 1;
                 self.step = Step::Flush;
                 return Ok(Event::Sync);
             }