apache / apache/pinot

Partitioned Aggregation Support

Open
#12,057 6 comments 2 reactions 0 assignees View on GitHub
design-review enhancement feature performance stale
Dominant language
Java
Stars
6.1k
Forks
1.5k
Avg merge
2d 3h
Merged PRs (30d)
195

Description

Add support to optimise aggregation for partitioned columns.

An example where the optimisation would be beneficial is distinct count, as it is sufficient to count distinct values within each partition and then a simple sum of the counts per partition would yield the final aggregate result.

There is in fact an aggregation function today that uses this notion, [`SEGMENTPARTITIONEDDISTINCTCOUNT` ](https://docs.pinot.apache.org/configuration-reference/functions/segmentpartitioneddistinctcount), however it only handles cases where the whole partition fits in a single segment, so it is not applicable to most Pinot use cases (eg time-based).

A reasonable approach is to support the partitioned aggregation optimisation for [partitioned segment assignment](https://docs.pinot.apache.org/operators/operating-pinot/segment-assignment#partitioned-replica-group-segment-assignment) configurations, so we can effectively assume that all segments for a given partition will be co-located on the same server.

This means that there are two key aspects that remain the same:
1. Aggregation within each segment remains the same, as a segment belongs to a single partition.
2. Broker aggregation across server results remains the same, all partitioning concerns can be resolved within the server.

The proposal is for the [`AggregationFunction`](https://github.com/apache/pinot/blob/master/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java) to have an extra stage allowing to merge results within each partition if the partitioning optimisation applies.

There are multiple ways to go about supporting this.

New methods could look something like this:
```
boolean isPartitionedAggregation();
IntermediateResult extractPartitionResult(IntermediateResult intermediateResult);
```

A more involved change would have a new generic type added, and consequently a new merge function to aggregate across partitions.
```
boolean isPartitionAggregation();
PartitionResult extractPartitionResult(IntermediateResult intermediateResult);
PartitionResult mergePartitions(PartitionResult partitionResult1, PartitionResult partitionResult2);
FinalResult extractFinalResult(PartitionResult partitionResult);
```

An even more involved change would allow to have a different column type (tho this is likely unnecessary given `OBJECT` can be used anyways).
```
ColumnDataType getPartitionResultColumnType();
```

We can either add default methods to `AggregationFunction` (eg returning `false`) or we can have a new interface `PartitionedAggregationFunction` extending `AggregationFunction` (then where necessary check if the aggregation function object implements the new interface).
The latter has the benefit that we could more easily introduce a new generic type to represent the partition level intermediate result.

The flow at a high level would be:
1. Aggregate blocks within segment as usual.
2. Extract intermediate aggregation result as usual.
3. Aggregate across segments as usual with `merge` function, but keeping aggregation across partitions separate.
4. Extract partition result from each aggregated partition via `extractPartitionResult`.
5. Merge across partitions via `mergePartitions`.
6. Extract the final result from merged partitioned result.

Another alternative is to consider the `IntermediateResult` to be the final phase of aggregation across partitions and instead add a new merge function for inter-segment aggregation within partitions, in this case the type within partitions is the one that is "different".
```
boolean isPartitionedAggregation();
PartitionedResult extractAggregationResult(AggregationResultHolder aggregationResultHolder);
PartitionedResult extractGroupByResult(GroupByResultHolder groupByResultHolder, int groupKey);
PartitionedResult mergeWithinPartition(PartitionedResult partitionedResult1, PartitionedResult partitionedResult2);
IntermediateResult extractPartitionResult(PartitionedResult partitionedResult);
```

In this alternative the flow at a high level would be:
1. Aggregate blocks within segment as usual.
2. Extract partitioned results via `extractAggregationResult` or `extractGroupByResult`.
3. Aggregate across segments within partition via `mergeWithinPartition`
4. Extract intermediate result from each aggregated partition via `extractPartitionResult`.
5. Merge across partitions via regular `merge`.
6. Extract the final result from intermediate result as usual.

The advantage of this alternative is that we could possibly consider the PartitionedResult to be an internal structure to the server that does not need to be serialized, similar to block level aggregation, then the intermediate result continues to be used for broker level aggregation.

In either case, the core of the changes to the aggregation logic in order to support this partitioning optimisation would be mainly in `AggregationCombineOperator`, `GroupByCombineOperator` and `AggregationResultsBlockMerger`.

Basically, the intermediate aggregation would be done across segments within each partition based on which partition the segment belongs to, partition results can then be extracted and finally merged across partitions.

Contributor guide

Open the contributing guide

Research direction

Start by reading AggregationFunction and tracing the aggregation flow through AggregationCombineOperator, GroupByCombineOperator, and AggregationResultsBlockMerger. Compare the proposed partition-level aggregation alternatives and determine which design fits partitioned segment assignment. Done means partition-aware aggregation works across segments within each partition while preserving the existing broker aggregation behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
databases, distributed-systems
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.