perf: track BoundedWindowAggExec Linear-mode watermark once per stream

[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, rows arrive ordered on a prefix of the window's ORDER BY expressions, so each new row bounds every row that will arrive in the future, even for other partitions. We exploit this by using the last row to arrive in a batch to close window frames for all live partitions, not just the partition to which that row belongs.

The most-recent-row-in-the-batch is conceptually per-batch state, but it was previously implemented as per-partition state:

• update_partition_batch copied the last row of each incoming batch into every live partition's PartitionBatchState.
• Each copy is ~(40 + 16 * n_cols) bytes (Option<RecordBatch> plus a Vec of ArrayRefs); for example, withn 100k live partitions over 10 columns, that is ~20MB of duplicate state.
• Every partition visit re-evaluated the row's ORDER BY expressions.

Instead, just store the row once. Rather than copying the row into every partition, we pass the row to window expression evaluation. This removes most_recent_row and its setter from PartitionBatchState, which is a breaking API change for datafusion-expr. We can also arrange to evaluate the ORDER BY once per batch instead of once per partition.

This is a modest performance improvement and memory savings, but also a conceptual cleanup/refactor.

Benchmarks: (using #24032)

• linear 100 partitions: 44.4 ms -> 44.5 ms (within noise)
• linear 10000 partitions: 203.3 ms -> 199.7 ms (-1.7%)
• linear sparse 32768 partitions: 236.9 ms -> 224.5 ms (-5.2%)
• linear rows 10000 partitions: 174.2 ms -> 169.4 ms (-2.4%)
• linear multi 10000 partitions: 304.6 ms -> 295.7 ms (-3.0%)
• sorted 10000 partitions: 34.2 ms -> 34.4 ms (within noise)

What changes are included in this PR?

• Add WindowEvalContext parameter to aggregate_evaluate_stateful
• Remove PartitionBatchState::most_recent_row and its setter; instead pass row via WindowEvalContext

Are these changes tested?

Yes, covered by existing tests.

Are there any user-facing changes?

Yes, API change to datafusion-expr.

[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 v54.1.0 (current)
       Built [  26.350s] (current)
     Parsing datafusion-expr v54.1.0 (current)
      Parsed [   0.074s] (current)
    Building datafusion-expr v54.1.0 (baseline)
       Built [  26.745s] (baseline)
     Parsing datafusion-expr v54.1.0 (baseline)
      Parsed [   0.074s] (baseline)
    Checking datafusion-expr v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   1.183s] 223 checks: 221 pass, 2 fail, 0 warn, 30 skip

--- failure inherent_method_missing: pub method removed or renamed ---

Description:
A publicly-visible method or associated fn is no longer available under its prior name. It may have been renamed or removed entirely.
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#item-remove
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.49.0/src/lints/inherent_method_missing.ron

Failed in:
  PartitionBatchState::set_most_recent_row, previously in file /home/runner/work/datafusion/datafusion/target/semver-checks/git-apache_main/e1a21a4874346164a6142dda64cc03399b2e6fdf/datafusion/expr/src/window_state.rs:287

--- failure struct_pub_field_missing: pub struct's pub field removed or renamed ---

Description:
A publicly-visible struct has at least one public field that is no longer available under its prior name. It may have been renamed or removed entirely.
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#item-remove
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.49.0/src/lints/struct_pub_field_missing.ron

Failed in:
  field most_recent_row of struct PartitionBatchState, previously in file /home/runner/work/datafusion/datafusion/target/semver-checks/git-apache_main/e1a21a4874346164a6142dda64cc03399b2e6fdf/datafusion/expr/src/window_state.rs:255

     Summary semver requires new major version: 2 major and 0 minor checks failed
    Finished [  55.460s] datafusion-expr
    Building datafusion-physical-expr v54.1.0 (current)
       Built [  28.164s] (current)
     Parsing datafusion-physical-expr v54.1.0 (current)
      Parsed [   0.046s] (current)
    Building datafusion-physical-expr v54.1.0 (baseline)
       Built [  28.004s] (baseline)
     Parsing datafusion-physical-expr v54.1.0 (baseline)
      Parsed [   0.048s] (baseline)
    Checking datafusion-physical-expr v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.350s] 223 checks: 222 pass, 1 fail, 0 warn, 30 skip

--- failure trait_method_parameter_count_changed: pub trait method parameter count changed ---

Description:
A trait method now takes a different number of parameters.
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#trait-item-signature
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.49.0/src/lints/trait_method_parameter_count_changed.ron

Failed in:
  WindowExpr::evaluate_stateful now takes 3 instead of 2 parameters, in file /home/runner/work/datafusion/datafusion/datafusion/physical-expr/src/window/window_expr.rs:105

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  57.395s] datafusion-physical-expr
    Building datafusion-physical-plan v54.1.0 (current)
       Built [  36.662s] (current)
     Parsing datafusion-physical-plan v54.1.0 (current)
      Parsed [   0.136s] (current)
    Building datafusion-physical-plan v54.1.0 (baseline)
       Built [  36.977s] (baseline)
     Parsing datafusion-physical-plan v54.1.0 (baseline)
      Parsed [   0.140s] (baseline)
    Checking datafusion-physical-plan v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.644s] 223 checks: 223 pass, 30 skip
     Summary no semver update required
    Finished [  75.649s] datafusion-physical-plan

[codecov-commenter]

Codecov Report

❌ Patch coverage is 86.66667% with 4 lines in your changes missing coverage. Please review.
✅ Project coverage is 80.85%. Comparing base (dbcb5c0) to head (b337cec).

Files with missing lines	Patch %	Lines
datafusion/physical-expr/src/window/window_expr.rs	85.71%	1 Missing and 1 partial ⚠️
datafusion/physical-expr/src/window/aggregate.rs	50.00%	0 Missing and 1 partial ⚠️
...ysical-plan/src/windows/bounded_window_agg_exec.rs	90.90%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24033      +/-   ##
==========================================
- Coverage   80.85%   80.85%   -0.01%     
==========================================
  Files        1101     1101              
  Lines      374933   374941       +8     
  Branches   374933   374941       +8     
==========================================
  Hits       303166   303166              
- Misses      53671    53677       +6     
- Partials    18096    18098       +2     

☔ 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.