[CH] Support `SortAggregateExec`

[lgbo-ustc]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

Offload SortAggregateExec into native engine.

[lgbo-ustc]

What we need

ClickHouse

CH seems has supported sorted-based aggregating. see src/Processors/QueryPlan/AggregatingStep.cpp. If sort_description_for_merging is not empty, it will use AggregatingInOrderTransform.

substrait

AggregateFunction has a repeated SortField field sorts, I think we could put the sorting information into it.

[lgbo-ustc]

How to test SortAggregateExec:

• spark.sql.execution.useObjectHashAggregateExec=false
• spark.sql.autoBroadcastJoinThreshold=-1
• query: select k1, collect_list(k1) as k3 from (select floor(id/4) as k1 from (select id from range(10)) as t1) as t2 group by k1

[lgbo-ustc]

There is special case that two sort agg exec connected without exchange.

               +- ColumnarExchange (257)
                  +- ^ CHHashAggregateExecTransformer (255)
                     +- ColumnarExchange (254)
                        +- ^ CHHashAggregateExecTransformer (252)
                           +- ^ FilterExecTransformer (251)
                              +- ^ ExpandExecTransformer (250)
                                 +- ^ ProjectExecTransformer (249)
                                    +- ^ CHShuffledHashJoinExecTransformer Inner (248)
                                       :- RowToCHNativeColumnar (238)
                                       :  +- SortAggregate (237)
                                       :     +- SortAggregate (236)
                                       :        +- CHNativeColumnarToRow (235)
                                       :           +- ^ SortExecTransformer (233)
                                       :              +- ^ ProjectExecTransformer (232)
                                       :                 +- ^ CHShuffledHashJoinExecTransformer LeftOuter (231)
                                       :                    :- ReusedExchange (229)
                                       :                    +- ReusedExchange (230)

[PHILO-HE]

For consecutive sort agg, I guess the second one is for partial merge.