[flink] supports agg pushdown with partitions groupby in flink

[steFaiz]

Purpose

Mirrors spark implementation.
This PR originates from #7779. In our internal case, we find it hard for us to quickly get the real records num for data evolution tables, especially for each partition.
Currently both snapshots system table and partitions system table only shows unmerged records num. We could get the accurate values by count(*) agg pushdown (probably with group by clause) through flink OLAP queries.

Tests

Unit Tests and ITCases

[JingsongLi]

extractAggregates uses aggregateExpression.getArgs().get(0).getFieldIndex() — This assumes the expression argument is a FieldReferenceExpression. If Flink ever returns a computed expression as arg (e.g., MIN(f0 + 1)), this would fail at runtime.

Can you add test to cover this?

[steFaiz]

Thanks for your remind! During implementation, I referred to Flink's original agg pushdown interface, which says it only supports pushing down simple fields:

For example, if the sql is:

SELECT MIN(f0 * 2) FROM paimon

The plan will be:

LocalHashAggregate(Partial_MIN($f0))
  +- Calc(select=[(f0 * 2) AS $f0])
     +- TableSourceScan(project=[f0])

Since Flink does not consider these situations, the args of AggregateExpression is hard-coded as FieldReferenceExpression:

I've added a test for these cases!

[JingsongLi]

+1