Make partial aggregation adaptive

[lukasz-stec]

Description

This is an optimization for the HashAggregationOperator that is split into partial and final steps.
In case when partial aggregation step does not reduce the number of rows too much (e.g. 90 % of rows are unique) this step brings a small benefit in terms of network savings but costs a lot of CPU to do.
In this case, it would be beneficial to skip partial aggregation altogether at the planning time,
but given we don't always have reliable statistics for the number of unique values, especially in the intermediate query stages it is not easy to do.
Instead (although it's complementary to the planner changes) this adds simple runtime adaptation for the partial aggregation step, that sends raw, ungrouped rows to the final step if the ratio of unique to input rows is big enough (0.8 by default).

With this change, there is a still significant overhead on the partial step mainly in the PartitionedOutputOperator that has to handle the superfluous accumulator state for the raw rows + in the HashAggregationOperator that needs to create and populate this state.
There are potential improvements for this in both HashAggregationOperator and PartitionedOutputOperator that would limit the overhead.
Another possible approach is to have a separate pipe (as it has a different layout) from partial to final step with only the input pages without the accumulator state. This would eliminate almost all of the overhead but require larger changes in the core engine.

tpch/tpcds benchmark results for orc sf1000

part

overall ~6% TPCH and 1.5 % tpcds improvement. Most queries are not affected, some gain between 10 to 35%
adaptive-pa-part-nocode.pdf

uppart

overall 3.5% for tpch and 2.5% for tpcds
adaptive-pa-unpart-nocode.pdf

General information

Is this change a fix, improvement, new feature, refactoring, or other?

performance improvement

Is this a change to the core query engine, a connector, client library, or the SPI interfaces? (be specific)

core query engine (HashAggregationOperator)

How would you describe this change to a non-technical end user or system administrator?

Improves group by performance by skipping partial aggregation step

Related issues, pull requests, and links

Documentation

( x) No documentation is needed.
( ) Sufficient documentation is included in this PR.
( ) Documentation PR is available with #prnumber.
( ) Documentation issue #issuenumber is filed, and can be handled later.

Release notes

( ) No release notes entries required.
( x) Release notes entries required with the following suggested text:

Improve performance of GROUP BY with a large number of groups.

[skrzypo987]

This looks much easier that I anticipated.

[skrzypo987]

a->an

[skrzypo987]

Have you tested different thresholds?

[lukasz-stec]

No, I haven't. minRows is more or less constant per partial aggregation memory limit (PA will either compute the whole split or stop when the memory limit is reached but then the number of rows will be > 100K). So this is mainly a sanity check for some small splits.
For uniqueRowsRatio, I suspect that in tpcds/tpch most of the cases are either well below 0.8 or close to 1 so the exact number does not matter. That said I will run some benchmarks to confirm that.

[lukasz-stec]

I additionally ran tpch/tpcds sf1000 orc part with ratio set to 0.5, 0.9, 0.95. As I expected mostly no change. most of the queries either trigger adaptation or don't consistently across ratios. There are some differences in the overall result but I think this is due to variability.
adaptive-pa-part-ratios-nocode.pdf

[skrzypo987]

Why is this needed here?

[lukasz-stec]

the default value of 100K is way too large for tests (it will never fire), and I think it's good to check the adaptive part outside of unit tests but checking all possible aggregation variation with and without adaptation seems too much.

[skrzypo987]

That is a bit fishy. For me it's ok, but I guess someone can have problems with that.

[lukasz-stec]

I think this is an example of a general issue with trino properties in tests. Some properties like this one, are targeted at larger data scales, which means the functionality behind the property will never fire if not explicitly tested.
This is especially true for memory-related properties e.g. task.max-partial-aggregation-memory.
It seems to me, we should use scaled-down property values in "query" tests.

[sopel39]

Do not change the property here. Change property in unit tests. You can also add another child of AbstractTestAggregations with minimal PA limits.

[lukasz-stec]

wouldn't it be better if all query tests were run with the lower limit and not just AbstractTestAggregations?

[sopel39]

wouldn't it be better if all query tests were run with the lower limit and not just AbstractTestAggregations?

no. We specifically split different queries into different abstract test classes (join, aggregation, etc) so that we don't have to cross test everything. It's a mess.

[lukasz-stec]

ok, I added TestAdaptivePartialAggregation and removed the change here.

[lukasz-stec]

review comments addresed

[lukasz-stec]

No, I haven't. minRows is more or less constant per partial aggregation memory limit (PA will either compute the whole split or stop when the memory limit is reached but then the number of rows will be > 100K). So this is mainly a sanity check for some small splits.
For uniqueRowsRatio, I suspect that in tpcds/tpch most of the cases are either well below 0.8 or close to 1 so the exact number does not matter. That said I will run some benchmarks to confirm that.

[lukasz-stec]

the default value of 100K is way too large for tests (it will never fire), and I think it's good to check the adaptive part outside of unit tests but checking all possible aggregation variation with and without adaptation seems too much.

[skrzypo987]

LGTM % comments

[skrzypo987]

That is a bit fishy. For me it's ok, but I guess someone can have problems with that.

[lukasz-stec]

last set of comments addressed

[lukasz-stec]

I think this is an example of a general issue with trino properties in tests. Some properties like this one, are targeted at larger data scales, which means the functionality behind the property will never fire if not explicitly tested.
This is especially true for memory-related properties e.g. task.max-partial-aggregation-memory.
It seems to me, we should use scaled-down property values in "query" tests.

[sopel39]

some initial comments

[sopel39]

This should be called AggregationConfig and should have all aggregation properties from FeaturesConfig. Also, it should be different commit

[lukasz-stec]

I added AggregationConfig and moved some properties there. PTAL if this is full list

[sopel39]

Ideally, this should be adaptive in both ways (on and off), not only on => off. Do you have ideas how this can be made?

[lukasz-stec]

One idea is to choose one or more splits randomly to use partial aggregation once in a while (e.g. once enough rows have been processed since the last check). Depending on the results of PA, we could switch PA back on for every ongoing or new split.
This will complicate the code a lot and only help with the unusual data distribution case where first you process a lot of unique groups and the switch to splits with a small number of groups.
If we add off switch, we will also be vulnerable to ping pong cases where we do the opposite of what we should be doing because of the data distribution (i.e. we do PA for unique rows and send raw rows for very duplicated rows).

[sopel39]

This will complicate the code a lot and only help with the unusual data distribution case where first you process a lot of unique groups and the switch to splits with a small number of groups.

How do you know it's unusual?
You can turn this argument around and say we hit a bad file and we turned off PA completely while the rest of data is rather flat.

In other places we turn adaptivness on/off: DictionaryAwarePageProjectionWork#createDictionaryBlockProjection. Generally, I think it's a preferred way because prefix of query might not be representative for the remainder of the query.

At very least we need to have a TODO and a plan for that

[lukasz-stec]

How do you know it's unusual?

No such case in tpch/tpcds that I know of + hard for me to come up with real-life data set that would have this.
Also, if the partial aggregation is intermediate node after partitioned exchange it seems not likely to have this kind of skew.
For the source stage, if we had split level NDV stats, we could decide per split to disable adaptation if expected number of groups is small, or even decide that we skip partial aggregation if we know the number of distinct values is big

[sopel39]

No such case in tpch/tpcds that I know of + hard for me to come up with real-life data set that would have this.

Tpch/Tpcds is just a benchmark, but I could imagine some data social data where some even is generating a lot of unique rows, but only 10% of time. If he hit that data at the beginning, we would turn PA even when it's efficient

[sopel39]

please create an issue and add a TODO (in PartialAggregationController) for enabling partial aggregation adaptively

[lukasz-stec]

done #11361

[sopel39]

missing javadoc

[lukasz-stec]

added javadoc

[sopel39]

That seems low. Preferably you should make decision after you flushed PA buffer at least once, because then you can check if PA managed to reduce anything or not.

[lukasz-stec]

this is what is actually happening (the check is after a flush). This threshold is just a failsafe in case of some strange case of very small splits (ie split with 10 rows, potentially due to partitioning or something else).
So for a normal case, the check is after a full split so moire than 1M rows or, when the PA hits the memory limit, and this should be then around 100K to 400K for the default 16M limit.

[sopel39]

Let's just make it class PartialAggregationController

[lukasz-stec]

do you mean class PartialAggregationController implements PartialAggregationControl or drop the interface?

[sopel39]

Drop the interface, drop the tracker. Just keep the class PartialAggregationController

[sopel39]

why?

[lukasz-stec]

we would duplicate counts during the next flush otherwise

[sopel39]

onPartialAggregationFlush -> onFlush

[sopel39]

Just move this logic to onPartialAggregationFlush (make tracker dumb). onFlush is not frequent

[sopel39]

Let's remove factory and NoOpPartialAggregationControl and just make Optional<PartialAggregationController> partialAggregationControler in HashAggregationOperatorFactory

[lukasz-stec]

What is the benefit from Optional<PartialAggregationController>?
With the current setup the code is simple in the HashAggregationOperator e.g. partialAggregationTracker.onFlush() vs partialAggregationController.ifPresent(controller -> controller.onFlush(x, y))

[sopel39]

What is the benefit from Optional?

If you use Optional then you don't have to have PartialAggregationControl or PartialAggregationControlFactory as interfaces (since there is just single implementation). There isn't really going to be another implementation and it won't be pluggable really. Hence, an interface just to have noop is an overkill.

[sopel39]

you can just make it onFlush(int totalPositionCount, int uniquePositionCount). Make row tracking internal to HashAggregationOperator. Then you don't even need this tracker at all

[lukasz-stec]

This would complicate HashAggregationOperator, especially because it handles more cases than partial aggregation. Having these counts here makes it explicit that this is only relevant for partial aggregation.

[sopel39]

But you call these methods anyway. What we do here is simple row tracking. We don't need factory, interfaces and tracker for that. It's an overkill for what is a simple increment

[lukasz-stec]

This adds 3 instead of 1 field to the class that already has ~25 fields. I don't consider this a good practice.
That said, I refactored this as requested.

[lukasz-stec]

I extracted the AggregatonConfig to a separate commit + some comments addressed, some repied.

[lukasz-stec]

One idea is to choose one or more splits randomly to use partial aggregation once in a while (e.g. once enough rows have been processed since the last check). Depending on the results of PA, we could switch PA back on for every ongoing or new split.
This will complicate the code a lot and only help with the unusual data distribution case where first you process a lot of unique groups and the switch to splits with a small number of groups.
If we add off switch, we will also be vulnerable to ping pong cases where we do the opposite of what we should be doing because of the data distribution (i.e. we do PA for unique rows and send raw rows for very duplicated rows).

[lukasz-stec]

I used 'null object pattern' here. For not partial aggregation, this is gonna be NoOpPartialAggregationControl.
Optional or null here makes code more convoluted as it needs to be handled on every access.

[lukasz-stec]

for !step.isOutputPartial() case NoOpPartialAggregationTracker. isPartialAggregationDisabled returns false

[lukasz-stec]

do you mean class PartialAggregationController implements PartialAggregationControl or drop the interface?

[lukasz-stec]

we would duplicate counts during the next flush otherwise

[lukasz-stec]

this is what is actually happening (the check is after a flush). This threshold is just a failsafe in case of some strange case of very small splits (ie split with 10 rows, potentially due to partitioning or something else).
So for a normal case, the check is after a full split so moire than 1M rows or, when the PA hits the memory limit, and this should be then around 100K to 400K for the default 16M limit.

[lukasz-stec]

I added AggregationConfig and moved some properties there. PTAL if this is full list

[lukasz-stec]

What is the benefit from Optional<PartialAggregationController>?
With the current setup the code is simple in the HashAggregationOperator e.g. partialAggregationTracker.onFlush() vs partialAggregationController.ifPresent(controller -> controller.onFlush(x, y))

[lukasz-stec]

added javadoc

[sopel39]

See #11066 (comment). We should extract OptimizerConfig config file. I suggest you skip this extract for now since it will be part of #11066

[sopel39]

This will complicate the code a lot and only help with the unusual data distribution case where first you process a lot of unique groups and the switch to splits with a small number of groups.

How do you know it's unusual?
You can turn this argument around and say we hit a bad file and we turned off PA completely while the rest of data is rather flat.

In other places we turn adaptivness on/off: DictionaryAwarePageProjectionWork#createDictionaryBlockProjection. Generally, I think it's a preferred way because prefix of query might not be representative for the remainder of the query.

At very least we need to have a TODO and a plan for that

[sopel39]

ADAPTIVE_PARTIAL_AGGREGATION_UNIQUE_ROWS_RATIO -> ADAPTIVE_PARTIAL_AGGREGATION_UNIQUE_ROWS_RATIO_THRESHOLD

[sopel39]

Expand this description, threshold for what (on, off)?

[sopel39]

Optional or null here makes code more convoluted as it needs to be handled on every access.

If you use Optional then you don't have to have PartialAggregationControl or PartialAggregationControlFactory as interfaces (since there is just single implementation). There isn't really going to be another implementation and it won't be pluggable really. Hence, an interface just to have noop is an overkill.

[sopel39]

Drop the interface, drop the tracker. Just keep the class PartialAggregationController

[sopel39]

But you call these methods anyway. What we do here is simple row tracking. We don't need factory, interfaces and tracker for that. It's an overkill for what is a simple increment

[sopel39]

What is the benefit from Optional?

If you use Optional then you don't have to have PartialAggregationControl or PartialAggregationControlFactory as interfaces (since there is just single implementation). There isn't really going to be another implementation and it won't be pluggable really. Hence, an interface just to have noop is an overkill.

[lukasz-stec]

AggregationConfig extraction dropped, properties moved to the FeaturesConfig for now.
NoOpPartialAggregationControl refactored to Optional

[lukasz-stec]

How do you know it's unusual?

No such case in tpch/tpcds that I know of + hard for me to come up with real-life data set that would have this.
Also, if the partial aggregation is intermediate node after partitioned exchange it seems not likely to have this kind of skew.
For the source stage, if we had split level NDV stats, we could decide per split to disable adaptation if expected number of groups is small, or even decide that we skip partial aggregation if we know the number of distinct values is big

[lukasz-stec]

This adds 3 instead of 1 field to the class that already has ~25 fields. I don't consider this a good practice.
That said, I refactored this as requested.

[sopel39]

mostly lgtm % comments

[sopel39]

you need to rebase. There is new OptimizerConfig class

[lukasz-stec]

moved to OptimizerConfig

[sopel39]

add description

[sopel39]

add description

[sopel39]

No such case in tpch/tpcds that I know of + hard for me to come up with real-life data set that would have this.

Tpch/Tpcds is just a benchmark, but I could imagine some data social data where some even is generating a lot of unique rows, but only 10% of time. If he hit that data at the beginning, we would turn PA even when it's efficient

[sopel39]

You need a controller factory. Just create PartialAggregationController in LocalExecutionPlanner.Visitor#createHashAggregationOperatorFactory.

[lukasz-stec]

removed the factory, added io.trino.operator.aggregation.partial.PartialAggregationController#duplicate

[sopel39]

constructOutputPage should return Page

[sopel39]

Do not change the property here. Change property in unit tests. You can also add another child of AbstractTestAggregations with minimal PA limits.

[sopel39]

so this intermediate page has 10 positions? Could we use aggregation and input data that would actually squash input positions (e.g. all rows belong to same group)

[lukasz-stec]

not sure I understand but the reason we need (almost) unique groups is to trigger adaptation.

[sopel39]

Ok, so can we use 9 unique rows out of 10 rows? Or maybe you change ration from 0.8 into 0.5? The reason is the I would like to see that aggregation actually happens before it's disabled

[lukasz-stec]

refactored to 9 out of 10 unique (0, 1, 2, 3, 4, 5, 6, 7, 8, 8)

[sopel39]

Also add a test (or change this one) that there needs to be at least one flush before PA is disabled (E.g. low min row count, and flush happening after second input page.

[lukasz-stec]

Added separate test case as this requires using HashAggregationOperatorFactory with a different maxPartialMemory limit

[sopel39]

please rebase due to conflicts

[lukasz-stec]

rebased on the master and moved config properties to OptimizerConfig + other comments addressed

[lukasz-stec]

moved to OptimizerConfig

[lukasz-stec]

I don't want to have if (step.isOutputPartial()) because if influences other branches/ifs below, but I added step.isOutputPartial() to the condition to make it clear that this works only for partial aggregation.
I added partialAggregationController.isPresent() vs step.isOutputPartial() to the constructor (we dont need to check it for every addInput)

[lukasz-stec]

wouldn't it be better if all query tests were run with the lower limit and not just AbstractTestAggregations?

[lukasz-stec]

not sure I understand but the reason we need (almost) unique groups is to trigger adaptation.

[lukasz-stec]

Added separate test case as this requires using HashAggregationOperatorFactory with a different maxPartialMemory limit

[lukasz-stec]

removed the factory, added io.trino.operator.aggregation.partial.PartialAggregationController#duplicate

[sopel39]

mostly lgtm

[sopel39]

Use descriptions from session properties (make these consistent)

[sopel39]

nit: inputHash -> channelIndex

[sopel39]

wouldn't it be better if all query tests were run with the lower limit and not just AbstractTestAggregations?

no. We specifically split different queries into different abstract test classes (join, aggregation, etc) so that we don't have to cross test everything. It's a mess.

[sopel39]

Ok, so can we use 9 unique rows out of 10 rows? Or maybe you change ration from 0.8 into 0.5? The reason is the I would like to see that aggregation actually happens before it's disabled

[sopel39]

you can use one of io.trino.operator.OperatorAssertion#assertOperatorEquals instead of creating a new method (see TestHashJoinOperator)

[lukasz-stec]

OperatorAssertion#assertOperatorEquals closes the factory and in this case I need the factory to be reused but I extended assertOperatorEquals with boolean closeOperatorFactory so it works now.

[sopel39]

please also add assertion on PartialAggregationController#isPartialAggregationDisabled

[lukasz-stec]

done

[sopel39]

please also add assertion on PartialAggregationController#isPartialAggregationDisabled

[lukasz-stec]

added

[lukasz-stec]

comments addressed + added a commit to skip types-check in TestShowQueries (tests failed because of new properties with longer names were added)

[lukasz-stec]

done #11361

[lukasz-stec]

renamed

[lukasz-stec]

done

[lukasz-stec]

ok, I added TestAdaptivePartialAggregation and removed the change here.

[lukasz-stec]

added

[lukasz-stec]

refactored to 9 out of 10 unique (0, 1, 2, 3, 4, 5, 6, 7, 8, 8)

[lukasz-stec]

done

[lukasz-stec]

OperatorAssertion#assertOperatorEquals closes the factory and in this case I need the factory to be reused but I extended assertOperatorEquals with boolean closeOperatorFactory so it works now.

[sopel39]

lgtm % comments

[sopel39]

mark result as @Nullable

[sopel39]

nit: restore newline

[lukasz-stec]

restored after the checkArgument

[sopel39]

future improvement. It would be great to actually collect metrics

• how many pages were processed via skip aggregation builder
• how many flushes there were for PA.
• what average row count per flush
• etc..

This can be returned via io.trino.operator.OperatorContext#setLatestMetrics

@lukasz-stec Maybe create an issue for that?

[lukasz-stec]

Good idea. This would allow easier monitoring of the adaptation.
#11376 created.

[sopel39]

technically, your output page might become quite large since we add more columns. However, this probably isn't a big issue.
In io.trino.operator.project.PageProcessor we avoid this by keeping page under 4MB

[lukasz-stec]

Good point. I guess this currently can increase page size easily 10x by adding the accumulator state (given enough aggregations).
Without much overhead, we could use Page.getRegion to partition the output page into smaller pages but that would retain the full page until all pages are processed. WDYT?

[sopel39]

nit: tha -> the

[sopel39]

unused, remove

[sopel39]

inline with previous line

set it to 0?

you should also reduce PA buffer to 0 so that flush happens after every page

[lukasz-stec]

yeah, min-rows=0 makes sense.
I would leave task.max-partial-aggregation-memory with default as flushing per page is not a realistic scenario and there should be enough splits in the test data for flush per split to use adaptation.

[sopel39]

and there should be enough splits in the test data for flush per split to use adaptation.

I'm not so sure since we use tiny schema. I would rather reduce it to make sure it triggers in most cases

[lukasz-stec]

ok, I added task.max-partial-aggregation-memory=0B

[martint]

Does this produce pages with a schema that matches the one defined by the query plan?

[lukasz-stec]

Yes. This produces a "projected" input page by selecting group by channels + optionally hash channel + aggregation accumulators state channels, which matches aggregation partial step output schema.

[martint]

I'm not sure I understand how the input data is passed to the final aggregation, then. Specifically, the comment above that says: "It passes the input pages, augmented with initial accumulator state to the output". If that's the case, then the columns from the input to the aggregation would not match the expected input of the final aggregation (according to the query plan)

[lukasz-stec]

The comment may be not precise enough. What is happening is we take the group by (or hash) channels from the input and add the accumulator state. This is done in the constructOutputPage method.
BlockBuilder[] outputBuilders contain accumaltor state.

private Page constructOutputPage(Page page, BlockBuilder[] outputBuilders)
{
        Block[] outputBlocks = new Block[hashChannels.length + outputBuilders.length];
        for (int i = 0; i < hashChannels.length; i++) {
            outputBlocks[i] = page.getBlock(hashChannels[i]);
        }
        for (int i = 0; i < outputBuilders.length; i++) {
            outputBlocks[hashChannels.length + i] = outputBuilders[i].build();
        }
        return new Page(page.getPositionCount(), outputBlocks);
    }

[sopel39]

@martint
With this PR partial aggregation will still produce intermediate rows when it's turned off. We haven't implemented dual representation (raw rows vs intermediate rows). Approach as in this PR is simpler and will be improved.
For example, @radek-starburst is working on splitting single decimal aggregation state into smaller, primitive states. When this is done, we would be able to essentially passthrough input decimals when PA is adaptively disabled without any extra CPU cost. For example, for sum decimal aggregation overflow can be represented as RLE block.
@lukasz-stec is working on sending RLE blocks via partitioned exchange, so that intermediate aggregation rows can be serialized and transmitted over network more efficiently.

[sopel39]

thanks!