for test

Author: zhyass

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

@@ -130,42 +130,46 @@ where
     }
 
     fn process(&mut self) -> Result<()> {
-        if let Some(mut block) = self.input_data.pop() {
-            let bound_len = self.bounds.len();
-            let num_rows = block.num_rows();
-            let last = block.get_last_column().clone();
-            block.pop_columns(1);
-            let mut builder = Vec::with_capacity(num_rows);
-            let last_col = T::try_downcast_column(&last.remove_nullable()).unwrap();
-            for index in 0..num_rows {
-                let val =
-                    T::to_owned_scalar(unsafe { T::index_column_unchecked(&last_col, index) });
-                if self.max_value.as_ref().is_some_and(|v| val >= *v) {
-                    let range_id = bound_len + 1;
-                    builder.push(range_id as u64);
-                    continue;
-                }
+        let start = Instant::now();
+        let mut block = {
+            let blocks = std::mem::take(&mut self.input_data);
+            DataBlock::concat(&blocks)?
+        };
+
+        let bound_len = self.bounds.len();
+        let num_rows = block.num_rows();
+        let last = block.get_last_column().clone();
+        block.pop_columns(1);
+        let mut builder = Vec::with_capacity(num_rows);
+        let last_col = T::try_downcast_column(&last.remove_nullable()).unwrap();
+        for index in 0..num_rows {
+            let val = T::to_owned_scalar(unsafe { T::index_column_unchecked(&last_col, index) });
+            if self.max_value.as_ref().is_some_and(|v| val >= *v) {
+                let range_id = bound_len + 1;
+                builder.push(range_id as u64);
+                continue;
+            }
 
-                let mut low = 0;
-                let mut high = bound_len;
-                while low < high {
-                    let mid = low + ((high - low) / 2);
-                    let bound = unsafe { self.bounds.get_unchecked(mid) }.clone();
-                    if val > bound {
-                        low = mid + 1;
-                    } else {
-                        high = mid;
-                    }
+            let mut low = 0;
+            let mut high = bound_len;
+            while low < high {
+                let mid = low + ((high - low) / 2);
+                let bound = unsafe { self.bounds.get_unchecked(mid) }.clone();
+                if val > bound {
+                    low = mid + 1;
+                } else {
+                    high = mid;
                 }
-                builder.push(low as u64);
             }
-
-            block.add_column(BlockEntry::new(
-                DataType::Number(NumberDataType::UInt64),
-                Value::Column(UInt64Type::from_data(builder)),
-            ));
-            self.output_data.push_back(block);
+            builder.push(low as u64);
         }
+
+        block.add_column(BlockEntry::new(
+            DataType::Number(NumberDataType::UInt64),
+            Value::Column(UInt64Type::from_data(builder)),
+        ));
+        self.output_data.push_back(block);
+        log::info!("Recluster range output: {:?}", start.elapsed());
         Ok(())
     }