[nat-openai]

Which issue does this PR close?

Closes #7466.

Rationale for this change

ROW_NUMBER currently uses the generic bounded window executor, which performs per-partition state management and batch slicing even though this function only needs a counter and partition boundary detection. That overhead is particularly expensive for many small window partitions.

What changes are included in this PR?

• Add RowNumberExec to generate row numbers directly within each input batch, carrying the final partition key and counter across batches.
• Select it when every window expression is built-in ROW_NUMBER with matching partition and ordering keys.
• Reuse input column buffers and share the generated array between repeated ROW_NUMBER expressions.
• Account for retained partition-key buffers through DataFusion’s memory pool, including cleanup on completion, errors, and cancellation.
• Preserve existing ordering, distribution, statistics, and result-cast behaviour, and document the new execution path.

How are these changes tested?

• Regression tests compare results against the generic executor across partition boundaries, sliced and empty batches, global windows, repeated expressions, and nullable, dictionary, and nested keys.
• Memory tests cover retained-buffer accounting, reservation failures, cancellation, and avoiding retention of unrelated dictionary or view buffers.
• Spark SQL regressions cover global and partitioned windows, multiple batch sizes, ANSI modes, repeated expressions, and mixed-function windows. A planner regression checks executor selection alongside result casts and timing metrics.
• Benchmarks compare both executors on identical inputs and verify every row number before timing. The benchmark shows improvements across all tested partition sizes, with the largest gains for small partitions.

The 12 operator tests pass in an isolated harness. The Spark test additions compile but have not been executed locally.

Benchmark results

Median runtime for 262,144 rows in 32 batches of 8,192 rows, using three string partition keys. Measurements pool two runs of 10 samples each on an Apple M3 Max. The repeat measures streaming first.

The unchanged benchmark source ran in an isolated harness using Comet’s release settings, DataFusion 55.2.0, and Arrow 59.3.2. These measurements cover window execution over already-sorted input.

Input / rows per partition	Base ms	Candidate ms	Speedup range
String keys / 1	270–284	3.76–13.67	20–75×
String keys / 4	71.47–72.01	3.01–3.07	23.3–23.9×
String keys / 64	10.61–10.81	2.55–2.57	4.1–4.2×
String keys / 16,384	7.26–7.57	2.54–2.56	2.8–3.0×
String keys / 262,144	8.46–22.53	2.52–2.54	3.3–8.9×
Bigint keys / 4	51.48–52.32	0.737–0.741	69.5–71.0×
Bigint keys / 16,384	4.83–4.85	0.318–0.320	15.1–15.3×
Global window	4.26–18.01	0.235–0.386	11–77×
18 carried columns / 4	137–288	3.04–12.56	11–95×
18 carried columns / 16,384	11.35–30.92	2.56–2.61	4.4–12.1×

[dwsmith1983]

Thanks for this. A counter plus boundary detection is all ROW_NUMBER needs, and skipping the per-partition state in BoundedWindowAggExec pays off. I ran the branch (2ab0d3e) in a Linux arm64 container on the default profile:

• The 13 row_number tests in datafusion-comet-operators and the planner test window_cast_projection_reports_window_time pass.
• The row_number_executor bench reproduces your numbers, and the gap is a bit wider here: 229 ms vs 1.97 ms at one row per partition, 65.6 ms vs 1.86 ms at 4, 10.9 ms vs 1.50 ms at 64, and 5.3 to 5.5 ms vs 1.5 ms at 16,384 and 262,144.

A few things:

The new Spark test fails as written. window: streaming ROW_NUMBER across input batches stops with PATH_ALREADY_EXISTS before any query runs. withTempDir has already created dir, so the write needs .mode("overwrite"), as the neighbouring tests use. With that one line, the whole of CometWindowExecSuite passes (92 tests).

The Rust tests don't pin the cross-batch key comparison. I changed let start = if range.start == 0 && continues { to if range.start == 0 { in row_number.rs, so every batch continues the previous partition whatever its key. All 13 tests still pass. In every multi-batch input in tests.rs, the first row of each batch belongs to the partition the previous batch ended on. Could you add a case where a partition ends exactly at a batch boundary, such as [1, 1] | [2, 2], with a dictionary variant too? Relatedly, dropping || ranges.len() != 1 from the key replacement condition is caught by only the struct half of nested_keys_and_nullable_children_match_generic_windows. A direct check that the stored key is the last row after a batch like [1, 2, 2, 3] would make that guard harder to break by accident.

Floating-point partition keys. The existing window: floating partition keys loop in CometWindowExecSuite covers RANK, PERCENT_RANK and NTILE but not ROW_NUMBER, which now has its own boundary logic. Adding "ROW_NUMBER()" -> false and "ROW_NUMBER()" -> true to that list covers -0.0 next to 0.0, NaN payloads and null on the new path. I tried it locally and all four cases match Spark, so it's a cheap guard against losing the key normalization later.

Which issue this belongs to. #7092 is about BoundedWindowAggExec not reserving the rows its frames keep, and dhruv-dagar has taken it. This PR sends ROW_NUMBER-only windows to a different executor rather than adding that reservation, so I don't think it addresses #7092, even partially. The cost it removes is the one in apache/datafusion#26541, assigned to comphead. Could you open a Comet issue for the performance change and link that one instead? It would also help to note in RowNumberExec's doc comment that the operator works around apache/datafusion#26541, so it's clear when it can be removed.

[comphead]

Thanks for working on this, and for the benchmark that checks every row number before timing. I left out the points from the earlier review (the PATH_ALREADY_EXISTS fix, the batch boundary tests, float keys and the issue link) and have a few others below.

docs/source/contributor-guide/debugging.md conflicts with #7256, which edits the same BoundedWindowAggExec example. The Rust in window.rs merges cleanly, so whichever lands second only needs to update that doc block.

[nat-openai]

Thanks for the comments @dwsmith1983 and @comphead. I'm going to spend some moments internalising them and will loop back to address them properly. I'll also open a new issue that this resolves specifically rather than linking the other issue this sort of jumps over.

[nat-openai]

@dwsmith1983/@comphead I believe I've now addressed all the of previous comments.

[dwsmith1983]

Thanks, this covers everything. I ran 91e3357 in a Linux container on the default profile, and the new boundary tests and ROW_NUMBER float-key cases catch the continues change that slipped through before.

The SQL file fails on its last query. The suite turns off constant folding, so 1 PRECEDING never becomes a literal and Spark's window throws Unhandled bound in windows expressions: -1. Switching to ROWS BETWEEN CURRENT ROW AND 1 FOLLOWING fixes it.