trino icon indicating copy to clipboard operation
trino copied to clipboard

Add support for aggregation pushdown in Druid connector

Open conor-mcavoy opened this issue 6 months ago • 1 comments

Description

This change adds support for aggregation pushdown in the Druid connector for some common aggregate functions, namely min/max, avg, sum, count, and variations of these. Actually adding this support requires little more than boilerplate, as the standard "Implements*" AggregateFunctionRules can be used for most functions. One special case, count(distinct), requires a new property for best support.

New tests are created for these changes in TestDruidConnectorTest. They are largely similar to tests in BaseJdbcConnectorTest, but those tests can't always be reused for Druid due to Druid's lack of support for table create statements.

See https://github.com/trinodb/trino/pull/4313 for a previous attempt. This fixes https://github.com/trinodb/trino/issues/4109.

Additional context and related issues

For efficient queries, Trino should "push down" aggregation functions into Druid when running queries on the Trino-Druid connector. This means that, for a query like SELECT sum(<col>) FROM druid.druid_table, Druid will actually perform the sum itself. There are generic classes already available to take care of this for JDBC connectors within Trino, see for example the io.trino.plugin.base.aggregation package. This PR largely just makes use of those existing classes to indicate which functions can be pushed down. Druid does not have functions that correspond to Trino's functions for covariance, correlation, and regression, and Druid's standard deviation and variance functions require an extension, so these functions are not supported for pushdown.

One special case is aggregations containing distinct, such as count(distinct). Druid has a property useApproximateCountDistinct that controls its behavior when running such functions. Essentially, if useApproximateCountDistinct=true, Druid will use an approximate cardinality algorithm to calculate count(distinct), presumably similar to the approx_distinct() function in Trino, whereas if useApproximateCountDistinct=false Druid will use an exact count. Druid rejects queries like sum(distinct) when useApproximateCountDistinct=true. To handle all this, this PR has added a property druid.count-distinct-strategy, with options DEFAULT, APPROXIMATE, and EXACT. When DEFAULT is selected, Trino does not set the property, leaving it up to Druid (Druid clusters can enforce this on their own end). APPROXIMATE corresponds to useApproximateCountDistinct=true, and EXACT to useApproximateCountDistinct=false. Using EXACT will produce technically fully correct aggregations.

A battery of Druid-specific aggregation pushdown tests is included in this PR. These are similar to the tests in BaseJdbcConnectorTest, but the BaseJdbcConnectorTest tests cannot be directly inherited by Druid because Druid does not support table create statements.

The Druid Docker image version had to be updated from 0.18.0 due to this issue, which is fixed in https://github.com/apache/druid/pull/9959 and included in 0.19.0. Without this, queries like SELECT agg(*) FROM (SELECT * FROM <table> LIMIT N) would fail due to a bug in Druid's planner.

Release notes

( ) This is not user-visible or is docs only, and no release notes are required. ( ) Release notes are required. Please propose a release note for me. (x) Release notes are required, with the following suggested text:

## Druid
* Support aggregation pushdown for the Druid plugin. ({https://github.com/trinodb/trino/issues/4109}`4109`)

conor-mcavoy avatar Apr 29 '25 22:04 conor-mcavoy