Support aggregate functions in row pattern recognition #8738

kasiafi · 2021-07-31T13:34:08Z

No description provided.

kasiafi · 2021-07-31T15:37:48Z

Example output from PlanPrinter with aggregations:

trino:tiny> EXPLAIN SELECT running_labels, max_id
         ->         FROM (VALUES 1) t(id)
         ->         MATCH_RECOGNIZE (
         ->             MEASURES
         ->                      RUNNING array_agg(CLASSIFIER(A)) AS running_labels,
         ->                      FIRST(A.id) + FINAL max_by(id, MATCH_NUMBER()) AS max_id
         ->             PATTERN (A*)
         ->             DEFINE A AS avg(A.id) > 0.5
         ->         );
                                    Query Plan
-----------------------------------------------------------------------------------
 Fragment 0 [SINGLE]
     Output layout: [running_labels, max_id]
     Output partitioning: SINGLE []
     Stage Execution Strategy: UNGROUPED_EXECUTION
     Output[running_labels, max_id]
     │   Layout: [running_labels:array(varchar), max_id:integer]
     │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
     └─ PatterRecognition[]
        │   Layout: [running_labels:array(varchar), max_id:integer]
        │   running_labels := "array_agg"
        │       array_agg := RUNNING array_agg("classifier"){A}
        │   max_id := ("field_0" + "max_by")
        │       field_0 := field[RUNNING FIRST({A})]
        │       max_by := FINAL max_by("field", "match_number"){}
        │   ONE ROW PER MATCH
        │   AFTER MATCH SKIP PAST LAST ROW
        │   pattern[A{0, ∞}] (INITIAL)
        │   subsets[]
        │   A := ("avg" > CAST(DECIMAL '0.5' AS double))
        │       avg := RUNNING avg("expr"){A}
        └─ LocalExchange[SINGLE] ()
           │   Layout: [expr:bigint, field:integer]
           │   Estimates: {rows: 1 (14B), cpu: 19, memory: 0B, network: 0B}
           └─ Project[]
              │   Layout: [expr:bigint, field:integer]
              │   Estimates: {rows: 1 (14B), cpu: 19, memory: 0B, network: 0B}
              │   expr := CAST("field" AS bigint)
              └─ LocalExchange[ROUND_ROBIN] ()
                 │   Layout: [field:integer]
                 │   Estimates: {rows: 1 (5B), cpu: 5, memory: 0B, network: 0B}
                 └─ Values
                        Layout: [field:integer]
                        Estimates: {rows: 1 (5B), cpu: 0, memory: 0B, network: 0B}
                        (1)


(1 row)

kasiafi · 2021-08-30T15:53:06Z

@mosabua could you please review the docs part?

mosabua

Just had a look at the docs and suggested minor changes. Great job. .. and sorry for being a bit late.

docs/src/main/sphinx/sql/match-recognize.rst

kasiafi · 2021-10-05T19:35:46Z

@mosabua I addressed your comments.

core/trino-main/src/main/java/io/trino/operator/window/matcher/ArrayView.java

core/trino-parser/src/main/antlr4/io/trino/sql/parser/SqlBase.g4

core/trino-main/src/main/java/io/trino/sql/planner/RelationPlanner.java

core/trino-main/src/main/java/io/trino/sql/planner/planprinter/PlanPrinter.java

core/trino-main/src/main/java/io/trino/sql/planner/rowpattern/AggregateArgumentsRewriter.java

...ain/src/main/java/io/trino/sql/planner/rowpattern/ExpressionAndValuePointersEquivalence.java

core/trino-main/src/main/java/io/trino/sql/planner/rowpattern/LogicalIndexExtractor.java

martint · 2021-11-05T23:59:16Z

core/trino-main/src/main/java/io/trino/sql/planner/rowpattern/ValuePointer.java

+        @JsonSubTypes.Type(value = ScalarValuePointer.class, name = "scalar"),
+        @JsonSubTypes.Type(value = AggregationValuePointer.class, name = "aggregation"),
+})
+public abstract class ValuePointer


This could be an interface.

I tried that, but this way the ValuePointers didn't serialize.

...main/src/test/java/io/trino/sql/planner/assertions/PatternRecognitionExpressionRewriter.java

...main/java/io/trino/sql/planner/iterative/rule/PushDownProjectionsFromPatternRecognition.java

core/trino-main/src/main/java/io/trino/operator/window/pattern/ArgumentComputation.java

docs/src/main/sphinx/sql/match-recognize.rst

core/trino-main/src/main/java/io/trino/operator/window/pattern/MatchAggregation.java

core/trino-main/src/main/java/io/trino/operator/window/pattern/MeasureComputation.java

martint · 2021-11-12T20:16:52Z

core/trino-main/src/main/java/io/trino/operator/window/pattern/SetEvaluator.java

+    private int aggregated;
+
+    // length of the prefix of the match where all applicable positions were identified,
+    // and stored in `allPositions`. During the pattern matching phase it might exceed
+    // the `aggregated` prefix due to `getAllPositions()` calls.
+    private int evaluated;
+
+    // all identified applicable positions for the current match until the `evaluated` position, starting from 0
+    // this list is updated:
+    // - by `resolveNewPositions()`
+    // - by `getAllPositions()` (only during the pattern matching phase)
+    private final IntList allPositions;


It's unclear what aggregated vs evaluated vs allPositions represent. I guess part of my confusion is due to not understanding how this class fits in the overall picture. Can you elaborate a bit?

The SetEvaluator is part of the state of MatchAggregation.
Its main purpose is to provide new applicable positions(*) to be aggregated by the MatchAggregation, since the previous call.
The additional purpose is to keep a collection of all applicable positions to be used by ThreadEquivalence.

Accordingly to the two purposes, two pointers are used to mark how much input was processed:

aggregated indicates how much input was processed by the MatchAggregation,

evaluated indicates how much input was processed to find applicable positions.

evaluated is never behind aggregated, because if we run the aggregation on some portion of input, we must have resolved the applicable positions first.
evaluated might be ahead of aggregated, in the case when we resolved the positions, but didn't run the actual aggregation. That happens via the getAllPositions() call, when ThreadEquivalence asks for the set of positions and compares them, but doesn't run the actual aggregation.

allPositions are all the applicable positions up until the evaluated pointer

(*) applicable positions are the positions with certain labels, e.g. for avg(A.x) these will be the positions labeled with A in the current match, wrt the running or final semantics.

core/trino-main/src/main/java/io/trino/operator/window/pattern/FeedablePagesWindowIndex.java

core/trino-main/src/main/java/io/trino/sql/analyzer/ExpressionAnalyzer.java

core/trino-parser/src/main/java/io/trino/sql/tree/DereferenceExpression.java

core/trino-main/src/main/java/io/trino/operator/window/matcher/ThreadEquivalence.java

core/trino-main/src/main/java/io/trino/operator/window/pattern/MeasureComputation.java

martint · 2021-11-15T16:47:35Z

core/trino-main/src/main/java/io/trino/operator/window/pattern/MeasureComputation.java

@@ -175,6 +184,18 @@ public static Block compute(
        return work.getResult();
    }

+    public static Block[] precomputeNulls(List<PhysicalValueAccessor> expectedLayout)


This method doesn't really need to know about PhysicalValueAccessors. It just cares about the types. Also, this doesn't seem like the right place to hold it.

This method doesn't really need to know about PhysicalValueAccessors. It just cares about the types

It cares about the types, but it also has to skip aggregations, which can be present in the list.
It all happens at construction time, and PhysicalValueAccessors are one of the constructor parameters, so I think we can pass just the accessors list instead of extracting the types in the constructor and passing them to the helper method.

this doesn't seem like the right place to hold it

Why? What is the right place?

core/trino-parser/src/test/java/io/trino/sql/parser/TestSqlParser.java

docs/src/main/sphinx/sql/match-recognize.rst

martint · 2021-12-01T23:52:00Z

core/trino-main/src/main/java/io/trino/sql/planner/rowpattern/LogicalIndexExtractor.java

+                FunctionCall classifier = classifierCall.get();
+                if (!classifier.getArguments().isEmpty()) {
+                    IrLabel label = irLabel(((Identifier) getOnlyElement(classifier.getArguments())).getCanonicalValue());
+                    Set<IrLabel> labels = subsets.get(label);


Can a subset refer to other subsets? If so, does this guarantee that all the subset references are expanded all the way down to the leaves?

No, according to the spec, a subset can only refer to primary row pattern variables.

core/trino-main/src/main/java/io/trino/sql/planner/iterative/rule/ExpressionRewriteRuleSet.java

martint · 2021-12-02T00:25:21Z

core/trino-main/src/main/java/io/trino/operator/window/matcher/ThreadEquivalence.java

+    private static class AggregationIndexes
+    {
+        final boolean foundNoClassifierAggregations;
+        final List<List<Integer>> noClassifierAggregations;


Add an explanation of what this list and each of the contained lists represents. The signature of the field doesn't make that clear enough.

That is explained in a comment above the method classifyAggregations(), and also mentioned in the class javadoc, and commented in the method equivalent() where those structures are used.

martint · 2021-12-02T00:25:40Z

core/trino-main/src/main/java/io/trino/operator/window/matcher/ThreadEquivalence.java

+
+    private static class AggregationIndexes
+    {
+        final boolean foundNoClassifierAggregations;


This seems unnecessary. Why not just rely on noClassifierAggregations.isEmpty() or something along those lines?

noClassifierAggregations as well as classifierAggregations are List<List<>>.
They have an entry for every label, and this entry is a list of relevant aggregations.

So, this is not that easy to determine if the whole structure is empty. If I didn't keep the boolean, I would have to iterate over the top-level list to find out.

I added the boolean in answer to your comment: #8738 (comment) It allows to skip part of the computation at runtime.

martint · 2021-12-02T00:28:15Z

core/trino-main/src/main/java/io/trino/operator/window/matcher/ThreadEquivalence.java

+                    }
+                }
+            }
+            noClassifierAggregations.add(noClassifierAggregationIndexes.build());


What does it mean if noClassifierAggregationIndexes is empty? (i.e., none of the pointers hit the first branch). Same question for the other list below.

It means that for a given label, in the label's defining condition, there are no aggregations of this kind:

"no-classifier aggregations" is the kind of aggregations whose result does not depend on the actual labels, e.g. max(A.price)

"classifier aggregations" is the kind of aggregations whose result depends on the actual labels, e.g. array_agg(CLASSIFIER(U)), where U is a subset {A, B}

These two kinds of aggregations are explained in the comment above the method classifyAggregations.

This might look like arbitrary partitioning, but it is based on the observation that for some aggregations it is enough to compare the sets of positions, while other aggregations require also comparing the matched labels.

The overall purpose is to optimize ThreadEquivalence and make it more accurate by skipping unnecessary comparisons.

core/trino-main/src/main/java/io/trino/operator/window/pattern/PhysicalValueAccessor.java

martint · 2021-12-02T00:42:32Z

core/trino-main/src/main/java/io/trino/operator/window/pattern/MatchAggregationPointer.java

+ */
+package io.trino.operator.window.pattern;
+
+public class MatchAggregationPointer


What does this represent?

MatchAggregationPointer replaces AggregationValuePointer after the optimization phase is finished.
This is an accessor to aggregation at execution time. At this time, all MatchAggregations are gathered in an array, and MatchAggregationPointer contains an index to that array.

martint · 2021-12-02T22:29:43Z

core/trino-main/src/main/java/io/trino/operator/window/pattern/SetEvaluator.java

+        checkArgument(currentRow >= patternStart && currentRow < patternStart + matchedLabels.length(), "current row is out of bounds of the match");
+        checkState(aggregated <= evaluated && evaluated <= matchedLabels.length(), "SetEvaluator in inconsistent state");
+
+        IntList positions = new IntList(DEFAULT_CAPACITY);


Isn't this a subset of allPositions? Why not return an ArrayView over a portion of that, instead?

positions are relative to partition start, and allPositions are relative to match start.

core/trino-main/src/main/java/io/trino/operator/window/pattern/MeasureComputation.java

This rule identifies and pre-projects all eligible arguments of aggregations in pattern recognition expressions. The purpose of it is to limit the number of runtime-evaluated arguments.

After this change, the pattern recognition expressions are processed by rewrite-based optimizations such as SimplifyExpressions or Desugar... transformations. The change applies to: - top-level pattern recognition expressions - aggregation arguments in pattern recognition context

cla-bot bot added the cla-signed label Jul 31, 2021

kasiafi added the WIP label Jul 31, 2021

kasiafi force-pushed the 263PatternAggregations branch 6 times, most recently from 51a7fb5 to 1302493 Compare August 7, 2021 12:06

kasiafi force-pushed the 263PatternAggregations branch 4 times, most recently from a60097c to e5ad1db Compare August 27, 2021 13:05

kasiafi removed the WIP label Aug 27, 2021

kasiafi force-pushed the 263PatternAggregations branch from e5ad1db to c11e350 Compare August 27, 2021 13:47

kasiafi requested a review from martint August 27, 2021 16:03

kasiafi force-pushed the 263PatternAggregations branch from 01d5018 to 5ecdc01 Compare August 30, 2021 15:51

kasiafi mentioned this pull request Sep 3, 2021

Fix too eager thread pruning in row pattern matching #9109

Merged

mosabua requested changes Sep 17, 2021

View reviewed changes

kasiafi force-pushed the 263PatternAggregations branch 4 times, most recently from b7d0fd6 to eedc5fc Compare October 5, 2021 11:33

kasiafi force-pushed the 263PatternAggregations branch 2 times, most recently from 5b48467 to ea1611f Compare October 7, 2021 08:45

martint reviewed Oct 7, 2021

View reviewed changes

core/trino-main/src/main/java/io/trino/operator/window/matcher/ArrayView.java Outdated Show resolved Hide resolved

core/trino-parser/src/main/antlr4/io/trino/sql/parser/SqlBase.g4 Outdated Show resolved Hide resolved

kasiafi force-pushed the 263PatternAggregations branch 2 times, most recently from ae16032 to f6e5e94 Compare October 8, 2021 12:22

kasiafi force-pushed the 263PatternAggregations branch 2 times, most recently from 2ffbc87 to e4338f8 Compare October 11, 2021 11:11

kasiafi requested a review from martint October 11, 2021 13:40

martint reviewed Nov 8, 2021

View reviewed changes

findepi mentioned this pull request Nov 9, 2021

Aggregation and window refactor #9837

Merged

kasiafi force-pushed the 263PatternAggregations branch 2 times, most recently from 367f91f to 2cf1fcf Compare November 9, 2021 14:11

martint reviewed Nov 12, 2021

View reviewed changes

martint reviewed Nov 15, 2021

View reviewed changes

kasiafi force-pushed the 263PatternAggregations branch from 12b7513 to 2ad80f3 Compare November 16, 2021 11:35

kasiafi force-pushed the 263PatternAggregations branch from 2ad80f3 to d70a37f Compare November 26, 2021 19:26

martint reviewed Dec 2, 2021

View reviewed changes

kasiafi force-pushed the 263PatternAggregations branch 2 times, most recently from 777d4df to 656a27f Compare December 2, 2021 14:05

martint reviewed Dec 2, 2021

View reviewed changes

kasiafi force-pushed the 263PatternAggregations branch from 656a27f to 93f05e0 Compare December 3, 2021 10:40

martint approved these changes Dec 3, 2021

View reviewed changes

kasiafi added 6 commits December 3, 2021 19:02

Support aggregations in pattern recognition expressions

3a2c75b

Test optimizer rules with aggregations

010900d

Add rule to push projections from PatternRecognitionNode

655d85e

This rule identifies and pre-projects all eligible arguments of aggregations in pattern recognition expressions. The purpose of it is to limit the number of runtime-evaluated arguments.

Add distributed tests for aggregations in row pattern matching

96ac4b0

Document aggregations in row pattern matching

42c96aa

kasiafi force-pushed the 263PatternAggregations branch from 93f05e0 to 42c96aa Compare December 3, 2021 18:02

kasiafi merged commit f2b0a4c into trinodb:master Dec 3, 2021

kasiafi mentioned this pull request Dec 3, 2021

Release notes for 365 #9826

Closed

12 tasks

github-actions bot added this to the 365 milestone Dec 3, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support aggregate functions in row pattern recognition #8738

Support aggregate functions in row pattern recognition #8738

kasiafi commented Jul 31, 2021

kasiafi commented Jul 31, 2021

kasiafi commented Aug 30, 2021

mosabua left a comment •

edited

Loading

kasiafi commented Oct 5, 2021

martint Nov 5, 2021

kasiafi Nov 9, 2021

martint Nov 12, 2021

kasiafi Nov 14, 2021

martint Nov 15, 2021

kasiafi Nov 16, 2021

martint Dec 1, 2021

kasiafi Dec 2, 2021

martint Dec 2, 2021

kasiafi Dec 2, 2021 •

edited

Loading

martint Dec 2, 2021

kasiafi Dec 2, 2021

martint Dec 2, 2021

kasiafi Dec 2, 2021

martint Dec 2, 2021

kasiafi Dec 2, 2021

martint Dec 2, 2021

kasiafi Dec 3, 2021

Support aggregate functions in row pattern recognition #8738

Support aggregate functions in row pattern recognition #8738

Conversation

kasiafi commented Jul 31, 2021

kasiafi commented Jul 31, 2021

kasiafi commented Aug 30, 2021

mosabua left a comment • edited Loading

Choose a reason for hiding this comment

kasiafi commented Oct 5, 2021

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

kasiafi Dec 2, 2021 • edited Loading

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

mosabua left a comment •

edited

Loading

kasiafi Dec 2, 2021 •

edited

Loading