refactor: tune aggregate spill settings to enhance performance

src/query/service/src/physical_plans/physical_aggregate_partial.rs

@@ -238,7 +238,7 @@ impl IPhysicalPlan for AggregatePartial {
             };
             let shared_partition_streams = SharedPartitionStream::new(
                 builder.main_pipeline.output_len(),
-                max_block_rows,
+                0,
                 max_block_bytes,
                 bucket_num,
             );

src/query/service/src/pipelines/processors/transforms/aggregator/new_aggregate/new_transform_final_aggregate.rs

@@ -47,7 +47,7 @@ use crate::pipelines::processors::transforms::aggregator::statistics::Aggregatio
 use crate::pipelines::processors::transforms::aggregator::transform_aggregate_partial::HashTable;
 use crate::sessions::QueryContext;
 
-const SPILL_BUCKET_NUM: usize = 2;
+const SPILL_BUCKET_NUM: usize = 4;
 const SPILL_BUCKET_BITS: u64 = SPILL_BUCKET_NUM.trailing_zeros() as u64;
 
 enum Stage {
@@ -113,11 +113,7 @@ impl NewTransformFinalAggregate {
             ctx.clone(),
             SPILL_BUCKET_NUM,
             params.spill_schema(),
-            LocalPartitionStream::new(
-                params.max_block_rows,
-                params.max_block_bytes,
-                SPILL_BUCKET_NUM,
-            ),
+            LocalPartitionStream::new(0, params.max_block_bytes, SPILL_BUCKET_NUM),
         )?;
 
         Ok(Box::new(NewTransformFinalAggregate {

src/query/service/src/spillers/async_buffer.rs

@@ -520,6 +520,7 @@ impl SpillsDataWriter {
                     .set_compression(Compression::LZ4_RAW)
                     .set_statistics_enabled(EnabledStatistics::None)
                     .set_bloom_filter_enabled(false)
+                    .set_dictionary_enabled(false)
                     .build();
 
                 let arrow_schema = Arc::new(Schema::from(table_schema.as_ref()));