enhancement(optimizer): merge common aggregation functions

[fuyufjh]

Discussed with @st1page

For example,

select avg(x) a, sum(x) s, count(x) c from t;

will be executed as

select expr1 / expr2, s, c from 
  (select sum(x) as expr1, count(x) as expr2, sum(x) s , count(x) c from t);

Here, the two sum(x) and two count(x) can be simply reused.

Another example is NexMark Q15

risingwave/src/frontend/planner_test/tests/testdata/output/nexmark_source.yaml

Line 1126 in 0259125

StreamHashAgg [append_only] { group_key: [$expr1, bidder, auction, flag], aggs: [count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32))] } { intermediate state table: 1, state tables: [], distinct tables: [] }

StreamHashAgg [append_only] {
  group_key: [$expr1, bidder, auction, flag],
  aggs: [
    count, 
    count filter((price < 10000:Int32)), 
    count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), 
    count filter((price >= 1000000:Int32)),
    count filter((price < 10000:Int32)), 
    count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), 
    count filter((price >= 1000000:Int32)),
    count filter((price < 10000:Int32)), 
    count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), 
    count filter((price >= 1000000:Int32))] 

Obviously the 9 count agg functions can be merged to 3, so that we reduce the aggregation state size to 1/3.

[st1page]

We do not need to implement it in Agg executor but implement it as an optimizer rule we can write a project ahead 🤔