Support StreamAggregation / streaming group by

[xiaoyong-z]

If the input group by columns has been sorted before the aggregation, we can enable stream aggregation, it is more efficient than HashAggregation.

[xiaoyong-z]

@alamb hello, it seems that datafusion currently doesn't have StreamAggregation. If no one works on this, i want to implement it.

[alamb]

Hi @xiaoyong-z -- that is great news! -- I believe that @metesynnada @ozankabak mentioned they wanted to work on this feature. Let's use this ticket to collaborate on a design.

I believe #1570 is also related as streaming grouping is often used to merge the spilled groups. @milenkovicm and I had some discussion about this #1570 (comment) but I never followed through on a writeup

[ozankabak]

This was on our roadmap and we would love to help out on this. @alamb, if you can share with us the papers/resources you mentioned on this we can digest them and share our thinking on the design. @xiaoyong-z, do you have a particular design in mind yet?

[alamb]

I will begin a google doc for us to collaborate on

[alamb]

Here is a google doc with some ideas https://docs.google.com/document/d/16rm5VR1nGkY6DedMCh1NUmThwf3RduAweaBH9b1h6AY/edit?usp=sharing

I have it in "comment" mode for everyone on the internet, but please feel free to request edit access and I will grant it

[ozankabak]

Thank you for putting this together, had an initial look. I expect us to do a deeper dive, take a look at the mentioned papers, and give meaningful comments in the next several days.

[Ted-Jiang]

Our team is also looking forward to this feature and the memory limited aggregation 👍

[xiaoyong-z]

Thank you all. I'm still in the very beginning stage, and i plan to investigate some papers and how other system implement it in the following days. @alamb thanks for sharing the google doc, i will put my system design plan on it in the future.

[xiaoyong-z]

I update some plans to implement the stream aggregation on https://docs.google.com/document/d/16rm5VR1nGkY6DedMCh1NUmThwf3RduAweaBH9b1h6AY/edit?usp=sharing

, PTAL. Detail design for fully stream aggregation will be given in the next following days.

[alamb]

PTAL. Detail design for fully stream aggregation will be given in the next following days.

Thank you @xiaoyong-z -- I read your addition and left some comments. Overall I think it is a great idea.

Here is one possibly approach to implementation (perhaps what you had in mind):

1. Implement StreamAggregate that handles pre-sorted data (where the data is already sorted according to the grouping keys/ partition keys).
2. Remove AggregateStream https://github.com/apache/arrow-datafusion/blob/master/datafusion/core/src/physical_plan/aggregates/no_grouping.rs and replace its use in the optimizer with the new StreamAggregate operator
3. Update the optimizer to recognize when the input to a GroupByHash is sorted appropriately and switch to using the AggregateStream operator.

I think that would get us pretty far.

I am not sure about the idea of "sort the data first and then run the stream aggregator" -- as I mentioned in the document I think it is unlikely that approach will be better in terms of overall memory usage or performance.

When we want to support spilling group by (external group by) that is when sort might be beneficial.

[ozankabak]

@mustafasrepo, can you take a detailed look at @xiaoyong-z's design? Thanks.

[mustafasrepo]

Thanks @xiaoyong-z, For the design. I asked some questions to understand the design better, and left some comments. Overall I think, your road map is well thought and planned.