fix(query): fix panic if before_partial shuffle in cluster mode (#18081)

* fix(query): fix panic if before_partial shuffle in cluster mode

* fix(query): fix panic if before_partial shuffle in cluster mode

Author: zhang2014

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

@@ -163,8 +163,10 @@ impl PipelineBuilder {
             )?))
         })?;
 
+        let before_partial =
+            self.settings.get_group_by_shuffle_mode()?.to_lowercase() == "before_partial";
         // If cluster mode, spill write will be completed in exchange serialize, because we need scatter the block data first
-        if !self.is_exchange_neighbor {
+        if !self.is_exchange_neighbor || before_partial {
             let operator = DataOperator::instance().spill_operator();
             let location_prefix = self.ctx.query_id_spill_prefix();
 

tests/sqllogictests/suites/base/03_common/03_0038_spill_aggregator.test

@@ -24,6 +24,18 @@ create or replace table t2  as select number, max(number) from numbers(10000000)
 statement ok
 drop table t2
 
+statement ok
+set group_by_shuffle_mode = 'before_partial';
+
+onlyif http
+query T
+SELECT COUNT() FROM (SELECT number::string, count() FROM numbers_mt(100000) group by number::string);
+----
+100000
+
+statement ok
+unset group_by_shuffle_mode;
+
 onlyif http
 statement ok
 unset max_threads;