refactor(query): Combining TransformSortMergeLimit, TransformSortMerge, and TransformSortSpill (#17762)

* transform_sort

* update

* fix

* fix

* fix

* fix

* refine

* fix

* refine

* fix

* move next_index

* refine

* arc

* refine

* memory_merger

* fix

Author: forsaken628

src/query/expression/src/row/row_converter.rs

@@ -38,6 +38,7 @@ use crate::SortField;
 /// Convert column-oriented data into comparable row-oriented data.
 ///
 /// **NOTE**: currently, Variant is treat as String.
+#[derive(Debug)]
 pub struct RowConverter {
     fields: Arc<[SortField]>,
 }

src/query/pipeline/transforms/src/processors/transforms/sort/mod.rs

@@ -19,12 +19,10 @@ mod list_domain;
 mod loser_tree;
 mod merger;
 mod rows;
-mod spill;
 pub mod utils;
 
 pub use cursor::*;
 pub use k_way_merge_sort_partition::KWaySortPartitioner;
 pub use k_way_merge_sort_partition::SortTaskMeta;
 pub use merger::*;
 pub use rows::*;
-pub use spill::*;

src/query/pipeline/transforms/src/processors/transforms/sort/rows/common.rs

@@ -77,7 +77,7 @@ impl RowConverter<BinaryColumn> for CommonRowConverter {
         CommonRowConverter::new(sort_fields)
     }
 
-    fn convert(&mut self, columns: &[BlockEntry], num_rows: usize) -> Result<BinaryColumn> {
+    fn convert(&self, columns: &[BlockEntry], num_rows: usize) -> Result<BinaryColumn> {
         let columns = columns
             .iter()
             .map(|entry| match &entry.value {

src/query/pipeline/transforms/src/processors/transforms/sort/rows/mod.rs

@@ -32,13 +32,13 @@ pub use utils::*;
 
 /// Convert columns to rows.
 pub trait RowConverter<T: Rows>
-where Self: Sized
+where Self: Sized + Debug
 {
     fn create(
         sort_columns_descriptions: &[SortColumnDescription],
         output_schema: DataSchemaRef,
     ) -> Result<Self>;
-    fn convert(&mut self, columns: &[BlockEntry], num_rows: usize) -> Result<T>;
+    fn convert(&self, columns: &[BlockEntry], num_rows: usize) -> Result<T>;
 }
 
 /// Rows can be compared.

src/query/pipeline/transforms/src/processors/transforms/sort/rows/simple.rs

@@ -117,6 +117,7 @@ where
 
 /// If there is only one sort field and its type is a primitive type,
 /// use this converter.
+#[derive(Debug)]
 pub struct SimpleRowConverter<T> {
     _t: PhantomData<T>,
 }
@@ -135,7 +136,7 @@ where
         Ok(Self { _t: PhantomData })
     }
 
-    fn convert(&mut self, columns: &[BlockEntry], num_rows: usize) -> Result<SimpleRowsAsc<T>> {
+    fn convert(&self, columns: &[BlockEntry], num_rows: usize) -> Result<SimpleRowsAsc<T>> {
         self.convert_rows(columns, num_rows, true)
     }
 }
@@ -154,14 +155,14 @@ where
         Ok(Self { _t: PhantomData })
     }
 
-    fn convert(&mut self, columns: &[BlockEntry], num_rows: usize) -> Result<SimpleRowsDesc<T>> {
+    fn convert(&self, columns: &[BlockEntry], num_rows: usize) -> Result<SimpleRowsDesc<T>> {
         self.convert_rows(columns, num_rows, false)
     }
 }
 
 impl<T: ArgType> SimpleRowConverter<T> {
     fn convert_rows<R: Rows>(
-        &mut self,
+        &self,
         columns: &[BlockEntry],
         num_rows: usize,
         asc: bool,

src/query/pipeline/transforms/src/processors/transforms/sort/rows/utils.rs

@@ -24,6 +24,7 @@ use databend_common_expression::with_number_mapped_type;
 use databend_common_expression::BlockEntry;
 use databend_common_expression::Column;
 use databend_common_expression::DataBlock;
+use databend_common_expression::DataSchema;
 use databend_common_expression::DataSchemaRef;
 use databend_common_expression::SortColumnDescription;
 use match_template::match_template;
@@ -86,8 +87,8 @@ fn convert_columns<R: Rows, C: RowConverter<R>>(
     columns: &[BlockEntry],
     num_rows: usize,
 ) -> Result<Column> {
-    let mut converter = C::create(sort_desc, schema)?;
-    let rows = C::convert(&mut converter, columns, num_rows)?;
+    let converter = C::create(sort_desc, schema)?;
+    let rows = C::convert(&converter, columns, num_rows)?;
     Ok(rows.to_column())
 }
 
@@ -103,25 +104,25 @@ pub fn select_row_type(visitor: &mut impl RowsTypeVisitor) {
         match sort_type {
             DataType::T => {
                 if asc {
-                    visitor.visit_type::<SimpleRowsAsc<T>>()
+                    visitor.visit_type::<SimpleRowsAsc<T>, SimpleRowConverter<T>>()
                 } else {
-                    visitor.visit_type::<SimpleRowsDesc<T>>()
+                    visitor.visit_type::<SimpleRowsDesc<T>, SimpleRowConverter<T>>()
                 }
             },
             DataType::Number(num_ty) => with_number_mapped_type!(|NUM_TYPE| match num_ty {
                 NumberDataType::NUM_TYPE => {
                     if asc {
-                        visitor.visit_type::<SimpleRowsAsc<NumberType<NUM_TYPE>>>()
+                        visitor.visit_type::<SimpleRowsAsc<NumberType<NUM_TYPE>>, SimpleRowConverter<NumberType<NUM_TYPE>>>()
                     } else {
-                        visitor.visit_type::<SimpleRowsDesc<NumberType<NUM_TYPE>>>()
+                        visitor.visit_type::<SimpleRowsDesc<NumberType<NUM_TYPE>>, SimpleRowConverter<NumberType<NUM_TYPE>>>()
                     }
                 }
             }),
-            _ => visitor.visit_type::<CommonRows>()
+            _ => visitor.visit_type::<CommonRows, CommonConverter>()
             }
         }
     } else {
-        visitor.visit_type::<CommonRows>()
+        visitor.visit_type::<CommonRows, CommonConverter>()
     }
 }
 
@@ -130,5 +131,26 @@ pub trait RowsTypeVisitor {
 
     fn sort_desc(&self) -> &[SortColumnDescription];
 
-    fn visit_type<R: Rows + 'static>(&mut self);
+    fn visit_type<R, C>(&mut self)
+    where
+        R: Rows + 'static,
+        C: RowConverter<R> + Send + 'static;
+}
+
+pub fn order_field_type(schema: &DataSchema, desc: &[SortColumnDescription]) -> DataType {
+    debug_assert!(!desc.is_empty());
+    if desc.len() == 1 {
+        let order_by_field = schema.field(desc[0].offset);
+        if matches!(
+            order_by_field.data_type(),
+            DataType::Number(_)
+                | DataType::Date
+                | DataType::Timestamp
+                | DataType::Binary
+                | DataType::String
+        ) {
+            return order_by_field.data_type().clone();
+        }
+    }
+    DataType::Binary
 }

src/query/pipeline/transforms/src/processors/transforms/sort/spill.rs

@@ -14,13 +14,14 @@
 
 use std::collections::BinaryHeap;
 
-use databend_common_expression::types::DataType;
 use databend_common_expression::DataField;
 use databend_common_expression::DataSchema;
 use databend_common_expression::DataSchemaRef;
 use databend_common_expression::DataSchemaRefExt;
 use databend_common_expression::SortColumnDescription;
 
+use super::order_field_type;
+
 pub const ORDER_COL_NAME: &str = "_order_col";
 
 /// Find the bigger child of the root of the heap.
@@ -35,33 +36,13 @@ pub fn find_bigger_child_of_root<T: Ord>(heap: &BinaryHeap<T>) -> &T {
     }
 }
 
-#[inline(always)]
-fn order_field_type(schema: &DataSchema, desc: &[SortColumnDescription]) -> DataType {
-    debug_assert!(!desc.is_empty());
-    if desc.len() == 1 {
-        let order_by_field = schema.field(desc[0].offset);
-        if matches!(
-            order_by_field.data_type(),
-            DataType::Number(_)
-                | DataType::Date
-                | DataType::Timestamp
-                | DataType::Binary
-                | DataType::String
-        ) {
-            return order_by_field.data_type().clone();
-        }
-    }
-    DataType::Binary
-}
-
 pub fn has_order_field(schema: &DataSchema) -> bool {
     schema
         .fields
         .last()
         .is_some_and(|f| f.name() == ORDER_COL_NAME)
 }
 
-#[inline(always)]
 pub fn add_order_field(schema: DataSchemaRef, desc: &[SortColumnDescription]) -> DataSchemaRef {
     if has_order_field(&schema) {
         schema

src/query/pipeline/transforms/src/processors/transforms/sort/utils.rs

@@ -56,7 +56,7 @@ pub fn add_k_way_merge_sort(
     worker: usize,
     block_size: usize,
     limit: Option<usize>,
-    sort_desc: Arc<Vec<SortColumnDescription>>,
+    sort_desc: Arc<[SortColumnDescription]>,
     remove_order_col: bool,
     enable_loser_tree: bool,
 ) -> Result<()> {

src/query/pipeline/transforms/src/processors/transforms/transform_k_way_merge_sort.rs

@@ -53,7 +53,7 @@ pub fn try_add_multi_sort_merge(
     schema: DataSchemaRef,
     block_size: usize,
     limit: Option<usize>,
-    sort_columns_descriptions: Arc<Vec<SortColumnDescription>>,
+    sort_columns_descriptions: Arc<[SortColumnDescription]>,
     remove_order_col: bool,
     enable_loser_tree: bool,
 ) -> Result<()> {
@@ -104,12 +104,10 @@ fn create_processor(
     schema: DataSchemaRef,
     block_size: usize,
     limit: Option<usize>,
-    sort_columns_descriptions: Arc<Vec<SortColumnDescription>>,
+    sort_desc: Arc<[SortColumnDescription]>,
     remove_order_col: bool,
     enable_loser_tree: bool,
 ) -> Result<Box<dyn Processor>> {
-    let sort_desc = sort_columns_descriptions;
-
     macro_rules! create {
         ($algo:ident, $rows:ty) => {
             MultiSortMergeProcessor::<$algo<$rows>>::create(
@@ -118,7 +116,6 @@ fn create_processor(
                 schema,
                 block_size,
                 limit,
-                sort_desc,
                 remove_order_col,
             )?
         };
@@ -220,7 +217,6 @@ where A: SortAlgorithm
         schema: DataSchemaRef,
         block_size: usize,
         limit: Option<usize>,
-        _sort_desc: Arc<Vec<SortColumnDescription>>,
         remove_order_col: bool,
     ) -> Result<Self> {
         let streams = inputs