feat(optimizer): coalesce peer first_value / last_value into a single struct aggregate

datafusion/common/src/config.rs

@@ -1390,6 +1390,14 @@ config_namespace! {
         /// cause regressions in both memory usage and runtime.
         pub enable_window_topn: bool, default = false
 
+        /// When set to true, coalesce peer `first_value` / `last_value`
+        /// expressions that share the same `ORDER BY` key — e.g.
+        /// `first_value(a ORDER BY o), first_value(b ORDER BY o), ... GROUP BY p`
+        /// — into a single `first_value(named_struct(a, b, ...) ORDER BY o)`,
+        /// cutting N argmax scans to one pass over the input.
+        /// Particularly beneficial for wide payloads.
+        pub enable_coalesce_first_last: bool, default = false
+
         /// When set to true, the optimizer will push TopK (Sort with fetch)
         /// below hash repartition when the partition key is a prefix of the
         /// sort key, reducing data volume before the shuffle.

datafusion/core/src/optimizer_rule_reference.md

@@ -58,10 +58,11 @@ Rule order matters. The default pipeline may change between releases.
 | 19    | `push_down_filter`                        | Moves filters as early as possible through filter-commutative operators.                                                    |
 | 20    | `single_distinct_aggregation_to_group_by` | Rewrites single-column `DISTINCT` aggregations into two-stage `GROUP BY` plans.                                             |
 | 21    | `eliminate_group_by_constant`             | Removes constant or functionally redundant expressions from `GROUP BY`.                                                     |
-| 22    | `common_sub_expression_eliminate`         | Computes repeated subexpressions once and reuses the result.                                                                |
-| 23    | `extract_leaf_expressions`                | Pulls cheap leaf expressions closer to data sources so later pruning and filter rules can act earlier.                      |
-| 24    | `push_down_leaf_projections`              | Pushes the helper projections created by leaf extraction toward leaf inputs.                                                |
-| 25    | `optimize_projections`                    | Prunes unused columns and removes unnecessary logical projections.                                                          |
+| 22    | `coalesce_first_last`                     | Coalesces peer `first_value` / `last_value` aggregates that share an `ORDER BY` into a single struct aggregate.             |
+| 23    | `common_sub_expression_eliminate`         | Computes repeated subexpressions once and reuses the result.                                                                |
+| 24    | `extract_leaf_expressions`                | Pulls cheap leaf expressions closer to data sources so later pruning and filter rules can act earlier.                      |
+| 25    | `push_down_leaf_projections`              | Pushes the helper projections created by leaf extraction toward leaf inputs.                                                |
+| 26    | `optimize_projections`                    | Prunes unused columns and removes unnecessary logical projections.                                                          |
 
 ### Physical Optimizer Rules
 

datafusion/optimizer/Cargo.toml

@@ -69,6 +69,7 @@ regex-syntax = "0.8.9"
 async-trait = { workspace = true }
 criterion = { workspace = true }
 ctor = { workspace = true }
+datafusion-functions = { workspace = true }
 datafusion-functions-aggregate = { workspace = true }
 datafusion-functions-window = { workspace = true }
 datafusion-functions-window-common = { workspace = true }