Reduce duplication between `BoundedAggregateStream` and `GroupedHashAggregateStream`

[alamb]

Is your feature request related to a problem or challenge?

We are trying to make hash based aggregation significantly faster -- see #4973

This will require some non trivial changes to the organization of how hash aggregation works. At the moment BoundedAggregateStream and GroupedHashAggregateStream both share significant amounts of code and so either we will have to duplicate the work to make hashing aggregation faster or else BoundedAggregateStream will not get the benefits.

Here is a visual depiction of the common code:

 meld datafusion/core/src/physical_plan/aggregates/bounded_aggregate_stream.rs datafusion/core/src/physical_plan/aggregates/row_hash.rs 

Describe the solution you'd like

Reduce duplication between BoundedAggregateStream and GroupedHashAggregateStream

The major differences are:

1. Choice of when output can be emitted
2. Clearing previous group state when groups have been emitted

Describe alternatives you've considered

No response

Additional context

No response

[alamb]

I wonder if you have any thoughts on this proposal @metesynnada and @mustafasrepo ?

[mustafasrepo]

During streaming support for aggregation I tried to reduce duplication between BoundedAggregateStream and GroupedHashAggregateStream. However, it reduced performance for some queries as seen in discussion

However, I think to benefit both implementations with the following major change. It is better to reduce duplication as much as possible even if it introduces minor overhead during benchmarks.

[alamb]

Thanks for the comments @mustafasrepo -- I have been studying the code and I am confident I can reduce the duplication. I hope to have a PR ready for review that brings us in this direction in a few days.

[alamb]

I plan to do this

[alamb]

I have completed my study of the code -- and I added some comments here #6881

Rather than try and consolidate the BoundedAggregateStream on main, what I think I would like to do is to merge in the new GroupByHash implementation - #4973 and then remove BoundedAggregateStream by porting its functionality to the new hash stream

[alamb]

I have a PR ready that consolidates the logic which I am quite pleased with: #6932