chore: make standalone work

Author: dqhl76

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

@@ -143,6 +143,7 @@ impl AggregatePayloadWriters {
             let start = Instant::now();
             self.writers[bucket].write_block(block)?;
 
+            // TODO: the time recorded may not be accurate due to the serialization is included
             let elapsed = start.elapsed();
             self.write_stats.accumulate(elapsed);
         }

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

@@ -61,6 +61,13 @@ impl AccumulatingTransform for NewTransformAggregateSpillWriter {
                 let meta = data.take_meta().unwrap();
                 let aggregate_meta = AggregateMeta::downcast_from(meta).unwrap();
                 if let AggregateMeta::AggregateSpilling(partition) = aggregate_meta {
+                    //  we use fixed size partitioning, the different bucket number will caused spilled data can't be merged correctly
+                    debug_assert_eq!(
+                        partition.payloads.len(),
+                        MAX_AGGREGATE_HASHTABLE_BUCKETS_NUM,
+                        "the number of payloads should be equal to MAX_AGGREGATE_HASHTABLE_BUCKETS_NUM for spilling partition"
+                    );
+
                     for (bucket, payload) in partition.payloads.into_iter().enumerate() {
                         if payload.len() == 0 {
                             continue;

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

@@ -294,7 +294,33 @@ impl TransformPartitionBucketScatter {
 
                         (payload.bucket, payload.max_partition_count)
                     }
-                    AggregateMeta::NewBucketSpilled(_) => unreachable!(),
+                    AggregateMeta::NewBucketSpilled(_) => {
+                        let meta = data_block.take_meta().unwrap();
+
+                        if let Some(AggregateMeta::NewBucketSpilled(payload)) =
+                            AggregateMeta::downcast_from(meta)
+                        {
+                            let bucket = payload.bucket;
+                            let partition_count = MAX_PARTITION_COUNT;
+                            self.max_partition_count =
+                                self.max_partition_count.max(partition_count);
+
+                            let data_block = DataBlock::empty_with_meta(
+                                AggregateMeta::create_new_spilled(payload),
+                            );
+                            match self.buckets_blocks.entry(bucket) {
+                                Entry::Vacant(v) => {
+                                    v.insert(vec![data_block]);
+                                }
+                                Entry::Occupied(mut v) => {
+                                    v.get_mut().push(data_block);
+                                }
+                            };
+
+                            return Ok((SINGLE_LEVEL_BUCKET_NUM, MAX_PARTITION_COUNT));
+                        }
+                        unreachable!()
+                    }
                 };
             } else {
                 return Err(ErrorCode::Internal(format!(