fix

Author: forsaken628

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

@@ -57,7 +57,7 @@ pub fn try_add_multi_sort_merge(
     remove_order_col: bool,
     enable_loser_tree: bool,
 ) -> Result<()> {
-    debug_assert!(remove_order_col == schema.has_field(ORDER_COL_NAME));
+    debug_assert!(!remove_order_col == schema.has_field(ORDER_COL_NAME));
 
     if pipeline.is_empty() {
         return Err(ErrorCode::Internal("Cannot resize empty pipe."));

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

@@ -94,13 +94,6 @@ impl PipelineBuilder {
             });
         }
 
-        let max_threads = self.settings.get_max_threads()? as usize;
-
-        // TODO(Winter): the query will hang in MultiSortMergeProcessor when max_threads == 1 and output_len != 1
-        if self.main_pipeline.output_len() == 1 || max_threads == 1 {
-            self.main_pipeline.try_resize(max_threads)?;
-        }
-
         let builder = SortPipelineBuilder::create(
             self.ctx.clone(),
             output_schema,
@@ -109,10 +102,14 @@ impl PipelineBuilder {
         )?
         .with_limit(sort.limit);
 
+        let max_threads = self.settings.get_max_threads()? as usize;
         match sort.step {
             SortStep::Single => {
                 // Build for single node mode.
                 // We build the full sort pipeline for it.
+                if max_threads == 1 {
+                    self.main_pipeline.try_resize(1)?;
+                }
                 builder
                     .remove_order_col_at_last()
                     .build_full_sort_pipeline(&mut self.main_pipeline)
@@ -122,6 +119,9 @@ impl PipelineBuilder {
                 // Build for each cluster node.
                 // We build the full sort pipeline for it.
                 // Don't remove the order column at last.
+                if max_threads == 1 {
+                    self.main_pipeline.try_resize(1)?;
+                }
                 builder.build_full_sort_pipeline(&mut self.main_pipeline)
             }
             SortStep::Final => {
@@ -130,6 +130,10 @@ impl PipelineBuilder {
                 // as the data is already sorted in each cluster node.
                 // The input number of the transform is equal to the number of cluster nodes.
                 if self.main_pipeline.output_len() > 1 {
+                    if self.main_pipeline.output_len() != 1 && max_threads == 1 {
+                        // TODO(Winter): the query will hang in MultiSortMergeProcessor when max_threads == 1 and output_len != 1
+                        self.main_pipeline.try_resize(1)?;
+                    }
                     builder
                         .remove_order_col_at_last()
                         .build_multi_merge(&mut self.main_pipeline)
@@ -141,6 +145,9 @@ impl PipelineBuilder {
             }
 
             SortStep::Sample => {
+                if max_threads == 1 {
+                    self.main_pipeline.try_resize(1)?;
+                }
                 builder.build_sample(&mut self.main_pipeline)?;
                 self.exchange_injector = TransformSortBuilder::exchange_injector();
                 Ok(())
@@ -155,6 +162,10 @@ impl PipelineBuilder {
                     self.build_pipeline(&sort.input)?;
                 }
 
+                if self.main_pipeline.output_len() != 1 && max_threads == 1 {
+                    // TODO(Winter): the query will hang in MultiSortMergeProcessor when max_threads == 1 and output_len != 1
+                    unimplemented!();
+                }
                 builder
                     .remove_order_col_at_last()
                     .build_bounded_merge_sort(&mut self.main_pipeline)

src/query/sql/src/executor/physical_plans/physical_sort.rs

@@ -228,7 +228,8 @@ impl PhysicalPlanBuilder {
             }));
         };
 
-        if !self.ctx.get_settings().get_enable_shuffle_sort()? {
+        let settings = self.ctx.get_settings();
+        if settings.get_max_threads()? == 1 || !settings.get_enable_shuffle_sort()? {
             let input_plan = self.build(s_expr.unary_child(), required).await?;
             return if !after_exchange {
                 Ok(PhysicalPlan::Sort(Sort {

tests/sqllogictests/suites/base/09_fuse_engine/09_0041_auto_compaction.test

@@ -7,6 +7,9 @@ use i15760;
 statement ok
 set auto_compaction_imperfect_blocks_threshold = 3;
 
+statement ok
+set enable_shuffle_sort = 0;
+
 statement ok
 set enable_parallel_multi_merge_sort = 0;
 

tests/sqllogictests/suites/mode/cluster/lazy_read.test

@@ -4,6 +4,9 @@ create or replace table t_lazy (a int not null, b float not null, c string not n
 statement ok
 set max_threads = 4;
 
+statement ok
+set enable_shuffle_sort = 0;
+
 statement ok
 insert into t_lazy select number + 1, 1.1, '1.1', (1,2), '2020-01-01' from numbers(100)
 
@@ -30,14 +33,14 @@ RowFetch
     ├── limit: 2
     ├── offset: 0
     ├── estimated rows: 2.00
-    └── Sort
+    └── Sort(Final)
         ├── output columns: [t_lazy.a (#0), t_lazy._row_id (#7)]
         ├── sort keys: [a DESC NULLS LAST]
         ├── estimated rows: 300.00
         └── Exchange
             ├── output columns: [t_lazy.a (#0), t_lazy._row_id (#7), #_order_col]
             ├── exchange type: Merge
-            └── Sort
+            └── Sort(Partial)
                 ├── output columns: [t_lazy.a (#0), t_lazy._row_id (#7), #_order_col]
                 ├── sort keys: [a DESC NULLS LAST]
                 ├── estimated rows: 300.00
@@ -98,14 +101,14 @@ Limit
 ├── limit: 2
 ├── offset: 0
 ├── estimated rows: 2.00
-└── Sort
+└── Sort(Final)
     ├── output columns: [t_lazy.a (#0), t_lazy.b (#1), t_lazy.c (#2), t_lazy.d (#3), t_lazy.e (#6)]
     ├── sort keys: [a DESC NULLS LAST]
     ├── estimated rows: 300.00
     └── Exchange
         ├── output columns: [t_lazy.a (#0), t_lazy.b (#1), t_lazy.c (#2), t_lazy.d (#3), t_lazy.e (#6), #_order_col]
         ├── exchange type: Merge
-        └── Sort
+        └── Sort(Partial)
             ├── output columns: [t_lazy.a (#0), t_lazy.b (#1), t_lazy.c (#2), t_lazy.d (#3), t_lazy.e (#6), #_order_col]
             ├── sort keys: [a DESC NULLS LAST]
             ├── estimated rows: 300.00

tests/sqllogictests/suites/mode/cluster/shuffle.test

@@ -22,14 +22,14 @@ set enable_shuffle_sort = 0;
 query T
 explain select * from (select t1.number, t2.number from t1 right outer join (SELECT number FROM t2 QUALIFY row_number() OVER (PARTITION BY number ORDER BY number DESC ) = 1) AS t2 ON t1.number = t2.number) as tt(a, b) order by a;
 ----
-Sort
+Sort(Final)
 ├── output columns: [t1.number (#0), t2.number (#1)]
 ├── sort keys: [number ASC NULLS LAST]
 ├── estimated rows: 0.01
 └── Exchange
     ├── output columns: [t1.number (#0), t2.number (#1), #_order_col]
     ├── exchange type: Merge
-    └── Sort
+    └── Sort(Partial)
         ├── output columns: [t1.number (#0), t2.number (#1), #_order_col]
         ├── sort keys: [number ASC NULLS LAST]
         ├── estimated rows: 0.01

tests/sqllogictests/suites/mode/cluster/window.test

@@ -200,7 +200,7 @@ Limit
 ├── limit: 10
 ├── offset: 0
 ├── estimated rows: 0.00
-└── Sort
+└── Sort(Single)
     ├── output columns: [sales.customer_id (#2), customer_avg (#8), diff_from_overall_avg (#9)]
     ├── sort keys: [diff_from_overall_avg DESC NULLS LAST, customer_id ASC NULLS LAST]
     ├── estimated rows: 0.00

tests/sqllogictests/suites/query/functions/02_0062_function_unnest.test

@@ -158,10 +158,7 @@ NULL [1,2] 3
 5 [3,4,5] 3
 
 statement ok
-drop table t;
-
-statement ok
-create table t (a array(int));
+create or replace table t (a array(int));
 
 statement ok
 insert into t values ([1,2])