feat(optimizer): logical rewrite Filter(row_number() = 1) over PARTITION BY → Aggregate(FIRST_VALUE(... ORDER BY))

datafusion/common/src/config.rs

@@ -1644,6 +1644,16 @@ config_namespace! {
         /// cause regressions in both memory usage and runtime.
         pub enable_window_topn: bool, default = false
 
+        /// When set to true, the optimizer will rewrite a "top-1 per group"
+        /// pattern of the form `Filter(row_number() = 1)` over a `PARTITION BY` window into an
+        /// `Aggregate(first_value(... ORDER BY ...) GROUP BY partition)`.
+        /// This avoids buffering / fully sorting the input, which is especially beneficial on wide payloads.
+        /// Disabled by default because on nested/wide value types the aggregate only becomes memory
+        /// efficient once the columnar nested-type `GroupsAccumulator` fast
+        /// path is available; enabling it without that support can route wide
+        /// payloads onto the slow per-group accumulator path.
+        pub enable_row_number_to_aggregate: 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

@@ -63,6 +63,7 @@ Rule order matters. The default pipeline may change between releases.
 | 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.                                                          |
+| 27    | `replace_filter_top1`                     | Rewrites `Filter(row_number() = 1)` over a partitioned window into a `first_value` aggregate.                               |
 
 ### Physical Optimizer Rules
 

datafusion/optimizer/src/lib.rs

@@ -54,6 +54,7 @@ pub mod propagate_empty_relation;
 pub mod push_down_filter;
 pub mod push_down_limit;
 pub mod replace_distinct_aggregate;
+pub mod replace_filter_top1;
 pub mod rewrite_set_comparison;
 pub mod scalar_subquery_to_join;
 pub mod simplify_expressions;

datafusion/optimizer/src/optimizer.rs

@@ -49,6 +49,7 @@ use crate::propagate_empty_relation::PropagateEmptyRelation;
 use crate::push_down_filter::PushDownFilter;
 use crate::push_down_limit::PushDownLimit;
 use crate::replace_distinct_aggregate::ReplaceDistinctWithAggregate;
+use crate::replace_filter_top1::ReplaceFilterTop1;
 use crate::rewrite_set_comparison::RewriteSetComparison;
 use crate::scalar_subquery_to_join::ScalarSubqueryToJoin;
 use crate::simplify_expressions::SimplifyExpressions;
@@ -302,6 +303,7 @@ impl Optimizer {
             Arc::new(ExtractLeafExpressions::new()),
             Arc::new(PushDownLeafProjections::new()),
             Arc::new(OptimizeProjections::new()),
+            Arc::new(ReplaceFilterTop1::new()),
         ];
 
         Self::with_rules(rules)