[fix](spill) Memory leak in the arena used by the aggregation source operator

After switching spill partition in partitioned aggregation source
operator, the agg_arena_pool also needs to be reset.

Author: dh-cloud

be/src/pipeline/exec/aggregation_source_operator.cpp

@@ -543,6 +543,14 @@ size_t AggSourceOperatorX::get_estimated_memory_size_for_merging(RuntimeState* s
     return size;
 }
 
+Status AggSourceOperatorX::reset_hash_table(RuntimeState* state) {
+    auto& local_state = get_local_state(state);
+    auto& ss = *local_state.Base::_shared_state;
+    RETURN_IF_ERROR(ss.reset_hash_table());
+    ss.agg_arena_pool.clear(true);
+    return Status::OK();
+}
+
 void AggLocalState::_emplace_into_hash_table(vectorized::AggregateDataPtr* places,
                                              vectorized::ColumnRawPtrs& key_columns,
                                              uint32_t num_rows) {

be/src/pipeline/exec/aggregation_source_operator.h

@@ -114,6 +114,8 @@ class AggSourceOperatorX : public OperatorX<AggLocalState> {
 
     size_t get_estimated_memory_size_for_merging(RuntimeState* state, size_t rows) const;
 
+    Status reset_hash_table(RuntimeState* state);
+
 private:
     friend class AggLocalState;
 

be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp

@@ -181,6 +181,8 @@ Status PartitionedAggSourceOperatorX::get_block(RuntimeState* state, vectorized:
                 local_state._need_to_merge_data_for_current_partition = true;
                 status = local_state._shared_state->in_mem_shared_state->reset_hash_table();
                 RETURN_IF_ERROR(status);
+                status = _agg_source_operator->reset_hash_table(runtime_state);
+                RETURN_IF_ERROR(status);
                 *eos = false;
             }
         }