Optimize distinct aggregation on multiple columns

[kaka11chen]

Fix #613 to optimize distinct aggregation on multi columns.

[cla-bot]

Thank you for your pull request and welcome to our community. We could not parse the GitHub identity of the following contributors: chenqi.
This is most likely caused by a git client misconfiguration; please make sure to:

1. check if your git client is configured with an email to sign commits git config --list | grep email
2. If not, set it up using git config --global user.email email@example.com
3. Make sure that the git commit email is configured in your GitHub account settings, see https://github.com/settings/emails

[cla-bot]

Thank you for your pull request and welcome to our community. We require contributors to sign our Contributor License Agreement, and we don't seem to have you on file. In order for us to review and merge your code, please submit the signed CLA to cla@prestosql.io. For more information, see https://github.com/prestosql/cla.

[sopel39]

Could you extract "$distinct" to static final. Which code produces symbols with $distinct suffix?

[kaka11chen]

OK, thanks. MultipleDistinctAggregationToMarkDistinct produces this.

[martint]

There's no guarantee that the name of the mask will be named xxx$distinct, so this check is brittle. We need a better way to identify which fields correspond to the distinct mask (by matching the field they originate from)

[wenleix]

What's the performance of this optimization compared with setting use_mark_distinct=false ? :)

[kaka11chen]

@wenleix

select count(distinct ss_item_sk), count(distinct ss_store_sk) from tpcds_bin_partitioned_orc_1000.store_sales;

Result: It is very slow, cost 60 seconds in our perf-test env, regardless of use-mark-distinct.

If I change it to

select count(case when grouping_id=1 and ss_item_sk is not null then 1 else null end) as c0, count(case when grouping_id=2 and ss_store_sk is not null then 1 else null end) as c1 from (select grouping(ss_item_sk,ss_store_sk) AS grouping_id, ss_item_sk, ss_store_sk from tpcds_bin_partitioned_orc_1000.store_sales group by grouping sets (ss_item_sk, ss_store_sk))

Result: It only cost 20 seconds in our perf-test env(3 nodes, each is 48 cores, 96G memory).

[kaka11chen]

@sopel39 @martint Any other suggestions?

[martint]

Some high-level comments:

• The OptimizeMixedDistinctAggregations is a legacy (Visitor-based optimizer), which we're trying to move away from. Ideally, we'd implement this as a Rule, instead.
• The optimization only works when use_mark_distinct is set to true. That causes a GROUP BY with mixed distincts to be turned into a sequence of MarkDistinct. OptimizeMixedDistinctAggregations tries to reverse-engineer that transformation and turn it into the grouping-sets based version.

I think a better approach than extending this optimizer is to implement a Rule that turns a GROUP BY + mixed distincts into a grouping sets-based GROUP BY. It will make the code much simpler as we wouldn't need to dig through the plan to find the MarkDistinct nodes that produce the corresponding "distinct" inputs to the aggregation. Once we do that, we can also get rid of this legacy optimizer.

Let me know if you need help doing this or if you need additional pointers and examples of where to look.

[martint]

Use a map builder for readability when there are many keys and values:

PlanMatchPattern tableScan = tableScan("orders", ImmutableMap.<String, String>builder()
        .put("TOTALPRICE", "totalprice")
        .put("CUSTKEY", "custkey")
        .put("ORDERDATE", "orderdate")
        .put("ORDERSTATUS", "orderstatus")
        .build());

[kaka11chen]

@martint Got it, thanks. I will implement as a Rule instead.

[kaka11chen]

@martint @sopel39 I have modified to implement as a Rule. Could you please review it? Thanks.

[martint]

Why is this needed?

[martint]

We can remove the OptimizeMixedDistinctAggregations class, too.

[martint]

<Set<Expression>> is not required. The compiler should be able to infer the correct type.

[martint]

If the goal is to find out whether the aggregation has any calls with DISTINCT, this can be simplified to:

return aggregation.getAggregations()
        .values().stream()
        .map(Aggregation::getCall)
        .anyMatch(FunctionCall::isDistinct);

[martint]

The name of this method is misleading. It returns true if the the aggregation has DISTINCT with a single argument or no aggregations with DISTINCT.

[martint]

This is trivial enough that I would just inline it where it's used.

[martint]

If you make nonDistinctAggregateSymbolsBuilder a Set builder, you can skip the calls to remove duplicates.

[martint]

This transformation is actually incorrect. Not all aggregations are insensitive to nulls, so the terms F(if(group = 1, c1)), etc. will call the aggregation with null values for other groups:

presto:default> select a, array_agg(b), array_agg(distinct c) from (values (1,1,1)) t(a,b,c) group by a;
 a | _col1 |   _col2
---+-------+-----------
 1 | [1]   | [null, 1]
(1 row)

This needs to be rewritten to F(c1) FILTER (WHERE group = 1). Also, please add a test for this case.

[martint]

I find this implementation hard to read. I think it would help if you structured it such that it builds things from the bottom up and return objects that contain the node that was created + anything that's needed to build the node above it. However, I haven't tried playing this out, so not sure how much cleaner it will make it. You'll have to give it a try and see how it goes.

[martint]

@kaka11chen, any updates on this?

[Donderia]

@martint @kaka11chen - is it still not available in in trino ? I had the same observation with multiple distincts with presto version 340

[colebow]

👋 @kaka11chen - this PR is inactive and doesn't seem to be under development. If you'd like to continue work on this at any point in the future, feel free to re-open.