feat: support WindowGroupLimitExec

dev/diffs/3.5.9.diff

@@ -1446,6 +1446,30 @@ index 005e764cc30..92ec088efab 100644
    }
 
    private def checkSorts(query: String, enabledCount: Int, disabledCount: Int): Unit = {
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala
+index 46ed8fdfd21..4585dbab5b8 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala
+@@ -18,6 +18,7 @@
+ package org.apache.spark.sql.execution
+
+ import org.apache.spark.sql.{DataFrame, QueryTest}
++import org.apache.spark.sql.comet.CometWindowGroupLimitExec
+ import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper, DisableAdaptiveExecutionSuite, EnableAdaptiveExecutionSuite}
+ import org.apache.spark.sql.execution.window.WindowGroupLimitExec
+ import org.apache.spark.sql.functions.lit
+@@ -30,7 +31,10 @@ abstract class RemoveRedundantWindowGroupLimitsSuiteBase
+
+   private def checkNumWindowGroupLimits(df: DataFrame, count: Int): Unit = {
+     val plan = df.queryExecution.executedPlan
+-    assert(collectWithSubqueries(plan) { case exec: WindowGroupLimitExec => exec }.length == count)
++    assert(collectWithSubqueries(plan) {
++      case exec: WindowGroupLimitExec => exec
++      case exec: CometWindowGroupLimitExec => exec
++    }.length == count)
+   }
+
+   private def checkWindowGroupLimits(query: String, count: Int): Unit = {
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/ReplaceHashWithSortAggSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/ReplaceHashWithSortAggSuite.scala
 index 47679ed7865..9ffbaecb98e 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/ReplaceHashWithSortAggSuite.scala

dev/diffs/4.0.4.diff

@@ -1862,6 +1862,30 @@ index 005e764cc30..92ec088efab 100644
    }
 
    private def checkSorts(query: String, enabledCount: Int, disabledCount: Int): Unit = {
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala
+index 46ed8fdfd21..4585dbab5b8 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala
+@@ -18,6 +18,7 @@
+ package org.apache.spark.sql.execution
+
+ import org.apache.spark.sql.{DataFrame, QueryTest}
++import org.apache.spark.sql.comet.CometWindowGroupLimitExec
+ import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper, DisableAdaptiveExecutionSuite, EnableAdaptiveExecutionSuite}
+ import org.apache.spark.sql.execution.window.WindowGroupLimitExec
+ import org.apache.spark.sql.functions.lit
+@@ -30,7 +31,10 @@ abstract class RemoveRedundantWindowGroupLimitsSuiteBase
+
+   private def checkNumWindowGroupLimits(df: DataFrame, count: Int): Unit = {
+     val plan = df.queryExecution.executedPlan
+-    assert(collectWithSubqueries(plan) { case exec: WindowGroupLimitExec => exec }.length == count)
++    assert(collectWithSubqueries(plan) {
++      case exec: WindowGroupLimitExec => exec
++      case exec: CometWindowGroupLimitExec => exec
++    }.length == count)
+   }
+
+   private def checkWindowGroupLimits(query: String, count: Int): Unit = {
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/ReplaceHashWithSortAggSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/ReplaceHashWithSortAggSuite.scala
 index 47679ed7865..9ffbaecb98e 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/ReplaceHashWithSortAggSuite.scala

dev/diffs/4.1.3.diff

@@ -1985,6 +1985,30 @@ index 005e764cc30..92ec088efab 100644
    }
 
    private def checkSorts(query: String, enabledCount: Int, disabledCount: Int): Unit = {
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala
+index 46ed8fdfd21..4585dbab5b8 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala
+@@ -18,6 +18,7 @@
+ package org.apache.spark.sql.execution
+
+ import org.apache.spark.sql.{DataFrame, QueryTest}
++import org.apache.spark.sql.comet.CometWindowGroupLimitExec
+ import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper, DisableAdaptiveExecutionSuite, EnableAdaptiveExecutionSuite}
+ import org.apache.spark.sql.execution.window.WindowGroupLimitExec
+ import org.apache.spark.sql.functions.lit
+@@ -30,7 +31,10 @@ abstract class RemoveRedundantWindowGroupLimitsSuiteBase
+
+   private def checkNumWindowGroupLimits(df: DataFrame, count: Int): Unit = {
+     val plan = df.queryExecution.executedPlan
+-    assert(collectWithSubqueries(plan) { case exec: WindowGroupLimitExec => exec }.length == count)
++    assert(collectWithSubqueries(plan) {
++      case exec: WindowGroupLimitExec => exec
++      case exec: CometWindowGroupLimitExec => exec
++    }.length == count)
+   }
+
+   private def checkWindowGroupLimits(query: String, count: Int): Unit = {
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/ReplaceHashWithSortAggSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/ReplaceHashWithSortAggSuite.scala
 index 47679ed7865..9ffbaecb98e 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/ReplaceHashWithSortAggSuite.scala

docs/source/contributor-guide/native_shuffle.md

@@ -49,7 +49,6 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
    columnar output. Row-based Spark operators require JVM shuffle.
 
 3. **Supported partitioning type**: Native shuffle supports:
-
    - `HashPartitioning`
    - `RangePartitioning`
    - `SinglePartition`
@@ -129,7 +128,6 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
 1. **Plan construction**: `CometNativeShuffleWriter` builds a protobuf operator tree with a
    `ShuffleWriter` operator at the root and `childNativeOp` as its child. `childNativeOp` takes
    one of two shapes:
-
    - The child plan's `nativeOp` directly, when `CometShuffleExchangeExec`'s child is a
      `CometNativeExec` subtree. The upstream operators run inside the same `CometExecIterator`
      as the writer, with no JVM-to-native batch boundary between them.
@@ -142,21 +140,18 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
 2. **Native execution**: A single `CometExecIterator` per partition runs the unified plan.
 
 3. **Partitioning**: `ShuffleWriterExec` receives batches and routes to the appropriate partitioner:
-
    - `MultiPartitionShuffleRepartitioner`: For hash/range/round-robin partitioning
    - `SinglePartitionShufflePartitioner`: For single partition (simpler path)
 
 4. **Buffering and spilling**: The partitioner buffers rows per partition. When memory pressure
    exceeds the threshold, partitions spill to temporary files.
 
 5. **Encoding**: `ShuffleBlockWriter` encodes each partition's data as compressed Arrow IPC:
-
    - Writes compression type header
    - Writes field count header
    - Writes compressed IPC stream
 
 6. **Output files**: Two files are produced:
-
    - **Data file**: Concatenated partition data
    - **Index file**: Array of 8-byte little-endian offsets marking partition boundaries
 
@@ -168,7 +163,6 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
 1. `CometBlockStoreShuffleReader` fetches shuffle blocks via `ShuffleBlockFetcherIterator`.
 
 2. For each block, `NativeBatchDecoderIterator`:
-
    - Reads the 8-byte compressed length header
    - Reads the 8-byte field count header
    - Reads the compressed IPC data

docs/source/contributor-guide/roadmap.md

@@ -29,8 +29,8 @@ Native window execution runs by default (`spark.comet.exec.window.enabled`). The
 `first_value`, `last_value`), and the `count`, `min`, `max`, `sum`, and `avg` aggregates are accelerated.
 Remaining work is to close the gaps that still fall back to Spark: statistical aggregates (`stddev`, `variance`,
 `corr`, `covar`) and `collect_list` / `collect_set` as window functions ([#4766]), `GROUPS` frames ([#4836]), `RANGE` frames with explicit date or
-decimal offsets ([#4834]), `first_value` / `last_value` on `RANGE` frames with a literal offset ([#4835]),
-non-literal `lag` / `lead` default values ([#4268]), and `WindowGroupLimitExec` ([#4837]). See the
+decimal offsets ([#4834]), `first_value` / `last_value` on `RANGE` frames with a literal offset ([#4835]), and
+non-literal `lag` / `lead` default values ([#4268]). See the
 [window function compatibility guide](../user-guide/latest/compatibility/operators.md) for the complete list of
 supported functions, frames, and fallback cases.
 
@@ -39,7 +39,6 @@ supported functions, frames, and fallback cases.
 [#4834]: https://github.com/apache/datafusion-comet/issues/4834
 [#4835]: https://github.com/apache/datafusion-comet/issues/4835
 [#4836]: https://github.com/apache/datafusion-comet/issues/4836
-[#4837]: https://github.com/apache/datafusion-comet/issues/4837
 
 ## Native Lambda Evaluation
 

docs/source/user-guide/latest/compatibility/floating-point.md

@@ -27,3 +27,16 @@ So Comet adds additional normalization expression of NaN and zero for comparison
 to Spark in some cases, especially when the data contains both positive and negative zero. This is likely an edge
 case that is not of concern for many users. If it is a concern, setting `spark.comet.exec.strictFloatingPoint=true`
 will make relevant operations fall back to Spark.
+
+## Ordering: signed zero (`-0.0` vs `+0.0`)
+
+Spark's `ORDER BY`, `RANK`, `DENSE_RANK`, and window frame comparisons route through
+`SQLOrderingUtil.compareDoubles` / `compareFloats`, which explicitly define `-0.0 == 0.0`. Comet's
+native sort and `WindowGroupLimitExec` use the `arrow-row` row-format encoder for `ORDER BY` keys,
+which applies Rust's total-ordering transform to the raw IEEE-754 bits. Under that encoding `-0.0`
+sorts strictly less than `+0.0`, so a partition that mixes the two zeros can produce a rank
+distribution that differs from Spark. For example, `RANK() OVER (ORDER BY v ASC)` over
+`[-0.0, 0.0, 1.0]` filtered to `rk <= 1` returns two rows in Spark (both zeros tied at rank 1) but
+one row in Comet (`-0.0` at rank 1, `+0.0` at rank 2). If your workload materially mixes `-0.0`
+and `+0.0` in a ranked column, prefer Spark for that stage or normalize the column to `+0.0`
+upstream.

docs/source/user-guide/latest/compatibility/operators.md

@@ -71,8 +71,19 @@ incorrect result. When any single window expression in a `WindowExec` falls back
   window are not supported by Spark either.
 - Any `PARTITION BY` or `ORDER BY` expression that Comet cannot serialize.
 
-`WindowGroupLimitExec` (window-based limit pushdown) is not yet supported and falls back to Spark
-([#4837](https://github.com/apache/datafusion-comet/issues/4837)).
+`WindowGroupLimitExec` (window-based limit pushdown for `ROW_NUMBER`, `RANK`, and `DENSE_RANK`)
+runs natively; it is controlled by `spark.comet.exec.windowGroupLimit.enabled` (default: true).
+
+**Falls back to Spark:**
+
+- Any `PARTITION BY` or `ORDER BY` key whose type carries a non-default `StringType` collation
+  (e.g. `UTF8_LCASE`). The native operator detects partitions and order-key peer groups by
+  comparing Arrow row-encoded keys for byte equality, which splits peers that Spark ties.
+
+**Known incompatibilities:**
+
+- Signed-zero ordering (`-0.0` vs `+0.0`) diverges from Spark's `RankLimitIterator`; see
+  [floating-point ordering](./floating-point.md#ordering-signed-zero-00-vs-00).
 
 ## Round-Robin Partitioning
 

docs/source/user-guide/latest/operators.md

@@ -103,7 +103,7 @@ omitted from the tables below and may be reconsidered based on demand:
 | Operator               | Status | Notes                                                                                                                                                                                            |
 | ---------------------- | ------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
 | `WindowExec`           | ⚠️     | Runs natively and is enabled by default. A broad set of window functions is accelerated; unsupported shapes fall back to Spark. See [window function compatibility](compatibility/operators.md). |
-| `WindowGroupLimitExec` | 🔜     | Window-based limit pushdown falls back today ([#4837](https://github.com/apache/datafusion-comet/issues/4837)).                                                                                  |
+| `WindowGroupLimitExec` | ✅     | Streaming per-partition top-K pushdown for `ROW_NUMBER`, `RANK`, and `DENSE_RANK`.                                                                                                               |
 
 ## Generators and set operations
 

native/core/src/execution/jni_api.rs

@@ -276,6 +276,7 @@ fn op_name(op: &OpStruct) -> &'static str {
         OpStruct::BroadcastNestedLoopJoin(_) => "BroadcastNestedLoopJoin",
         OpStruct::Sample(_) => "Sample",
         OpStruct::ContribScan(_) => "ContribScan",
+        OpStruct::WindowGroupLimit(_) => "WindowGroupLimit",
     }
 }
 

native/core/src/execution/operators/mod.rs

@@ -35,6 +35,8 @@ mod csv_scan;
 pub mod projection;
 mod sample;
 pub use sample::SampleExec;
+mod rank_limit;
+pub use rank_limit::{PartitionedRankLimitExec, WindowFnKind};
 mod scan;
 mod shuffle_scan;
 pub use csv_scan::init_csv_datasource_exec;