Implement aggregation pushdown in Pinot #6069

elonazoulay · 2020-11-24T08:59:27Z

Resolves #4140

elonazoulay · 2020-11-24T09:00:01Z

Resolves #4140

presto-pinot/src/main/java/io/prestosql/pinot/PinotConfig.java

presto-pinot/src/main/java/io/prestosql/pinot/PinotMetadata.java

presto-pinot/src/main/java/io/prestosql/pinot/query/DynamicTableBuilder.java

presto-pinot/src/test/java/io/prestosql/pinot/PinotQueryRunner.java

presto-pinot/src/main/java/io/prestosql/pinot/PinotMetadata.java

findepi · 2021-01-27T13:03:16Z

@elonazoulay please rebase. See especially #6667 which changes JdbcTableHandle a bit and how agg pushdown is handled now.
This may ideally allow to remove or simplify PinotTableHandle and DynamicTable.

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/query/DynamicTableBuilder.java

elonazoulay · 2021-03-05T22:12:05Z

@findepi Sounds good! I will be removing the DynamicTableBuilder in an upcoming pull request.

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/query/DynamicTableBuilder.java

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotMetadata.java

plugin/trino-pinot/src/test/java/io/trino/plugin/pinot/TestMinimalFunctionality.java

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotConfig.java

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotSessionProperties.java

plugin/trino-pinot/src/test/java/io/trino/plugin/pinot/TestAggregationPushdown.java

plugin/trino-pinot/src/test/java/io/trino/plugin/pinot/TestMinimalFunctionality.java

findepi · 2021-04-19T20:36:41Z

plugin/trino-pinot/src/test/java/io/trino/plugin/pinot/TestMinimalFunctionality.java

+                .setCatalogSessionProperty("pinot", "enable_aggregation_pushdown", "true")
+                .build();
+        String noGroupingsQueryTemplate = "SELECT COUNT(*), SUM(%1$s), AVG(%1$s), MIN(%1$s), MAX(%1$s) FROM " + TOPIC_AND_TABLE;
+        assertQueryFails(format(noGroupingsQueryTemplate, "long_number"), "Segment query returned.*");


Why this is expected to fail? please add explanation.

(same below)

Without aggregation pushdown enabled, the connector will query the servers directly. This is a way to test the limit for segment queries. A limit is enforced to avoid pinot servers crashing when too many large selects are run.

plugin/trino-pinot/src/test/java/io/trino/plugin/pinot/TestPinotIntegrationSmokeTest.java

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/query/DynamicTableBuilder.java

findepi · 2021-04-21T11:11:20Z

As i understand, this currently depends on #7627.
(Thanks for separating the PRs!)

findepi · 2021-06-08T10:40:43Z

@elonazoulay let me know whet this is ready for review. Also. it seems that

Fix mixed case column handling
Fix filter pushdown for segment queries
Guarantee limit for broker queries
Implement aggregation pushdown in Pinot

-- some commits could be run in separate PRs, which could perhaps reduce review overhead?

elonazoulay · 2021-06-08T20:22:08Z

@findepi - this is ready, but to make it easier I will break it up into separate PR's, thanks!

plugin/trino-pinot/src/test/java/io/trino/plugin/pinot/PinotQueryRunner.java

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/client/PinotClient.java

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotMetadata.java

findepi · 2021-06-09T10:05:52Z

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotMetadata.java

+            if (inputExpression.isEmpty() || (isCountFunction(aggregate) && !isCountStarFunction(aggregate))) {
+                return Optional.empty();
+            }
+            // Distinct aggregations other than distinctcounthll are not supported
+            if (aggregate.isDistinct()) {
+                return Optional.empty();
+            }


This should be handled for individual functions separately. What would it take to use AggregateFunctionRewriter here?

Also, checking isDistinct is not enough.
See io.trino.plugin.jdbc.expression.AggregateFunctionPatterns#basicAggregation which matches only "basic" aggregation functions, without features that are both: easy to forget about and hard to implement in the remote system.

findepi · 2021-06-09T10:11:22Z

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotMetadata.java

+                        .map(PinotColumnHandle.class::cast)
+                        .map(PinotColumnHandle::getColumnName)
+                        .collect(toImmutableList()),
+                tableHandle.getQuery().flatMap(DynamicTable::getFilter),


in DynamicTable what are the semantics of filters, aggregations, sorting, etc?
Which is applied first, which is second, etc.?

Current ordering of fields suggest that aggregation comes before filter, so this line looks incorrect -- you are transposing pre-existing filter with an aggregation.

If this is "just" a matter of field ordering in DynamicTable, please update it and add commentary like here:

trino/presto-base-jdbc/src/main/java/io/prestosql/plugin/jdbc/JdbcTableHandle.java

Lines 47 to 56 in b52bcd7

private final TupleDomain<ColumnHandle> constraint;

// semantically aggregation is applied after constraint

private final Optional<List<List<JdbcColumnHandle>>> groupingSets;

// semantically limit is applied after aggregation

private final OptionalLong limit;

// columns of the relation described by this handle, after projections, aggregations, etc.

private final Optional<List<JdbcColumnHandle>> columns;

When doing that, sure the ordering of fields matches that of constructor arguments, they are equally important!

Removed AggregationExpression and now the connector returns the exact pinot data type for aggregations. Also implemented approx_distinct pushdown.

hashhar

@elonazoulay I resolved the addressed comments. I think I have an alternative proposal that would work with #8578 and not require the copying. PTAL if what I'm saying makes sense and is feasible to implement.

There's another option to remove count_distinct pushdown (which is the only rule which needs the grouping columns in rewrite context) from this PR so that it works with #8578 and address count_distinct in a follow-up.

hashhar · 2021-08-11T19:41:54Z

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotMetadata.java

+            projections.add(new Variable(pinotColumnHandle.getColumnName(), pinotColumnHandle.getDataType()));
+            resultAssignments.add(new Assignment(pinotColumnHandle.getColumnName(), pinotColumnHandle, pinotColumnHandle.getDataType()));
+        }
+        List<String> groupingColumns = getOnlyElement(groupingSets).stream()


@elonazoulay IIUC the grouping columns in rewrite context are only needed to verify that the grouping columns are part of the SELECT clause. If so, can we instead do that validation here after the rewrite is done?

Yes, the connector will end up doing work that gets discarded but that seems cleaner because even if we add grouping columns to RewriteContext I don't see an easy way to populate them for other connectors so we won't be able to provide a sane impl. which doesn't return null for that method.

I think this is possible to do because here we have tableHandle available from which we can get the grouping columns in case of DynamicTable.

Yep, in a previous version of this pr it was done before the rewrite rules, I can add it back. Another idea: can RewriteContext.getGroupingColumns() return an Optional<List> as a default method?

Was there some specific reason to change from the old implementation? Maybe I'm missing something. If it's feasible to do I'd like that.

As for the RewriteContext changes I'll defer to @martint (and Piotr once he's back). IMO if we have alternatives which are cleaner then I'd like to avoid the change. IMO the validation that grouping columns are part of select isn't exclusive to just count_distinct and makes sense to do before any rewrite.

Sounds good. Would another option be to pass the table handle into the rewrite context? I can get the grouping columns there also.

@hashhar - I extracted the check for grouping columns and reverted the changes for AggregateFunctionRewriter and RewriteContext. What do you think about adding TableHandle to the RewriteContext instead of this approach?
Just in case I saved the previous version.

I like the newer version. We can revisit changing RewriteContext once we have other non-JDBC connectors that implement agg pushdown. We'll have better information at that time to decide the correct abstraction.

hashhar

Looks good to go from my perspective % a question about behaviour of applyCountDistinct.

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotMetadata.java

hashhar · 2021-08-18T12:20:03Z

@elonazoulay I think you can rebase on top of #8578 and remove the copied classes where possible.
If it all builds and passes CI, I'll proceed with merging #8578 following which we can merge this change too.

elonazoulay · 2021-08-19T00:23:40Z

@elonazoulay I think you can rebase on top of #8578 and remove the copied classes where possible.
If it all builds and passes CI, I'll proceed with merging #8578 following which we can merge this change too.

Will do, that's great news!!! Thanks @hashhar !

Reorder members and constructor parameters in order of pushdown application.

Add a config to set broker query limit. If the limit is set too high, pinot can oom: apache/pinot#7110

This is a prepatory commit for aggregation pushdown which sets values to null, otherwise pinot never returns null values.

hashhar

LGTM % question.

@findepi I think your comments are addressed. Can you PTAL? The major update is that this is now based on #8578 and the last commit (distinct count pushdown) is the one that has changed the most.

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotBrokerPageSource.java

elonazoulay · 2021-08-23T17:32:44Z

LGTM % question.

@findepi I think your comments are addressed. Can you PTAL? The major update is that this is now based on #8578 and the last commit (distinct count pushdown) is the one that has changed the most.
@hashhar - I added a comment in the code to explain this. Just so you have context: it is similar to the limit for segment queries behavior. Since you cannot know apriori how many rows a query returns, especially if it is pushed down by aggregation (turns into passthrough query), set the limit to one higher so the connector can check it and fail. This is similar to the behavior in the elastic search connector, It was suggested by @martint and has been implemented for segment queries already.

stale, comments addressed

hashhar · 2021-08-24T14:05:48Z

Thanks a lot for working on this @elonazoulay . ❤️

mattsfuller · 2021-08-24T15:21:57Z

Thank you!

cla-bot bot added the cla-signed label Nov 24, 2020

findepi changed the title ~~Implement aggregation pushdown~~ Implement aggregation pushdown in Pinot Nov 24, 2020

findepi added the enhancement New feature or request label Nov 24, 2020

elonazoulay force-pushed the agg_pushdown branch from 6781317 to cd77bf2 Compare November 24, 2020 20:15

martint self-requested a review November 24, 2020 21:10

elonazoulay force-pushed the agg_pushdown branch from cd77bf2 to 9e82d36 Compare November 25, 2020 03:55

martint reviewed Jan 26, 2021

View reviewed changes

findepi self-requested a review January 27, 2021 13:01

elonazoulay force-pushed the agg_pushdown branch from 9e82d36 to 4088407 Compare March 4, 2021 17:46

findepi reviewed Mar 4, 2021

View reviewed changes

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/query/DynamicTableBuilder.java Outdated Show resolved Hide resolved

elonazoulay force-pushed the agg_pushdown branch 2 times, most recently from 591c3b2 to 45ba425 Compare March 5, 2021 22:11

findepi reviewed Mar 6, 2021

View reviewed changes

elonazoulay force-pushed the agg_pushdown branch 3 times, most recently from 9dc0553 to 754b5ab Compare April 19, 2021 12:28

findepi reviewed Apr 19, 2021

View reviewed changes

elonazoulay force-pushed the agg_pushdown branch from 754b5ab to ec5f4af Compare April 20, 2021 23:49

elonazoulay force-pushed the agg_pushdown branch 2 times, most recently from 7adff3c to 3515f68 Compare June 3, 2021 17:58

xiangfu0 self-requested a review June 3, 2021 21:04

elonazoulay force-pushed the agg_pushdown branch 2 times, most recently from 0d90917 to e9189eb Compare June 7, 2021 23:42

findepi previously requested changes Jun 9, 2021

View reviewed changes

elonazoulay force-pushed the agg_pushdown branch from a352f5e to b3d6360 Compare August 10, 2021 21:21

hashhar reviewed Aug 11, 2021

View reviewed changes

elonazoulay force-pushed the agg_pushdown branch 2 times, most recently from e0a146a to fdff656 Compare August 17, 2021 09:16

hashhar reviewed Aug 18, 2021

View reviewed changes

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotMetadata.java Show resolved Hide resolved

elonazoulay force-pushed the agg_pushdown branch 6 times, most recently from 07baf9d to 53b968d Compare August 21, 2021 19:44

elonazoulay added 6 commits August 22, 2021 14:52

Reorder DynamicTable members

a30d875

Reorder members and constructor parameters in order of pushdown application.

Add filter to equals, hashCode and toString in DynamicTable

a0a92dd

Add limit for broker queries

e9e6ad8

Add a config to set broker query limit. If the limit is set too high, pinot can oom: apache/pinot#7110

Add null handling for decoders

f58979f

This is a prepatory commit for aggregation pushdown which sets values to null, otherwise pinot never returns null values.

Implement aggregation pushdown in Pinot

a1feed2

Implement Distinct Count Pushdown in Pinot

8723d41

elonazoulay force-pushed the agg_pushdown branch from 53b968d to 8723d41 Compare August 22, 2021 22:41

hashhar approved these changes Aug 23, 2021

View reviewed changes

plugin/trino-pinot/src/main/java/io/trino/plugin/pinot/PinotBrokerPageSource.java Show resolved Hide resolved

hashhar approved these changes Aug 24, 2021

View reviewed changes

hashhar merged commit c2c4f3f into trinodb:master Aug 24, 2021

hashhar added this to the 361 milestone Aug 24, 2021

hashhar mentioned this pull request Aug 24, 2021

Release notes for 361 #8732

Closed

10 tasks

mosabua mentioned this pull request Sep 10, 2021

Document aggregate pushdown in Pinot connector #9208

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implement aggregation pushdown in Pinot #6069

Implement aggregation pushdown in Pinot #6069

elonazoulay commented Nov 24, 2020 •

edited by findepi

Loading

elonazoulay commented Nov 24, 2020

findepi commented Jan 27, 2021

elonazoulay commented Mar 5, 2021

findepi Apr 19, 2021

elonazoulay Apr 20, 2021

findepi commented Apr 21, 2021

findepi commented Jun 8, 2021

elonazoulay commented Jun 8, 2021

findepi Jun 9, 2021

findepi Jun 9, 2021

elonazoulay Jul 4, 2021 •

edited

Loading

hashhar left a comment

hashhar Aug 11, 2021

elonazoulay Aug 12, 2021

hashhar Aug 12, 2021

elonazoulay Aug 12, 2021

elonazoulay Aug 17, 2021

hashhar Aug 18, 2021

hashhar left a comment

hashhar commented Aug 18, 2021

elonazoulay commented Aug 19, 2021

hashhar left a comment

elonazoulay commented Aug 23, 2021 •

edited

Loading

hashhar commented Aug 24, 2021

mattsfuller commented Aug 24, 2021

	private final TupleDomain<ColumnHandle> constraint;

	// semantically aggregation is applied after constraint
	private final Optional<List<List<JdbcColumnHandle>>> groupingSets;

	// semantically limit is applied after aggregation
	private final OptionalLong limit;

	// columns of the relation described by this handle, after projections, aggregations, etc.
	private final Optional<List<JdbcColumnHandle>> columns;

Implement aggregation pushdown in Pinot #6069

Implement aggregation pushdown in Pinot #6069

Conversation

elonazoulay commented Nov 24, 2020 • edited by findepi Loading

elonazoulay commented Nov 24, 2020

findepi commented Jan 27, 2021

elonazoulay commented Mar 5, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

findepi commented Apr 21, 2021

findepi commented Jun 8, 2021

elonazoulay commented Jun 8, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

elonazoulay Jul 4, 2021 • edited Loading

Choose a reason for hiding this comment

hashhar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hashhar left a comment

Choose a reason for hiding this comment

hashhar commented Aug 18, 2021

elonazoulay commented Aug 19, 2021

hashhar left a comment

Choose a reason for hiding this comment

elonazoulay commented Aug 23, 2021 • edited Loading

hashhar commented Aug 24, 2021

mattsfuller commented Aug 24, 2021

elonazoulay commented Nov 24, 2020 •

edited by findepi

Loading

elonazoulay Jul 4, 2021 •

edited

Loading

elonazoulay commented Aug 23, 2021 •

edited

Loading