[comphead]

Which issue does this PR close?

• Closes BoundedWindowAggExec can emit batches larger than batch_size #26959.

Rationale for this change

BoundedWindowAggExec emits every row that becomes final in one batch, so a long RANGE peer group, ROWS ... N FOLLOWING or lead(x, N) can produce batches far larger than datafusion.execution.batch_size. The issue has a reproducer.

What changes are included in this PR?

execute wraps the stream in BatchSplitStream and ObservedStream, as UnnestExec does since #24384. Larger batches go out as zero-copy slices of at most batch_size rows, and output_batches counts the slices.

What is the testing strategy for this PR?

A new window.slt case runs EXPLAIN ANALYZE with batch_size = 2 over two 5-row peer groups, one finishing mid-stream and one at the end of the input. BoundedWindowAggExec reports output_batches=6, against 2 on main. bounded_window_linear_quiet_partition_resume_standard now uses the default batch size, so it still checks when rows become final. The bounded_window benchmark is within 3% of main on every case, in both directions (best of six alternating runs).

Are there any user-facing changes?

Only batch boundaries change, and BoundedWindowAggExec gains a batches_split metric. Known limitations:

• Small batches are not merged. With repeated ORDER BY keys, an emission slightly over batch_size becomes a full batch plus a short one. sum(v) OVER (ORDER BY k) over 1M rows with 3-row peer groups goes from 124 to 205 output batches.
• The slices share their parent's buffers. Operators that reserve get_array_memory_size() per input batch, such as NestedLoopJoinExec and CrossJoinExec, charge each slice for the whole parent, as they already do for aggregate output. A nested loop join building on a 65,536-row peer group at batch_size = 1024 reserves 96 MB instead of 1.5 MB. HashJoinExec counts shared buffers once since fix: count shared buffers once in hash join build-side memory accounting #22862.

[2010YOUY01]

LGTM, thank you!