[C++] Implement hash_aggregate exact quantile kernel

[asfimport]

Requires ARROW-12759.

We have a scalar aggregate kernel for this already and hopefully the implementation can be reused.

Also create a pair of kernels median() and hash_median() that wrap this for convenient use by bindings (like in ARROW-14052).

Reporter: David Li / @lidavidm

Related issues:

• [C++] Implement hash_aggregate kernels (umbrella issue) (is a child of)
• [C++] Implement hash_aggregate approximate quantile kernel (is related to)
• [R] Bind median() and quantile() to exact not approximate median and quantile (is depended upon by)

_{Note: This issue was originally created as ARROW-13309. Please see the migration documentation for further details.}

[asfimport]

David Li / @lidavidm:
This is kind of tricky right now since we'll have to keep all the data materialized somewhere. The equivalent scalar aggregate kernel is actually a vector kernel and hence sidesteps this.

[asfimport]

Ian Cook / @ianmcook:
It's common for distributed query engines to have approximate quantile functions. I believe Spark and Impala have only approximate quantile functions, no exact versions. Some others like Hive have both, but with limitations on the exact version (for example in Hive I believe it only works on integer data.) I think it would be wise for us to implement an approximate version first: ARROW-13520.

[asfimport]

Ian Cook / @ianmcook:
@lidavidm does the exact hash_quantile kernel seem doable for 6.0.0? I'm asking not to apply pressure but because we need to know for the R bindings. Thanks!

[asfimport]

David Li / @lidavidm:
@ianmcook I think if we want to have one, we certainly can, but we'd be accumulating all data in-memory until we figure out a design for spilling to disk. That may be fine? I would only worry about potentially wasting work if we come up with a drastically different design for spilling to disk.

And we'd want to refactor the scalar aggregate to actually be a scalar aggregate kernel again (right now it is actually a vector function and hence not usable in an aggregation).

[asfimport]

Ian Cook / @ianmcook:
@lidavidm thanks, I think it's fine to wait for now, I do not think this needs to take precedence over other work

[github-actions]

This issue has been marked as stale because it has had no activity in the past 365 days. Please remove the stale label or comment below, or this issue will be closed in 14 days. If this improvement is still desired but has no current owner, please add the 'Status: needs champion' label.

[ghostiee-11]

could this be reopened? it was closed by the stale bot rather than by a decision, and the last human comment was a deferral rather than a no.

concrete impact: hash_approximate_median on a two element group returns the lower value instead of the mean.

import pyarrow as pa

t = pa.table({"g": ["a", "a"], "v": [10.0, 20.0]})
print(t.group_by("g").aggregate([("v", "approximate_median")]).to_pydict())
# {'g': ['a'], 'v_approximate_median': [10.0]} exact median is 15.0

hash_tdigest returns 10.0 too.

narwhals maps its median() onto these, so a grouped median is silently wrong there for small groups: narwhals-dev/narwhals#3853. the scalar path can be fixed with quantile(q=0.5) since that one is exact, but there is no hash_quantile to switch to for grouped.

one data point on cost, scalar only: pc.quantile(q=0.5, interpolation="linear") measured 3 to 4x faster than pc.approximate_median from n=2 up to n=10M here. that says nothing about what a grouped implementation would cost, and the in memory accumulation concern from 2021 still stands.

pyarrow 23.0.1