Wrong results: window aggregates over RANGE ... N PRECEDING depend on batch boundaries when the ORDER BY value is NULL #25859

Describe the bug

For a RANGE frame with an offset, the frame of a row whose ORDER BY value is NULL is its group of NULL peers. When the frame ends in N PRECEDING and that group spans more than one input batch, BoundedWindowAggExec computes the NULL rows before the rest of the group has arrived, so they get results over part of the group. At default settings this happens once a NULL group crosses a batch boundary (8192 rows). NULLS FIRST, NULLS LAST and DESC orderings are all affected, and so is a group of NaN keys, since NaN - 1 is NaN. ROWS and GROUPS frames are not.

To Reproduce

Default settings, datafusion-cli on main (bdad988):

CREATE TABLE t AS SELECT value AS id, CASE WHEN value <= 10000 THEN value END AS x FROM generate_series(1, 20000);

SELECT n, count(*) AS rows FROM (
  SELECT x, count(*) OVER (ORDER BY x RANGE BETWEEN 5 PRECEDING AND 1 PRECEDING) AS n FROM t
) WHERE x IS NULL GROUP BY n ORDER BY n;

+-------+------+
| n     | rows |
+-------+------+
| 6384  | 6384 |
| 10000 | 3616 |
+-------+------+

Every NULL row should see all 10,000 NULL peers. DuckDB 1.5.5 and Postgres 17 return (10000, 10000).

A smaller version with a tiny batch size:

SET datafusion.execution.batch_size = 2;
CREATE TABLE s (id INT, x INT, v INT) AS VALUES (1,1,10),(2,2,20),(3,NULL,30),(4,NULL,40),(5,NULL,50);
SELECT id,
  sum(v) OVER (ORDER BY x RANGE BETWEEN 5 PRECEDING AND 1 PRECEDING) AS s,
  sum(v) OVER (ORDER BY x RANGE BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS u
FROM s ORDER BY id;

This returns s = NULL, 10, 70, 70, 120 and u = NULL, 10, 100, 100, 150. With the default batch size DataFusion returns 120 and 150 for all three NULL rows, the same as DuckDB and Postgres.

Expected behavior

The result doesn't depend on the batch size: NULL rows get the aggregate over all their NULL peers (120 and 150 in the small example).

Additional context

In the aggregate path (window_expr.rs), when a frame reaches the end of the buffered input the loop asks is_end_bound_safe whether more rows could still change it. For RANGE with a non-zero PRECEDING end, is_end_bound_safe_for_range always returns true, but for a NULL (or NaN) row the frame ends at the last peer, which can still be in a later batch. Frames that WindowFrame::new_bounds builds from typed bounds, e.g. a FOLLOWING frame reversed to reuse another window's sort order, are also marked causal, so they skip that check entirely. On that route nth_value is wrong as well:

SELECT id,
  count(*) OVER (ORDER BY x DESC NULLS LAST, id) AS c,
  nth_value(1, 3) OVER (ORDER BY x ASC NULLS FIRST RANGE BETWEEN 1 FOLLOWING AND 5 FOLLOWING) AS n
FROM s ORDER BY id;
-- n is NULL, NULL, 1 for the three NULL rows; it should be 1 for all of them

This seems to go back to #8842, which added early results for causal frames.

I have a fix with sqllogictest cases and I'm happy to open a PR.