Allow combining adjacent partial and final AggregateExec

[mingmwang]

Is your feature request related to a problem or challenge?

TPCH-Q17 physical plan:

ProjectionExec: expr=[CAST(SUM(lineitem.l_extendedprice)@0 AS Float64) / 7 as avg_yearly]
  AggregateExec: mode=Final, gby=[], aggr=[SUM(lineitem.l_extendedprice)]
    AggregateExec: mode=Partial, gby=[], aggr=[SUM(lineitem.l_extendedprice)]
      ProjectionExec: expr=[l_extendedprice@1 as l_extendedprice]
        CoalesceBatchesExec: target_batch_size=8192
          HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(Column { name: "p_partkey", index: 2 }, Column { name: "l_partkey", index: 0 })], filter=BinaryExpr { left: CastExpr { expr: Column { name: "l_quantity", index: 0 }, cast_type: Decimal128(30, 15), cast_options: CastOptions { safe: false } }, op: Lt, right: CastExpr { expr: Column { name: "__value", index: 1 }, cast_type: Decimal128(30, 15), cast_options: CastOptions { safe: false } } }
            ProjectionExec: expr=[l_quantity@1 as l_quantity, l_extendedprice@2 as l_extendedprice, p_partkey@3 as p_partkey]
              CoalesceBatchesExec: target_batch_size=8192
                HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(Column { name: "l_partkey", index: 0 }, Column { name: "p_partkey", index: 0 })]
                  ParquetExec: limit=None, partitions={1 group: [[/gitrepo/apache/arrow-datafusion/benchmarks/parquet_data/lineitem/part-0.parquet]]}, projection=[l_partkey, l_quantity, l_extendedprice]
                  ProjectionExec: expr=[p_partkey@0 as p_partkey]
                    CoalesceBatchesExec: target_batch_size=8192
                      FilterExec: p_brand@1 = Brand#23 AND p_container@2 = MED BOX
                        ParquetExec: limit=None, partitions={1 group: [[/gitrepo/apache/arrow-datafusion/benchmarks/parquet_data/part/part-0.parquet]]}, predicate=p_brand@3 = Brand#23 AND p_container@6 = MED BOX, pruning_predicate=p_brand_min@0 <= Brand#23 AND Brand#23 <= p_brand_max@1 AND p_container_min@2 <= MED BOX AND MED BOX <= p_container_max@3, projection=[p_partkey, p_brand, p_container]
            ProjectionExec: expr=[l_partkey@0 as l_partkey, 0.2 * CAST(AVG(lineitem.l_quantity)@1 AS Float64) as __value]
              AggregateExec: mode=Final, gby=[l_partkey@0 as l_partkey], aggr=[AVG(lineitem.l_quantity)]
                AggregateExec: mode=Partial, gby=[l_partkey@0 as l_partkey], aggr=[AVG(lineitem.l_quantity)]
                  ParquetExec: limit=None, partitions={1 group: [[/gitrepo/apache/arrow-datafusion/benchmarks/parquet_data/lineitem/part-0.parquet]]}, projection=[l_partkey, l_quantity]

We should allow combine/merge the adjacent partial and final AggregateExec in this case if there is no RepartitionExec among them.

Describe the solution you'd like

No response

Describe alternatives you've considered

No response

Additional context

No response

[mingmwang]

I'm going to add a new physical optimizer rule for this purpose. This rule should be applied after the EnforceSorting and EnforceDistribution. We might need to add new Aggregation model.