apache / apache/arrow

[C++] Options for handling non-decomposable aggregate functions

Open
#34,387 3 comments 0 reactions 0 assignees View on GitHub
Component: C++ Type: enhancement
Dominant language
C++
Stars
17.1k
Forks
4.3k
Avg merge
3d 13h
Merged PRs (30d)
88

Description

### Describe the enhancement requested

Currently we have `SCALAR_AGGREGATE` and `HASH_AGGREGATE` function kinds. These are used for ["decomposable aggregate functions"](https://en.wikipedia.org/wiki/Aggregate_function#Decomposable_aggregate_functions). The execution is split across four steps, `init`, `consume` (e.g. map) , `merge` (e.g. reduce), and `finalize` (though only `consume` and `merge` are truly essential).

However, some aggregate functions are not decomposable (e.g. median, mode), typically because they want to work on sorted input. These functions can technically be mapped to a decomposable function where the `consume` function simply accumulates all values and the `merge` function then sorts and calculates the statistic. However, this is not a terribly useful formulation and requires UDF authors to write boilerplate.

I can think of a variety of potential approaches. We don't have to pick one and maybe none of these make sense in the long run.

### Simple Aggregate Function

These functions would have a single step `exec` which receives a table (not a batch!) as input and outputs a single scalar value. These are very similar to `VECTOR` functions except they must output only a single value. The main advantage over `SCALAR_AGGREGATE` is that you can avoid accumulation. The primary disadvantage is that they require loading all of the data in memory. Still, these would be useful when the data fits in memory. They also make it very simple to write custom aggregates without having to worry about whether things fit in RAM or not.

* Question: Is it worth doing this when we already have vector functions? The constraint that we output only a single row would make it easier to mix these in with regular aggregates. I don't know if that is worthwhile.

### Ordered Stream Semi-Aggregate Function

These functions would have an `init` step which receives the total # of elements. They would then have a `consume` step which receives a batch of data. However, these batches are guaranteed to be ordered in some way (usually I would assume, by one of the input arguments). The output could be any number of rows (hence **semi**-aggregate). This could be a useful way to implement something like "value_counts" in a streaming fashion. Each time the value changes you could simply output a row with the # of instances of the previous value.

* Question: Are there other functions that could benefit from this? If not, maybe "distinct counts" could just be a standalone node and we avoid the complexity of having a new function type.

### Something else?

Neither of the above would yield the optimal implementation of median. So that may end up being a dedicated node no matter what. The optimal way (to the best of my knowledge) to handle median would be to do a spilling sort where you spill into a range-partitioned spill file. Then, you should be able to identify a single bucket to read from. For example, if your values range from 0-1000 you could partition into 0-100, 100-200, 200-300, ... while you are spilling and keep track of the size of each bucket. Then, you just pick the bucket that contains the median and use either of the above approaches.

Is this some kind of general and extensible function here? E.g. you could define an `ordered stream aggregate function` that returns a single row. Then the `init` function could receive the "total # of values" and return the "# of values to skip" but I find it hard to imagine that would be useful for anything other than just median.

### Component(s)

C++

Contributor guide

Open the contributing guide

Research direction

Start by reviewing the existing SCALAR_AGGREGATE and HASH_AGGREGATE execution model, including their init, consume, merge, and finalize steps. Compare the proposed simple aggregate and ordered stream semi-aggregate options, and define a focused API and execution scope before implementation; the issue does not identify files or tests, so completion criteria need maintainer agreement.

Written by the indexing model from the issue text.

Assessment

Tech stack
cpp
Domain
data-engineering
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.