perf: skip eval of standard window functions for unchanged partitions

[neilconway]

Which issue does this PR close?

• Related to BoundedWindowAggExec in Linear mode is slow for many-partitions #23982

Rationale for this change

In Linear mode, BoundedWindowAggStream does an evaluation sweep that visits every live partition for every window expression on every input batch. For standard (non-aggregate) window functions (e.g., row_number, rank, lag/lead, and nth_value), a partition that received no rows since the previous evaluation pass can be skipped: evaluating a function only depends on the contents of the partition itself.

New rows are detected with a received-row counter on PartitionBatchState that each expression's WindowAggState snapshots at the end of every evaluation pass.

This test is deliberately different from the "every buffered row has a result" test used to skip aggregate window expressions (#24127). Aggregate evaluation also consults the ORDER BY values of the most recent input row across all partitions, which can produce new results for a partition whose own input is unchanged.

Benchmarks:

• linear dense count 100: 44.4 ms -> 44.0 ms (-0.8%)
• linear dense count 10000: 161.9 ms -> 160.7 ms (-0.9%)
• linear sparse count 32768: 89.3 ms -> 88.4 ms (-0.9%)
• linear dense count rows-frame: 136.4 ms -> 136.1 ms (~noise)
• linear dense count+sum 10000: 253.1 ms -> 235.3 ms (-7.0%)
• linear dense row_number 10000: 114.7 ms -> 110.9 ms (-3.2%)
• linear sparse row_number 32768: 73.5 ms -> 42.2 ms (-42.5%)
• linear sparse lead 32768: 108.5 ms -> 46.6 ms (-57.0%)
• linear dense rank 10000: 125.8 ms -> 121.5 ms (-3.4%)
• sorted count 10000: 34.5 ms -> 33.4 ms (-3.6%)

What changes are included in this PR?

• Add counter to track # of received rows per partition
• Use received-row-count to avoid redundant evaluation of standard window functions
• Add tests

Are these changes tested?

Yes; new tests added.

Are there any user-facing changes?

No.

[github-actions]

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion-expr v55.0.0 (current)
       Built [  29.348s] (current)
     Parsing datafusion-expr v55.0.0 (current)
      Parsed [   0.077s] (current)
    Building datafusion-expr v55.0.0 (baseline)
       Built [  29.526s] (baseline)
     Parsing datafusion-expr v55.0.0 (baseline)
      Parsed [   0.076s] (baseline)
    Checking datafusion-expr v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   1.395s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure constructible_struct_adds_field: struct exhaustively constructible through public API adds field ---

Description:
A pub struct that could be exhaustively constructed with a literal using only public API has a new pub field, breaking existing exhaustive literals.
        ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/constructible_struct_adds_field.ron

Failed in:
  field PartitionBatchState.n_rows_received in /home/runner/work/datafusion/datafusion/datafusion/expr/src/window_state.rs:304
  field WindowAggState.n_rows_received in /home/runner/work/datafusion/datafusion/datafusion/expr/src/window_state.rs:55

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  61.525s] datafusion-expr
    Building datafusion-physical-expr v55.0.0 (current)
       Built [  29.502s] (current)
     Parsing datafusion-physical-expr v55.0.0 (current)
      Parsed [   0.048s] (current)
    Building datafusion-physical-expr v55.0.0 (baseline)
       Built [  29.448s] (baseline)
     Parsing datafusion-physical-expr v55.0.0 (baseline)
      Parsed [   0.049s] (baseline)
    Checking datafusion-physical-expr v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.341s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  60.261s] datafusion-physical-expr
    Building datafusion-physical-plan v55.0.0 (current)
       Built [  38.424s] (current)
     Parsing datafusion-physical-plan v55.0.0 (current)
      Parsed [   0.148s] (current)
    Building datafusion-physical-plan v55.0.0 (baseline)
       Built [  38.442s] (baseline)
     Parsing datafusion-physical-plan v55.0.0 (baseline)
      Parsed [   0.150s] (baseline)
    Checking datafusion-physical-plan v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.636s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  79.004s] datafusion-physical-plan

[codecov-commenter]

Codecov Report

❌ Patch coverage is 82.78689% with 21 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.31%. Comparing base (dbdc627) to head (ba5d95b).
⚠️ Report is 6 commits behind head on main.

Files with missing lines	Patch %	Lines
...ysical-plan/src/windows/bounded_window_agg_exec.rs	81.94%	3 Missing and 10 partials ⚠️
datafusion/expr/src/window_state.rs	82.97%	2 Missing and 6 partials ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##             main   #24532    +/-   ##
========================================
  Coverage   81.31%   81.31%            
========================================
  Files        1117     1117            
  Lines      395911   396290   +379     
  Branches   395911   396290   +379     
========================================
+ Hits       321918   322249   +331     
- Misses      55177    55197    +20     
- Partials    18816    18844    +28     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Dandandan]

Not this PR per se, but something I notice right now.
How often is this extend called? It would probably be better to buffer to a Vec (and delay concatenation) or use the batch coalescer (https://docs.rs/arrow/latest/arrow/compute/struct.BatchCoalescer.html) here otherwise it would be O(n^2) and do a lot of copying.