[BugFix] Fix chunk accumulator miss assign _max_size (backport #52579)

be/src/exec/pipeline/aggregate/spillable_aggregate_blocking_source_operator.cpp

@@ -24,6 +24,7 @@ Status SpillableAggregateBlockingSourceOperator::prepare(RuntimeState* state) {
     RETURN_IF_ERROR(AggregateBlockingSourceOperator::prepare(state));
     RETURN_IF_ERROR(_stream_aggregator->prepare(state, state->obj_pool(), _unique_metrics.get()));
     RETURN_IF_ERROR(_stream_aggregator->open(state));
+    _accumulator.set_max_size(state->chunk_size());
     return Status::OK();
 }
 

be/src/exec/pipeline/aggregate/spillable_aggregate_distinct_blocking_operator.cpp

@@ -170,6 +170,7 @@ Status SpillableAggregateDistinctBlockingSourceOperator::prepare(RuntimeState* s
     RETURN_IF_ERROR(AggregateDistinctBlockingSourceOperator::prepare(state));
     RETURN_IF_ERROR(_stream_aggregator->prepare(state, state->obj_pool(), _unique_metrics.get()));
     RETURN_IF_ERROR(_stream_aggregator->open(state));
+    _accumulator.set_max_size(state->chunk_size());
     return Status::OK();
 }
 

be/src/exec/pipeline/hash_partition_context.cpp

@@ -37,6 +37,7 @@ Status HashPartitionContext::prepare(RuntimeState* state, RuntimeProfile* profil
         _has_nullable_key = _has_nullable_key || _partition_types[i].is_nullable;
     }
 
+    _acc.set_max_size(state->chunk_size());
     _chunks_partitioner = std::make_unique<ChunksPartitioner>(_has_nullable_key, _partition_exprs, _partition_types);
     return _chunks_partitioner->prepare(state, profile);
 }

test/sql/test_spill/R/test_spill_aggregate

@@ -28,4 +28,4 @@ insert into t2 select sum(k1),k2 from t1 group by k2;
 select * from t2;
 -- result:
 1	1
--- !result
+-- !result

test/sql/test_spill/T/test_spill_aggregate

@@ -15,4 +15,4 @@ DUPLICATE KEY(k1)
 DISTRIBUTED BY HASH(k1) PROPERTIES('replication_num'='1');
 insert into t1 values (1,"1");
 insert into t2 select sum(k1),k2 from t1 group by k2;
-select * from t2;
+select * from t2;