Support pushdown of count aggregation with distinct

[alexjo2144]

Moves the distinct aggregation rewriting optimizers to after PushAggregationIntoTableScan.

TODO: I'm not as familiar with testing around PlanOptimizers, is there a good way to test this?

#4323

[findepi]

TODO: I'm not as familiar with testing around PlanOptimizers, is there a good way to test this?

i am afraid there isn't, since we don't have a mock connector for exercising aggregation pushdown SPI calls.
Unless you create one.

[findepi]

Taking these words back. Since you're making ImplementCount support DISTINCT aggregations, you should be able to verify pushdown is happening in io.trino.plugin.jdbc.BaseJdbcConnectorTest#testAggregationPushdown (eg with PostgreSQL)

[findepi]

@alexjo2144 can you please review build output?

[alexjo2144]

@alexjo2144 can you please review build output?

Yeah, I need to think about this a little more. There are some pushdowns that I'm still not fully understanding.

For example, with the original code SELECT min(DISTINCT regionkey) is fully pushed down as SELECT min("regionkey") AS "tmp" FROM (SELECT "regionkey" FROM "tpch"."nation" GROUP BY "regionkey"), which is a little odd. However, after my changes only the groupby is pushded down, which is definitely not what we want.

[losipiuk]

@findepi should be good now (I hope). Please give it another look.

[findepi]

Allows pushdown of multiple distinct aggregations as long as
only one non-count aggregation is distinct.

i may have asked about that but don't remember answer -- why "only one non-count" ?

[losipiuk]

I am not sure what exactly @alexjo2144 had in mind here.
There are many constraints here. And definitely not all queries which satisfy the condition from commit message will be be fully pushed down

E.g. this one will not:

select max(distinct regionkey), count(distinct regionkey), count(distinct nationkey) from nation;

I suggest to just leave title line in commit message and drop the rest.

[findepi]

redudantn move

[hashhar]

Is this related?

[findepi]

"core/trino-main/src/main/java/io/trino/sql/planner/PlanOptimizers.java"

[findepi]

add a comment why this is inside pushIntoTableScanRulesExceptJoins set, otherwise looks like unintentional

maybe it shouldn't be here? instead, PushLimitThroughMarkDistinct could come together with MultipleDistinctAggregationToMarkDistinct?

[losipiuk]

maybe it shouldn't be here? instead, PushLimitThroughMarkDistinct could come together with MultipleDistinctAggregationToMarkDistinct?

per @kasiafi

After removing MultipleDistinctAggregationToMarkDistinct from an IterativeOptimizer, there's no way that a MarkDistinctNode could be present at that stage, so the rule PushLimitThroughMarkDistinct can be also removed. However, I'd keep it there next to other PushLimit rules in case it's needed again in the future. Actually, the PushLimit rules could be extracted like projectionPushdownRules.

[losipiuk]

never mind. Reworking after chat with @kasiafi

[findepi]

We had MultipleDistinctAggregationToMarkDistinct once and now we have twice. What's the reason?

[losipiuk]

Does not make sense.

[losipiuk]

@kasiafi done (hopefully) as we discussed. Much simpler. Thanks. Let's see what CI says.

[findepi]

Suggested change

	.addAll(limitPushdownRules)
	.addAll(limitPushdownRules) // including through MarkDistinct

[losipiuk]

Actually having this here results in test failures. @kasiafi looked deeply into that and the problem comes from extra call of PushLimitThroughOuterJoin and the fact that applyJoin will not consume same limit twice.
So without this optimization went

... Limit -> OuterJoin -> Scan(no limit)
(PushLimitThroughOuterJoin)
... Limit -> OuterJoin -> Limit -> Scan(no limit)
(PushLimitIntoTableScan)
... Limit -> OuterJoin -> Scan(with limit)

now join is pushable to TS

With this extra line the optimization continues from when it completed previously

... Limit -> OuterJoin -> Scan(with limit)
(PushLimitThroughOuterJoin)
... Limit -> OuterJoin -> Limit -> Scan(with limit)
(PushLimitIntoTableScan; applyJoin return Optional.empty() because of https://github.com/trinodb/trino/blob/408476e5d4067db61324f773510c27731373b91b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultJdbcMetadata.java#L459-L461)
... Limit -> OuterJoin -> Limit -> Scan(with limit)

!!!!! join no longer pushable to table scan

This could be addressed with #9056 but I am not super convinced this is the way to go.

[losipiuk]

@findepi, @kasiafi I suggest we go without limit pushdown here for now. It should not matter. What was to be pushed down should be already.
Do you feel it is good to be merged?

[kasiafi]

The change in PlanOptimizers looks OK.

[findepi]

... so that multiple distinct aggregations can be pushed into a connector

[findepi]

i wonder whether we actually considered solving the problem differently. instead of changing rule order, we could have a rule that matches to MarkDistinct and invokes aggregation pushdown.

trying this out could be a follow up. if successful, we could simply this class, presumably even rolling the changes back

[kasiafi]

MarkDistinct is not always a result of the MultipleDistinctAggregationToMarkDistinct rule. And when it is, it could be hard to identify the original aggregations because of intervening rules.
Also, applying rule MultipleDistinctAggregationToMarkDistinct later might be possibly beneficial from the viewpoint of decorrelation.