Merge branch 'main' into add_group_benchmarks_count_distinct

Author: coderfender

benchmarks/queries/h2o/window.sql

@@ -109,4 +109,11 @@ SELECT
     id3,
     v2,
     sum(v2) OVER (PARTITION BY id2 ORDER BY v2 RANGE BETWEEN 3 PRECEDING AND CURRENT ROW) AS my_range_between_by_id2
-FROM large;
+FROM large;
+
+-- Window Top-N (ROW_NUMBER top-2 per partition)
+SELECT id2, largest2_v2 FROM (
+    SELECT id2, v2 AS largest2_v2,
+           ROW_NUMBER() OVER (PARTITION BY id2 ORDER BY v2 DESC) AS order_v2
+    FROM large WHERE v2 IS NOT NULL
+) sub_query WHERE order_v2 <= 2;

datafusion/common/src/config.rs

@@ -1087,6 +1087,15 @@ config_namespace! {
         /// past window functions, if possible
         pub enable_window_limits: bool, default = true
 
+        /// When set to true, the optimizer will replace
+        /// Filter(rn<=K) → Window(ROW_NUMBER) → Sort patterns with a
+        /// PartitionedTopKExec that maintains per-partition heaps, avoiding
+        /// a full sort of the input.
+        /// When the window partition key has low cardinality, enabling this optimization
+        /// can improve performance. However, for high cardinality keys, it may
+        /// cause regressions in both memory usage and runtime.
+        pub enable_window_topn: 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/tests/physical_optimizer/mod.rs

@@ -37,5 +37,6 @@ mod sanity_checker;
 #[expect(clippy::needless_pass_by_value)]
 mod test_utils;
 mod window_optimize;
+mod window_topn;
 
 mod pushdown_utils;