for test

Author: zhyass

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

@@ -24,18 +24,22 @@ use crate::pipelines::processors::transforms::WindowPartitionMeta;
 pub struct ReclusterPartitionExchange {
     start: u64,
     width: usize,
+    start_time: Instant,
 }
 
 impl ReclusterPartitionExchange {
     pub fn create(start: u64, width: usize) -> Arc<ReclusterPartitionExchange> {
-        Arc::new(ReclusterPartitionExchange { start, width })
+        Arc::new(ReclusterPartitionExchange {
+            start,
+            width,
+            start_time: Instant::now(),
+        })
     }
 }
 
 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()
@@ -60,7 +64,7 @@ impl Exchange for ReclusterPartitionExchange {
                 output_data_blocks[target].push((partition_id, block));
             }
         }
-        log::info!("Recluster range exchange: {:?}", start.elapsed());
+        log::info!("Recluster range exchange: {:?}", self.start_time.elapsed());
 
         // Union data blocks for each processor.
         Ok(output_data_blocks

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

@@ -162,6 +162,7 @@ impl Processor for TransformReclusterPartition {
     fn process(&mut self) -> Result<()> {
         match std::mem::replace(&mut self.step, Step::Consume) {
             Step::Collect => {
+                let start_cost = self.start.elapsed();
                 let data_block = self.input.pull_data().unwrap()?;
                 if let Some(meta) = data_block
                     .get_owned_meta()
@@ -212,6 +213,11 @@ impl Processor for TransformReclusterPartition {
                         }
                     }
                 }
+                log::info!(
+                    "Recluster: start collect: {:?}, end: {:?}",
+                    start_cost,
+                    self.start.elapsed()
+                );
             }
             Step::Flush => {
                 while let Some(mut partition_data) = self.partition_data.pop() {

tests/sqllogictests/suites/mode/standalone/explain/window.test

@@ -59,7 +59,7 @@ digraph {
     2 [ label = "DeserializeDataTransform" ]
     3 [ label = "ShufflePartition(Window)" ]
     4 [ label = "ShuffleMergePartition(Window)" ]
-    5 [ label = "TransformWindowPartitionCollect(Sort)" ]
+    5 [ label = "TransformPartitionCollect(Window)" ]
     6 [ label = "Transform Window" ]
     7 [ label = "Resize" ]
     8 [ label = "SortPartialTransform" ]
@@ -108,7 +108,7 @@ digraph {
     2 [ label = "DeserializeDataTransform" ]
     3 [ label = "ShufflePartition(Window)" ]
     4 [ label = "ShuffleMergePartition(Window)" ]
-    5 [ label = "TransformWindowPartitionCollect(Sort)" ]
+    5 [ label = "TransformPartitionCollect(Window)" ]
     6 [ label = "Transform Window" ]
     7 [ label = "Resize" ]
     8 [ label = "SortPartialTransform" ]
@@ -429,7 +429,7 @@ digraph {
     2 [ label = "DeserializeDataTransform" ]
     3 [ label = "ShufflePartition(Window)" ]
     4 [ label = "ShuffleMergePartition(Window)" ]
-    5 [ label = "TransformWindowPartitionCollect(Sort)" ]
+    5 [ label = "TransformPartitionCollect(Window)" ]
     6 [ label = "Transform Window" ]
     7 [ label = "LimitTransform" ]
     8 [ label = "CompoundBlockOperator(Project)" ]
@@ -457,7 +457,7 @@ digraph {
     2 [ label = "DeserializeDataTransform" ]
     3 [ label = "ShufflePartition(Window)" ]
     4 [ label = "ShuffleMergePartition(Window)" ]
-    5 [ label = "TransformWindowPartitionCollect(Sort)" ]
+    5 [ label = "TransformPartitionCollect(Window)" ]
     6 [ label = "Transform Window" ]
     7 [ label = "LimitTransform" ]
     8 [ label = "CompoundBlockOperator(Project)" ]
@@ -486,7 +486,7 @@ digraph {
     2 [ label = "DeserializeDataTransform" ]
     3 [ label = "ShufflePartition(Window)" ]
     4 [ label = "ShuffleMergePartition(Window)" ]
-    5 [ label = "TransformWindowPartitionCollect(Sort)" ]
+    5 [ label = "TransformPartitionCollect(Window)" ]
     6 [ label = "Transform Window" ]
     7 [ label = "LimitTransform" ]
     8 [ label = "CompoundBlockOperator(Project)" ]
@@ -510,7 +510,7 @@ digraph {
     2 [ label = "DeserializeDataTransform" ]
     3 [ label = "ShufflePartition(Window)" ]
     4 [ label = "ShuffleMergePartition(Window)" ]
-    5 [ label = "TransformWindowPartitionCollect(Sort)" ]
+    5 [ label = "TransformPartitionCollect(Window)" ]
     6 [ label = "Transform Window" ]
     7 [ label = "LimitTransform" ]
     8 [ label = "CompoundBlockOperator(Project)" ]
@@ -534,7 +534,7 @@ digraph {
     2 [ label = "DeserializeDataTransform" ]
     3 [ label = "ShufflePartition(Window)" ]
     4 [ label = "ShuffleMergePartition(Window)" ]
-    5 [ label = "TransformWindowPartitionCollect(Sort)" ]
+    5 [ label = "TransformPartitionCollect(Window)" ]
     6 [ label = "Transform Window" ]
     7 [ label = "LimitTransform" ]
     8 [ label = "CompoundBlockOperator(Project)" ]
@@ -559,7 +559,7 @@ digraph {
     2 [ label = "DeserializeDataTransform" ]
     3 [ label = "ShufflePartition(Window)" ]
     4 [ label = "ShuffleMergePartition(Window)" ]
-    5 [ label = "TransformWindowPartitionCollect(Sort)" ]
+    5 [ label = "TransformPartitionCollect(Window)" ]
     6 [ label = "Transform Window" ]
     7 [ label = "Resize" ]
     8 [ label = "SortPartialTransform" ]
@@ -610,7 +610,7 @@ digraph {
     4 [ label = "TransformFilter" ]
     5 [ label = "ShufflePartition(Window)" ]
     6 [ label = "ShuffleMergePartition(Window)" ]
-    7 [ label = "TransformWindowPartitionCollect(Sort)" ]
+    7 [ label = "TransformPartitionCollect(Window)" ]
     8 [ label = "Transform Window" ]
     9 [ label = "Resize" ]
     10 [ label = "SortPartialTransform" ]
@@ -708,7 +708,7 @@ digraph {
     2 [ label = "DeserializeDataTransform" ]
     3 [ label = "ShufflePartition(WindowTopN)" ]
     4 [ label = "ShuffleMergePartition(WindowTopN)" ]
-    5 [ label = "TransformWindowPartitionCollect(Sort)" ]
+    5 [ label = "TransformPartitionCollect(Window)" ]
     6 [ label = "Transform Window" ]
     7 [ label = "TransformFilter" ]
     8 [ label = "LimitTransform" ]
@@ -786,12 +786,12 @@ digraph {
     1 [ label = "CompoundBlockOperator(Map)" ]
     2 [ label = "ShufflePartition(Window)" ]
     3 [ label = "ShuffleMergePartition(Window)" ]
-    4 [ label = "TransformWindowPartitionCollect(Sort)" ]
+    4 [ label = "TransformPartitionCollect(Window)" ]
     5 [ label = "Transform Window" ]
     6 [ label = "CompoundBlockOperator(Map)" ]
     7 [ label = "ShufflePartition(Window)" ]
     8 [ label = "ShuffleMergePartition(Window)" ]
-    9 [ label = "TransformWindowPartitionCollect(Sort)" ]
+    9 [ label = "TransformPartitionCollect(Window)" ]
     10 [ label = "Transform Window" ]
     11 [ label = "CompoundBlockOperator(Project)" ]
     0 -> 1 [ label = "" ]