Implement GroupsAccumulator for corr(x,y) aggregate function

[2010YOUY01]

Which issue does this PR close?

Closes #13549

Rationale for this change

Implement GroupsAccumulator for corr aggregation function, for better performance when group cardinality is high

I rerun the H2o benchmark:

Data Generation

falsa groupby --path-prefix=/Users/yongting/data/ --size MEDIUM --data-format PARQUET
https://github.com/mrpowers-io/falsa

Run benchmark in datafusion-cli

CREATE EXTERNAL TABLE IF NOT EXISTS h2o_100m (
    id1 VARCHAR NOT NULL,
    id2 VARCHAR NOT NULL,
    id3 VARCHAR NOT NULL,
    id4 INTEGER NOT NULL,
    id5 INTEGER NOT NULL,
    id6 INTEGER NOT NULL,
    v1 INTEGER NOT NULL,
    v2 INTEGER NOT NULL,
    v3 DOUBLE PRECISION NOT NULL
)
STORED AS parquet
LOCATION '/Users/yongting/data/G1_1e8_1e8_100_0.parquet';

select id2, id4, power(corr(v1, v2), 2) as r2 from h2o_100m group by id2, id4;

Result

Main: 12s
This PR: 4s
(On my MacBook with m4 pro)

---

Remaining tasks
Implement convert_to_states() This requires changes in aggregate fuzzer for test coverage, which can be done later to keep this PR small

What changes are included in this PR?

1. Implement two utility functions: accumulate_multiple and accumulate_correlation_states to accumulate states in correlation function. (existing util functions is for aggregate functions with 1 input expr avg(expr1) v.s. corr(expr1, expr2))
2. Implement GroupsAccumulator for corr()

Are these changes tested?

Unit tests for util functions
corr() is covered by existing tests

Are there any user-facing changes?

No

[alamb]

This looks amazing -- thank you @2010YOUY01

I plan to review it over the next day or two

It seems like maybe we should add the data generator for h2o benchmark to the bench.sh script 🤔

[Dandandan]

If passing combined_nulls to NullBuffer::union it will take care of handling Option

[Dandandan]

I think collecting into Vec might not be necessary?

[2010YOUY01]

This is true, I have updated

[Dandandan]

It would be nice to avoid collecting here? Can we take an iterator instead in value_fn?

[2010YOUY01]

The bench query above went from 4s -> 3s after this change! Great catch
Though I have to structure the code differently to make it compile, it's a bit more complex so I added more comment

[Dandandan]

Nice!

[jayzhan211]

I think casting should be handled in logical optimizer. Fixing the signature of Correlation might helps

[2010YOUY01]

You're right, this line is redundant
Addressed in 98cba91

[2010YOUY01]

I tried to remove it, all tests passed but the benchmark query above won't run 🤔 The existing signature looks correct to me, this might need further investigation
Issued #13721

[Dandandan]

Hi, I think this one is pretty close, do you have time to look at the review comments @2010YOUY01 ?

[2010YOUY01]

Hi, I think this one is pretty close, do you have time to look at the review comments @2010YOUY01 ?

Yes, I will be back and finish this PR in next 2 days, I'm traveling and afk this week. Thanks for the attention to this ticket

[alamb]

I also harbor hopes of contributing a benchmark for corr, hopefully

(to be clear not for this PR)

[2010YOUY01]

Thank you all for the review, it's ready for another look

[Dandandan]

Very nice work @2010YOUY01