Expose accumulator state to allow prefix scanning

[avantgardnerio]

Summary

Expose state of aggregate streams within BWAG so downstream prefix scanning can take place.

API

// physical-plan/src/windows/bounded_window_agg_exec.rs
pub type FinalizedWindowStateObserver = Arc<
    dyn Fn(usize, &PartitionKey, &[Option<Vec<ScalarValue>>]) -> Result<()>
        + Send + Sync,
>;

impl BoundedWindowAggExec {
    pub fn with_finalized_state_observer(mut self, obs: FinalizedWindowStateObserver) -> Self { … }
}

// physical-expr/src/window/window_expr.rs
impl WindowState {
    /// `Accumulator::state()` if this is an aggregate window function, `None` otherwise.
    pub fn aggregate_state(&mut self) -> Result<Option<Vec<ScalarValue>>> { … }
}

🤖 Generated with Claude Code

[codecov-commenter]

Codecov Report

❌ Patch coverage is 83.82353% with 22 lines in your changes missing coverage. Please review.
✅ Project coverage is 80.85%. Comparing base (dbcb5c0) to head (0767f73).
⚠️ Report is 11 commits behind head on main.

Files with missing lines	Patch %	Lines
...ysical-plan/src/windows/bounded_window_agg_exec.rs	83.96%	8 Missing and 13 partials ⚠️
datafusion/physical-expr/src/window/window_expr.rs	80.00%	1 Missing ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##             main   #24035    +/-   ##
========================================
  Coverage   80.85%   80.85%            
========================================
  Files        1101     1101            
  Lines      374933   375580   +647     
  Branches   374933   375580   +647     
========================================
+ Hits       303166   303692   +526     
- Misses      53671    53781   +110     
- Partials    18096    18107    +11     

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

[neilconway]

Thanks for this contribution @avantgardnerio ! I'd find it helpful if you'd elaborate a little bit more about the motivation for this change in the PR description. For example, some intended use-cases, what kind of performance improvement this unlocks, etc.

[avantgardnerio]

@neilconway I'm trying to speed up window functions using parallel prefix scans. I am presently incubating this in Ballista, and this is the minimum API exposure that I need to do it for non-decomposable operations like approx_distinct() (vs others like AVG, STDDEV via Welford / Chan, etc). Though, the work is certainly not limited to Ballista, it has shown improvement in DataFusion as well.

The jury is still out about re-partition cost vs performance benefit, but the signs are hopeful:

And at least from a big-O time perspective (table 1) it should be optimal for some queries (select my_agg() over unbounded preceding...)