fix(functions-aggregate): drain CORR state vectors for streaming aggregation

[geoffreyclaude]

Which issue does this PR close?

• N/A

Rationale for this change

This change addresses a failure in the CORR aggregate function when running in streaming mode. The CorrelationGroupsAccumulator (introduced in PR #13581) was failing to drain its state vectors during EmitTo::First calls, causing internal state to persist across emissions. This led to memory leaks, incorrect results for subsequent groups, and "length mismatch" errors because the internal vector sizes diverged from the number of emitted groups.

Reproducer

# Setup data
CREATE TABLE stream_test (
    g INT,
    x DOUBLE,
    y DOUBLE
) AS VALUES
(1, 1.0, 1.0), (1, 2.0, 2.0),
(2, 1.0, 5.0), (2, 2.0, 5.0),
(3, 1.0, 1.0), (3, 2.0, 2.0);

# Trigger streaming aggregation via sorted subquery
SELECT
  g,
  CORR(x, y)
FROM (SELECT * FROM stream_test ORDER BY g LIMIT 10000)
GROUP BY g
ORDER BY g;

Before: DataFusion error: Arrow error: Invalid argument error: all columns in a record batch must have the same length

After:

1 1
2 NULL
3 1

What changes are included in this PR?

This PR is structured into two commits: the first adds a failing test case to demonstrate the issue, and the second implements the fix.

The accumulator now uses emit_to.take_needed() in both evaluate and state to properly consume the emitted portions of the state vectors. Additionally, the size() implementation has been updated to use vector capacity for more accurate memory accounting.

Are these changes tested?

Yes, a new test case in aggregate.slt triggers streaming aggregation via an ordered subquery. This test previously crashed with an Arrow length mismatch error and now produces correct results.

Are there any user-facing changes?

Yes, SQL queries that trigger streaming aggregation using CORR (typically those with specific ordering requirements) will now succeed instead of failing with a length mismatch error.

[martin-g]

It would be good to add a companion EXPLAIN query to verify that it uses the streaming path.

[geoffreyclaude]

I had it at first, and removed it as I found it too verbose. Same with a dedicated unit test in correlation.rs, which seemed out of place and only serving as a "demo" of the bug.

Adding just the EXPLAIN for CORR seems too specific to me here. However, I think it would make a lot of sense to actually have a dedicated .slt that runs EXPLAIN and the actual query for all aggregates.

@martin-g WDYT?

EDIT: pushed new comprehensive tests in commit test: add comprehensive aggregate tests for streaming aggregation

[martin-g]

Either way is fine as long as there is a way to assert that it behaves the way it is supposed to be.

[martin-g]

LGTM

[Jefffrey]

I think just need to resolve conflict then should be good to go @geoffreyclaude

[Jefffrey]

Thanks @geoffreyclaude & @martin-g