SPI and engine changes for aggregation pushdown #3697

Parth-Brahmbhatt · 2020-05-11T17:03:02Z

…connectors.

The commit also includes the top level optimizer rule and feature config to control
the experimental behavior.

tooptoop4 · 2020-05-11T20:01:44Z

@Parth-Brahmbhatt does https://github.com/prestosql/presto/blob/master/presto-base-jdbc/src/main/java/io/prestosql/plugin/jdbc/QueryBuilder.java#L104-L122 need to change too ?

Parth-Brahmbhatt · 2020-05-11T20:08:06Z

@Parth-Brahmbhatt does https://github.com/prestosql/presto/blob/master/presto-base-jdbc/src/main/java/io/prestosql/plugin/jdbc/QueryBuilder.java#L104-L122 need to change too ?

These are just the SPI changes and the optimizer rule that unblocks any connectors that can support agg pushdown. I will follow up with another PR where we would do this for jdbc connector so we can leverage it in druid connector.

martint · 2020-05-12T19:56:17Z

presto-main/src/main/java/io/prestosql/SystemSessionProperties.java

+                booleanProperty(
+                        ENABLE_AGG_PUSHDOWN,
+                        "Enable aggregate pushdown",
+                        featuresConfig.isEnableAggPushdown(),
+                        false));


I don't think we need a global toggle for this. Connectors can choose whether to support aggregation pushdown by not implementing the new API (or doing it selectively, possibly based on a connector-specific setting)

presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java

martint · 2020-05-12T20:00:57Z

presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java

+     * @param groupBy
+     * @return
+     */
+    default Optional<AggregatePushdownResult<ConnectorTableHandle>> applyAggregation(ConnectorSession session, ConnectorTableHandle handle, List<AggregateFunction> aggregates,


Why is groupBy a list of lists? For grouping sets?

yes. let me know if you think we should remove that right now and add it when we do review for grouping sets rule.

martint · 2020-05-12T22:39:31Z

presto-spi/src/main/java/io/prestosql/spi/function/AggregateFunction.java

+    private final List<ConnectorExpression> inputs;
+    private final Type outputType;
+    private final Optional<List<ConnectorExpression>> sortBy;
+    private final Optional<Map<ConnectorExpression, SortOrder>> sortOrder;


Ordering of the sorting clauses matters, so this should be a list of tuples containing an expression and a sort order.

I was using sortBy to preserve the ordering and sortOrder to lookup the actual order. I like just having a list of tuples so I have updated the code with a single list of entries.

martint · 2020-05-12T22:39:56Z

presto-spi/src/main/java/io/prestosql/spi/function/AggregateFunction.java

+    // Something to check with Martin what he thinks is the best course here, for now just exposing the fields that the connector can use to derive if
+    // this is an agg function it can handle and information they need to push it down.
+    private final String projectionName; // may be call it assignmentId? the purpose for this is to uniquely identify an agg but we are just using the projected expression's output symbol as the name here
+    private final String aggregateFunctionName;


Rename to functionName

martint · 2020-05-12T23:14:56Z

...rc/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregatesIntoTableScanRuleSet.java

+        }
+    }
+
+    private ImmutableList<AggregateFunction> toAggFunction(Map<Symbol, AggregationNode.Aggregation> aggregations)


Use interface type for collection: List<AggregateFunction>. Also, don't abbreviate names: toAggregateFunction would be better.

that was my bad, should have noticed the type difference. name changed.

martint · 2020-05-12T23:15:54Z

...rc/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregatesIntoTableScanRuleSet.java

+            AggregationNode.Aggregation aggregation = entry.getValue();
+            Signature signature = aggregation.getResolvedFunction().getSignature();
+            Type returnType = typeManager.getType(signature.getReturnType());
+            String aggFunctionName = signature.getName();


Suggested change

String aggFunctionName = signature.getName();

String name = signature.getName();

martint · 2020-05-12T23:22:33Z

presto-spi/src/main/java/io/prestosql/spi/function/AggregateFunction.java

+    private final boolean isDistinct;
+    // TODO It is unclear if filtering and mask from the aggregate needs to be forwarded to connectors so skipping them for now.
+
+    public AggregateFunction(String projectionName, String name, List<ConnectorExpression> inputs, Type outputType)


This constructor is unnecessary at this point. Let's rely one the one below until it becomes clear that we need it.

martint · 2020-05-12T23:23:41Z

presto-spi/src/main/java/io/prestosql/spi/function/AggregateFunction.java

+        this(projectionName, name, inputs, outputType, Optional.empty(), Optional.empty(), false);
+    }
+
+    public AggregateFunction(String projectionName, String aggregateFunctionName, List<ConnectorExpression> inputs, Type outputType, Optional<List<ConnectorExpression>> sortBy, Optional<Map<ConnectorExpression, SortOrder>> sortOrder, boolean isDistinct)


Move outputType before inputs. That way, if we ever have to add other constructors, it's easier to support varargs for the inputs if needed and keep things consistent.

martint · 2020-05-12T23:29:54Z

...rc/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregatesIntoTableScanRuleSet.java

+    public PushAggregatesIntoTableScanRuleSet(Metadata metadata)
+    {
+        this.metadata = metadata;
+        this.typeManager = new InternalTypeManager(metadata);


InternalTypeManager is not meant to be used by anything other that the metadata APIs. use Metadata.getType() in the calls below, instead.

findepi · 2020-05-13T13:03:02Z

presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java

+     * Attempt to push down the aggregates into the table.
+     * <p>
+     * Connectors can indicate whether they don't support aggregate pushdown or that the action had no effect
+     * by returning {@link Optional#empty()}. Connectors should expect this method to be called multiple times


Suggested change

* by returning {@link Optional#empty()}. Connectors should expect this method to be called multiple times

* by returning {@link Optional#empty()}. Connectors should expect this method may be called multiple times

-- connector cannot assume a method will be called multiple times (it may or may not)

findepi · 2020-05-13T13:03:30Z

presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java

+     * by returning {@link Optional#empty()}. Connectors should expect this method to be called multiple times
+     * during the optimization of a given query.
+     * <p>
+     * <b>Note</b>: it's critical for connectors to return Optional.empty() if calling this method has no effect for that


Optional.empty() -> {@code or {@link it

findepi · 2020-05-13T13:04:02Z

presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java

+     * to loop indefinitely.
+     * </p>
+     * <p>
+     * If the method returns a result, the list of assignment in the result will replace the existing assignments given


Suggested change

* If the method returns a result, the list of assignment in the result will replace the existing assignments given

* If the method returns a result, the list of assignments in the result will replace the existing assignments given

findepi · 2020-05-13T13:05:52Z

presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java

+     * to loop indefinitely.
+     * </p>
+     * <p>
+     * If the method returns a result, the list of assignment in the result will replace the existing assignments given


I understand the "given" part. It seems it's OK to return partial assignemnts (they will just be ignored).

(I think it's ok to drop this part of the sentence)

Dropped and modified it now that we don't want to really support this behavior.

findepi · 2020-05-13T13:07:50Z

presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java

+     *  - aggregate
+     *          v0 = agg_fn1(a)
+     *          v1 = agg_fn2(b, 2)
+     *          v2 = group_by c


what does v2 mean here?

those are just variable names, I could make it explicit and I tried to do that by adding {@link Variable} at some places in javadoc.

findepi · 2020-05-13T13:23:17Z

presto-spi/src/main/java/io/prestosql/spi/function/AggregateFunction.java

+    // Something to check with Martin what he thinks is the best course here, for now just exposing the fields that the connector can use to derive if
+    // this is an agg function it can handle and information they need to push it down.


Make sure to reword (or remove) before the PR is merged.

findepi · 2020-05-13T13:24:47Z

presto-spi/src/main/java/io/prestosql/spi/function/AggregateFunction.java

+        this(projectionName, name, inputs, outputType, Optional.empty(), Optional.empty(), false);
+    }
+
+    public AggregateFunction(String projectionName, String aggregateFunctionName, List<ConnectorExpression> inputs, Type outputType, Optional<List<ConnectorExpression>> sortBy, Optional<Map<ConnectorExpression, SortOrder>> sortOrder, boolean isDistinct)


nit: put each arg on separate line

findepi · 2020-05-13T13:25:47Z

presto-spi/src/main/java/io/prestosql/spi/function/AggregateFunction.java

+        this.inputs = requireNonNull(inputs, "inputs is null");
+        this.outputType = requireNonNull(outputType, "outputType is null");
+        this.sortBy = requireNonNull(sortBy, "sortBy is null");
+        this.sortOrder = requireNonNull(sortOrder, "sortOrder is null");


Make defensive copies fo Lists and Maps (eg unmodifiableList(new ArrayList<>(...)))

findepi · 2020-05-13T13:26:51Z

presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java

+
+        ConnectorSession connectorSession = session.toConnectorSession(catalogName);
+        return metadata.applyAggregation(connectorSession, table.getConnectorHandle(), aggregates, assignments, groupBy)
+                .map(result -> new AggregatePushdownResult<>(


Here we need to verify the return data is sane & safe.
See #3324

Added. Given the results are identical I extracted a common verification method.

findepi · 2020-05-13T13:27:44Z

presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java

@@ -1100,6 +1102,26 @@ public boolean usesLegacyTableLayouts(Session session, TableHandle table)
                        Optional.empty()));
    }

+    @Override
+    public Optional<AggregatePushdownResult<TableHandle>> applyAggregation(Session session, TableHandle table, List<AggregateFunction> aggregates, Map<String, ColumnHandle> assignments, Optional<List<List<ColumnHandle>>> groupBy)


put each arg on separate line

...rc/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregatesIntoTableScanRuleSet.java

martint · 2020-05-21T05:54:55Z

presto-main/src/main/java/io/prestosql/metadata/Metadata.java

@@ -352,6 +354,12 @@

    Optional<TableHandle> applySample(Session session, TableHandle table, SampleType sampleType, double sampleRatio);

+    Optional<AggregationPushdownResult<TableHandle>> applyAggregation(Session session,


Place the first argument on the next line

martint · 2020-05-21T06:01:32Z

presto-spi/src/main/java/io/prestosql/spi/connector/AggregationPushdownResult.java

+import static java.util.Collections.unmodifiableList;
+import static java.util.Objects.requireNonNull;
+
+public class AggregationPushdownResult<T>


Rename to AggregationApplicationResult for consistency with ProjectionApplicationResult

martint · 2020-05-21T06:07:51Z

presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java

+                .map(
+                        result -> {


Format as:

Suggested change

.map(

result -> {

.map(result -> {

martint · 2020-05-21T06:09:39Z

presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java

+                .flatMap(connectorExpression -> ConnectorExpressions.extractVariables(connectorExpression).stream())
+                .map(Variable::getName)
+                .filter(variableName -> !assignedVariables.contains(variableName))
+                .findAny().ifPresent(variableName -> { throw new IllegalStateException("Unbound variable: " + variableName); });


Format as:

Suggested change

.findAny().ifPresent(variableName -> { throw new IllegalStateException("Unbound variable: " + variableName); });

.findAny()

.ifPresent(variableName -> { throw new IllegalStateException("Unbound variable: " + variableName); });

martint · 2020-05-21T06:12:12Z

presto-main/src/test/java/io/prestosql/metadata/AbstractMockMetadata.java

@@ -447,6 +449,16 @@ public boolean usesLegacyTableLayouts(Session session, TableHandle table)
        return Optional.empty();
    }

+    @Override
+    public Optional<AggregationPushdownResult<TableHandle>> applyAggregation(Session session,


Put the first argument on the next line

martint · 2020-05-21T21:16:24Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

+            String name = signature.getName();
+            List<Expression> arguments = aggregation.getArguments();
+            List<Symbol> symbols = arguments.stream()
+                    .filter(expression -> expression instanceof SymbolReference)


This should unnecessary once you add the filter to the pattern as described above.

martint · 2020-05-21T21:17:49Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

+                    .map(SymbolsExtractor::extractUnique)
+                    .flatMap(Collection::stream)


This can only be a SymbolReference, so it'd be more appropriate to do:

Suggested change

.map(SymbolsExtractor::extractUnique)

.flatMap(Collection::stream)

.map(Symbol::from)

martint · 2020-05-21T21:18:06Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

+            if (arguments.size() != symbols.size()) {
+                return ImmutableList.of();
+            }


This should unnecessary once you add the filter to the pattern as described above.

martint · 2020-05-21T21:24:03Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

+                        assignmentBuilder.build()));
+    }
+
+    private List<AggregateFunction> toAggregateFunction(List<AggregationNode.Aggregation> aggregations)


Once you apply all the suggestions above and below, this method can simplify to:

private AggregateFunction toAggregateFunction(AggregationNode.Aggregation aggregation) { Signature signature = aggregation.getResolvedFunction().getSignature(); ImmutableList.Builder<ConnectorExpression> arguments = new ImmutableList.Builder<>(); for (int i = 0; i < aggregation.getArguments().size(); i++) { SymbolReference argument = (SymbolReference) aggregation.getArguments().get(i); arguments.add(new Variable(argument.getName(), metadata.getType(signature.getArgumentTypes().get(i)))); } Optional<OrderingScheme> orderingScheme = aggregation.getOrderingScheme(); Optional<List<Map.Entry<String, SortOrder>>> sortBy = orderingScheme.map(orderings -> orderings.getOrderBy().stream() .map(orderBy -> new AbstractMap.SimpleEntry<>( orderBy.getName(), SortOrder.valueOf(orderings.getOrderings().get(orderBy).name()))) .collect(toImmutableList())); return new AggregateFunction( signature.getName(), metadata.getType(signature.getReturnType()), arguments.build(), sortBy.orElse(ImmutableList.of()), aggregation.isDistinct(), aggregation.getFilter().map(Symbol::getName)); }

what about aggregation.getMask()? we should verify aggregation.getMask() is not present (yet).

Done, added check for masks in pattern.

martint · 2020-05-21T21:26:50Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

+        // The new scan outputs and assignments are union of existing and the ones returned by connector pushdown.
+        ImmutableList.Builder<Symbol> newScanOutputs = new ImmutableList.Builder<>();
+        newScanOutputs.addAll(tableScan.getOutputSymbols());


This is incorrect. The new scan outputs should be the symbols associated with grouping columns plus the symbols associated with aggregations.

comment removed.

findepi · 2020-05-25T11:28:40Z

Specially to support the future use case of sum(x+y) -> being converted by the connector to sum(x) + sum(y)

Can't reply inline. This is not correct rewrite, unless data does not contain NULLs.

presto> SELECT sum(x), sum(y), sum(x+y), sum(x)+sum(y)
     -> FROM (VALUES (1,1), (NULL, 2), (3, NULL)) t(x, y);
 _col0 | _col1 | _col2 | _col3
-------+-------+-------+-------
     4 |     3 |     2 |     7

findepi · 2020-05-25T10:37:23Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

+                assignments,
+                groupByColumns.isEmpty() ? ImmutableList.of() : ImmutableList.of(groupByColumns));
+
+        if (!aggregationPushdownResult.isPresent()) {


findepi · 2020-05-25T10:41:05Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

+                tableScan.getTable(),
+                aggregateFunctions,
+                assignments,
+                groupByColumns.isEmpty() ? ImmutableList.of() : ImmutableList.of(groupByColumns));


Why not ImmutableList.of(groupByColumns) always?

[ [] ] would correctly describe the fact that there is 1 grouping set and that it is a global aggregation.

findepi · 2020-05-25T10:51:58Z

presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java

+     * to loop indefinitely.
+     * </p>
+     * <p>
+     * If the method returns a result, the list of assignments in the result will be unioned with existing assignments. The projections


i think #3697 (comment) is still waiting for final resolution. @martint ?

(nit: if this stays, would be good to avoid "unioned" as intellij spellchecker highlights this)

unioned removed. I have assumed @martint 's last comment is what we want the API to do and changed both the contract and docs to reflect that. Let me know if there is a gap in my understanding vs what is expected.

findepi · 2020-05-25T10:53:23Z

presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java

+     * As an example, given the following plan:
+     *
+     * <pre>
+     *


Suggested change

*

findepi · 2020-05-25T10:54:24Z

presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java

+     * <pre>
+     * handle = TH0
+     * aggregates = [
+     *              { functionName=agg_fn1, outputType = <some presto type> inputs = [{@link Variable} a]} ,


Suggested change

* { functionName=agg_fn1, outputType = <some presto type> inputs = [{@link Variable} a]} ,

* { functionName=agg_fn1, outputType = «some presto type» inputs = [{@link Variable} a]} ,

otherwise you would need to use < >

findepi · 2020-05-25T11:49:02Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

+            variableMappings.put(assignment.getVariable(), symbol);
+        }
+
+        List<Expression> newPartialProjections = result.getProjections().stream()


what does "partial project" mean here? (can this be newProjections?)

findepi · 2020-05-25T11:50:01Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

+        groupingSets.getGroupingKeys().stream()
+                .forEach(groupBySymbol -> assignmentBuilder.put(groupBySymbol, groupBySymbol.toSymbolReference()));


Suggested change

groupingSets.getGroupingKeys().stream()

.forEach(groupBySymbol -> assignmentBuilder.put(groupBySymbol, groupBySymbol.toSymbolReference()));

groupingSets.getGroupingKeys()

.forEach(groupBySymbol -> assignmentBuilder.put(groupBySymbol, groupBySymbol.toSymbolReference()));

-- use Collection#forEach without going thru Stream

findepi · 2020-05-25T11:54:37Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

+        IntStream.range(0, aggregationOutputSymbols.size())
+                .forEach(index -> assignmentBuilder.put(aggregationOutputSymbols.get(index), newPartialProjections.get(index)));


add verify(aggregationOutputSymbols.size() == newPartialProjections.size()); just before this line as this is where you depend on this

no need for good exc msg, since it's in fact already validated in MetadataManager, so it's more for documenting assumptions/dependencies of the code here

findepi · 2020-05-25T11:58:16Z

presto-spi/src/main/java/io/prestosql/spi/connector/AggregateFunction.java

+    private final List<ConnectorExpression> inputs;
+    private final List<Map.Entry<String, SortOrder>> sortOrder;
+    private final boolean isDistinct;
+    private final Optional<String> filter;


Optional<Variable>

Not sure about this, Why would this need to be a variable given the filter on the agg side is just a Symbol?

Symbols (engine things) are mapped to Variables in ConnectorExpression.
Using Variable here would give semantics to this field -- connector would know it needs to check assignments to know which column handle this is.

BTW why isn't filter a Optional<ConnectorExpression> (or ConnectorExpression without Optional, using new Constant(true, BOOLEAN) instead of empty).
@Parth-Brahmbhatt @martint did you discuss this bit already?

We haven't discussed this, originally I did not have mask or filter as I couldn't quite understand how those were used. I will figure out how to get the corresponding connector expression from the symbol and update the PR today.

we haven't discussed this, I originally had no filters or masks as I did not completely understand what masks are for. @martint asked to add the filter in this PR in the first pass so the connectors have to explicitly decide to not handle them. For the current rule given it will only be matched when agg is followed by scan node the only possible place to look for filter symbol is tablescan assignment so I think it makes sense to just map it to Variable for now. I can't think of a scenario where we match this rule and it has a filter but the symbol is not in tablescan assignment so I have not added a check for that If you think we should add a defensive check I can add a condition.

I personally prefer Optional so I have kept it as is, let me know if you would like me to remove Optional.

findepi · 2020-05-25T12:04:27Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

+                        assignmentBuilder.build()));
+    }
+
+    private List<AggregateFunction> toAggregateFunction(List<AggregationNode.Aggregation> aggregations)


what about aggregation.getMask()? we should verify aggregation.getMask() is not present (yet).

findepi · 2020-06-09T10:29:53Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

@@ -119,7 +120,7 @@ public Result apply(AggregationNode node, Captures captures, Context context)
                .map(Entry::getKey)
                .collect(toImmutableList());

-        if (aggregateFunctions.isEmpty()) {
+        if (aggregateFunctions.isEmpty() && !aggregations.isEmpty()) {


Apparently aggregateFunctions.size() == aggregations.size(), so you don't need the && part.

I forgot that I no longer return empty list when an aggregate can not be transformed as we already check for only simple symbol references. This check makes no sense any more.

findepi · 2020-06-09T10:30:33Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

@@ -62,7 +62,8 @@
 {
    private static final Capture<TableScanNode> TABLE_SCAN = newCapture();

-    private static final Pattern<AggregationNode> PATTERN = aggregation()
+    private static final Pattern<AggregationNode> PATTERN =
+            aggregation()


update indent of the following lines

findepi · 2020-06-09T10:54:21Z

...main/src/main/java/io/prestosql/sql/planner/iterative/rule/PushAggregationIntoTableScan.java

+                .forEach(groupBySymbol -> {
+                    // if the connector returned a new mapping from oldColumnHandle to newColumnHandle, groupBy needs to point to
+                    // new columnHandle's symbol reference, otherwise it will continue pointing at oldColumnHandle.
+                    ColumnHandle originalColumnHandle = assignments.get(groupBySymbol);


Intellij warns me

'Map<String, ColumnHandle>' may not contain keys of type 'Symbol'

and indeed i get NPE in the next line, as originalColumnHandle comes out as null

updated the code so it now does assignments.get(groupBySymbol.getName()); so there is no warning anymore. I think it is a valid assumption though that the groupBy Symbol name must be present in original scan assignment. If you think that is a wrong assumption I can add a defensive check and return an empty plan if the defensive check fails.

martint

Just one minor comment.

Also:

Squash the review commits into the main one
Rename the main commit to "Add support for aggregation pushdown", since the current title is too long.

Once you apply those changes, I'll go ahead and merge it. Thanks for working on this!

martint · 2020-06-11T17:51:59Z

presto-spi/src/main/java/io/prestosql/spi/connector/AggregationApplicationResult.java

+        return assignments;
+    }
+
+    public Map<ColumnHandle, ColumnHandle> getGroupingSetMapping()


Let's call this groupingColumnMapping. Grouping sets are made of sets of columns, but this method is not specifically about the grouping sets themselves.

@Parth-Brahmbhatt @martint what is groupingColumnMapping used for ? Thanks.

Parth-Brahmbhatt · 2020-06-11T20:02:54Z

Just one minor comment.

Also:

Squash the review commits into the main one

Rename the main commit to "Add support for aggregation pushdown", since the current title is too long.

Once you apply those changes, I'll go ahead and merge it. Thanks for working on this!

Done

martint · 2020-06-11T21:54:10Z

CI failures are unrelated

martint · 2020-06-11T21:54:46Z

Merged. Thanks for working on this @Parth-Brahmbhatt!

findepi · 2020-06-12T09:40:18Z

🎉 🚀 thank you @Parth-Brahmbhatt!

cla-bot bot added the cla-signed label May 11, 2020

martint self-requested a review May 11, 2020 17:57

martint requested changes May 12, 2020

View reviewed changes

findepi reviewed May 13, 2020

View reviewed changes

Parth-Brahmbhatt force-pushed the agg-pushdown branch 2 times, most recently from b771c0e to 5e5678d Compare May 18, 2020 21:38

Parth-Brahmbhatt requested review from martint and findepi May 20, 2020 17:16

martint reviewed May 21, 2020

View reviewed changes

findepi reviewed May 25, 2020

View reviewed changes

Parth-Brahmbhatt force-pushed the agg-pushdown branch from 5e5678d to 1d225aa Compare May 28, 2020 21:13

findepi mentioned this pull request May 29, 2020

Aggregation pushdown in PostgreSQL, base-jdbc #3881

Merged

findepi changed the title ~~SPI changes to support nonpartial and complete aggregate pushdown to …~~ SPI and engine changes for aggregation pushdown May 29, 2020

Parth-Brahmbhatt force-pushed the agg-pushdown branch from 79db175 to fce51f4 Compare June 6, 2020 07:47

findepi mentioned this pull request Jun 6, 2020

Allow the connector to delegate filter statistics estimation to the engine #844

Closed

Parth-Brahmbhatt requested a review from martint June 8, 2020 18:06

Parth-Brahmbhatt force-pushed the agg-pushdown branch 2 times, most recently from 3325220 to b6a4432 Compare June 8, 2020 22:53

findepi reviewed Jun 9, 2020

View reviewed changes

Parth-Brahmbhatt requested a review from findepi June 10, 2020 02:55

findepi approved these changes Jun 10, 2020

View reviewed changes

findepi assigned martint Jun 10, 2020

martint reviewed Jun 11, 2020

View reviewed changes

Add support for aggregation pushdow

0a598fd

Parth-Brahmbhatt force-pushed the agg-pushdown branch from 7c31097 to 0a598fd Compare June 11, 2020 18:06

martint merged commit 4052f33 into trinodb:master Jun 11, 2020

This was referenced Jun 11, 2020

Release notes for 335 #3886

Closed

Allow connectors to participate in query optimization #18

Open

martint added this to the 335 milestone Jun 11, 2020

findepi mentioned this pull request Nov 4, 2020

Push DistinctLimit into TableScan #5756

Merged

findepi mentioned this pull request Jan 15, 2021

Aggregation pushdown #6613

Open

18 tasks

	String aggFunctionName = signature.getName();
	String name = signature.getName();

	* by returning {@link Optional#empty()}. Connectors should expect this method to be called multiple times
	* by returning {@link Optional#empty()}. Connectors should expect this method may be called multiple times

	* If the method returns a result, the list of assignment in the result will replace the existing assignments given
	* If the method returns a result, the list of assignments in the result will replace the existing assignments given

		// Something to check with Martin what he thinks is the best course here, for now just exposing the fields that the connector can use to derive if
		// this is an agg function it can handle and information they need to push it down.

		@@ -352,6 +354,12 @@

		Optional<TableHandle> applySample(Session session, TableHandle table, SampleType sampleType, double sampleRatio);

		Optional<AggregationPushdownResult<TableHandle>> applyAggregation(Session session,

	.findAny().ifPresent(variableName -> { throw new IllegalStateException("Unbound variable: " + variableName); });
	.findAny()
	.ifPresent(variableName -> { throw new IllegalStateException("Unbound variable: " + variableName); });

		.map(SymbolsExtractor::extractUnique)
		.flatMap(Collection::stream)

	.map(SymbolsExtractor::extractUnique)
	.flatMap(Collection::stream)
	.map(Symbol::from)

	* { functionName=agg_fn1, outputType = <some presto type> inputs = [{@link Variable} a]} ,
	* { functionName=agg_fn1, outputType = «some presto type» inputs = [{@link Variable} a]} ,

		groupingSets.getGroupingKeys().stream()
		.forEach(groupBySymbol -> assignmentBuilder.put(groupBySymbol, groupBySymbol.toSymbolReference()));

		IntStream.range(0, aggregationOutputSymbols.size())
		.forEach(index -> assignmentBuilder.put(aggregationOutputSymbols.get(index), newPartialProjections.get(index)));

SPI and engine changes for aggregation pushdown #3697

SPI and engine changes for aggregation pushdown #3697

Conversation

Parth-Brahmbhatt commented May 11, 2020

tooptoop4 commented May 11, 2020

Parth-Brahmbhatt commented May 11, 2020

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

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

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

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

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

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

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

findepi commented May 25, 2020

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment