[FLINK-31260] PushLocalHashAggIntoScanRule should also work with union RelNode

• Type: Improvement
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Component/s: Table SQL / Planner
• Assignee: Unassigned
• Reporter: WenJun Min
• Created: 28/Feb/23 12:53
• Updated: 10/Mar/23 12:02

Description

As discussed in comments Currently, PushLocalHashAggIntoScanRule match for the Exchange -> LocalHashAggregate -> Scan. As a result, the following pattern can not be optimized

      +- Union(all=[true], union=[type, sum$0])
         :- Union(all=[true], union=[type, sum$0])
         :  :- LocalHashAggregate(groupBy=[type], select=[type, Partial_SUM(price) AS sum$0])
         :  :  +- TableSourceScan(table=[[default_catalog, default_database, table1, project=[type, price], metadata=[]]], fields=[type, price])
         :  +- LocalHashAggregate(groupBy=[type], select=[type, Partial_SUM(price) AS sum$0])
         :     +- TableSourceScan(table=[[default_catalog, default_database, table2, project=[type, price], metadata=[]]], fields=[type, price])
         +- LocalHashAggregate(groupBy=[type], select=[type, Partial_SUM(price) AS sum$0])
            +- TableSourceScan(table=[[default_catalog, default_database, table3, project=[type, price], metadata=[]]], fields=[type, price])

We should extend the rule to support this pattern.