[FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE #7177

dawidwys · 2018-11-26T15:38:54Z

What is the purpose of the change

This change allows using aggregates in MEASURES and DEFINE clauses of MATCH_RECOGNIZE in sql.

Based on #7189

Brief change log

Split AggregateUtil#transformToAggregateFunctions
enabled open/close methods in code generated PatternSelectFunction and IterativeCondition
generating code for aggregates in org.apache.flink.table.codegen.MatchCodeGenerator.AggBuilder
extended verification of aggregates' operands in org.apache.flink.table.codegen.PatternVariableFinder that checks for operands in udags
updated documentation

Still missing:

do not support distinct aggregates (depends on https://issues.apache.org/jira/browse/CALCITE-2707)

Verifying this change

Added tests:

org.apache.flink.table.match.MatchOperatorValidationTest#testAggregatesOnMultiplePatternVariablesNotSupported
org.apache.flink.table.runtime.stream.sql.MatchRecognizeITCase#testCepAggregates
org.apache.flink.table.match.MatchOperatorValidationTest#testAggregatesOnMultiplePatternVariablesNotSupportedInUDAGs

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

twalthr

I just reviewed the refactoring of the aggregate util. I like the refactoring but I added some comments to further improve the maintainability in the future.

twalthr · 2018-11-29T09:54:26Z

...ries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala

+    (aggFieldIndexes, aggregates, isDistinctAggs, accTypes, accSpecs)
+  }
+
+  private def createFlinkAggFunction(


Can you add a comment to every method in this class? This class is quite big every comment helps in understanding what is going on. For example, Converts Calcite's [[SqlAggFunction]] to a Flink UDF [[TableAggregationFunction]].

twalthr · 2018-11-29T10:01:15Z

...ries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala

+
+    (aggregate, accumulatorType, accSpecs)
+  }
+
  private def transformToAggregateFunctions(


We definitely need more documentation for this function. Can you create case classes for the return type of this function for proper naming because Array[Int] or Array[Boolean] is not very helpful.

twalthr · 2018-11-29T10:02:30Z

...ries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala


-    // create aggregate function instances by function type and aggregate field data type.


Restore this and the following line?

...ries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala

twalthr

Thank you @dawidwys. I added some comments. Maybe I also found a bug in MatchCodeGenerator.AggBuilder#generateAggAccess.

...nk-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/MatchRecognizeITCase.scala

twalthr · 2018-11-29T10:49:37Z

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

+  }
+}
+
+class PatternVariableFinder extends RexDefaultVisitor[Option[String]] {


Why not moving this to MatchUtil and perform the validation in the DataStreamRelNode instead?

This class has one additional function other than just validation. It also extracts the variable from aggregate RexCall, which I need for grouping aggregates by it. I could add the validation at an earlier stage, but effectively I would need to duplicate/reuse the same code. If you feel it's worth doing, I will update it accordingly.

I think it is ok to duplicate this class. Because we are just talking about 25 lines. We should try to separate validation and translation. I know we are also not doing it at other places but we should aim for it.

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

...es/flink-table/src/test/scala/org/apache/flink/table/match/MatchOperatorValidationTest.scala

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

twalthr

I added some feedback for the docs changes.

docs/dev/table/streaming/match_recognize.md

twalthr · 2018-11-29T13:47:25Z

docs/dev/table/streaming/match_recognize.md

+ACME       01-APR-11 10:00:04  01-APR-11 10:00:09     13.5
+{% endhighlight %}
+
+An important thing to have in mind is how aggregates behave in situation when no rows where mapped to certain pattern variable. Every aggregate, beside `COUNT` will produce `null` in those cases. `COUNT` on the other hand will


An important thing to have in mind is how aggregates behave in situation when no rows were mapped to a certain pattern variable. Every aggregate (except for `COUNT`) will produce `null` in those cases. `COUNT` on the other hand will produce 0.

But is this statement true? Doesn't the empty case depend on the aggregate function that you implemented? If you are using a UDF it must not be null right?

You are right, didn't think of UDAGs. Do you think it is worth adding such disclaimer at all. I've added it in the first place, because there was something similar in the SQL standard, but there are no UDAGs there.

flink-libraries/flink-table/src/main/java/org/apache/calcite/sql/validate/SqlValidatorImpl.java

dianfu

@dawidwys thanks a lot for the PR. Just left a few minor comments.

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

dianfu · 2018-11-29T12:47:47Z

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

+      }
+    }
+
+    private def generateAggAccess(call: RexCall) = {


What about renaming to generateAggregation?

It actually generates only the access. It does not generate computing the aggregate. That name might not be perfect, but actually I would like to differentiate it somehow.

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

dianfu · 2018-11-30T03:37:01Z

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

+    }
+
+    private def doGenerateAggAccess(call: RexCall) = {
+      val singleResultTerm = newName("result")


What about removing the prefix single?

I think it makes sense here, as it is a term for a single aggregation from a set of all calculated for a pattern variable. It makes even more sense as I moved a variable that holds a term name for result containing all aggregates into this function.

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

dawidwys · 2018-12-03T11:18:20Z

Hi @dianfu, @twalthr . Thank you for your review. I've updated the PR according to your comments.

There are still two main discussion points from @twalthr review:

Moving PatternVariableFinder, which also validates that there is only a single pattern variable in an aggregation, to e.g. DataStreamMatchRule. The main role of this class is to extract that single variable though. If I performed the validation at an earlier stage, I would have to traverse the expression with the same code in MatchCodeGenerator anyway, but I am ok with it, if you think it is worth it.
Using aggCall.toString as a key for reusing code for aggregation for a single variable that was used multiple times in an expression. I could not think of better way to identify that it is the same expression. Do you have an idea what could I use instead?

I would appreciate if you had another look.

twalthr

Thanks for the update @dawidwys. I added my last couple of comments. Can you prepare the PR for merging with proper commits? I will have a final look before merging.

twalthr · 2018-12-04T15:21:36Z

...ries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala

      namedAggregates.map(_.getKey),
      physicalInputRowType,
      needRetract,
      tableConfig)

-    val aggMapping = aggregates.indices.map(_ + groupings.length).toArray
+    val aggMapping = (0 until aggregateMetadata.getAggregateCallsCount).map(_ + groupings.length)


This occurs quite often (I counted 10 times). Maybe create helper function in AggregateMetadata that takes the grouping length such as getAdjustedMapping()?

twalthr · 2018-12-04T15:29:07Z

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

+  *
+  * }
+  * }}}
+  *


Awesome documentation. Very helpful.

twalthr · 2018-12-04T15:32:38Z

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

-          val exp: GeneratedExpression = generateAggAccess(call)
-          aggregates += call
-          reusableInputUnboxingExprs((call.toString, 0)) = exp
+          val exp: GeneratedExpression = doGenerateAggAccess(aggCall)


The naming is still not perfect here. What is the difference between generateAggAccess and doGenerateAggAccess? How about generateDeduplicatedAggAccess and generateAggAccess?

twalthr · 2018-12-04T15:36:39Z

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

           |  $singleResultNullTerm = false;
           |} else {
           |  $singleResultNullTerm = true;
           |  $singleResultTerm = $defaultValue;
           |}
           |""".stripMargin
+      } else {
+        j"""
+           |boolean $singleResultNullTerm = false;


This line can be dropped. The null term is not evaluated if nullCheck is disabled.

twalthr · 2018-12-04T15:42:08Z

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

        transformFuncName)

      generateAggCalculation(aggFunc, transformFuncName, inputTransform)
    }

+    private case class LogicalMatchAggCall(


nit: move all classes to the bottom to logically separate them from methods.

twalthr · 2018-12-04T15:45:59Z

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

+  }
+}
+
+class PatternVariableFinder extends RexDefaultVisitor[Option[String]] {


I think it is ok to duplicate this class. Because we are just talking about 25 lines. We should try to separate validation and translation. I know we are also not doing it at other places but we should aim for it.

twalthr · 2018-12-04T15:46:55Z

...libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala

@@ -691,12 +795,14 @@ class MatchCodeGenerator(

 class PatternVariableFinder extends RexDefaultVisitor[Option[String]] {

+  val ALL_PATTERN_VARIABLE = "*"


We could even move this constant definition to MatchUtil?

twalthr

Thank you @dawidwys. I had one tiny comment. Otherwise the PR looks good to me. +1 to merge

twalthr · 2018-12-14T12:38:53Z

...ries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala

-    val outputArity = groupings.length + aggregates.length
-
-    val aggregationStateType: RowTypeInfo = new RowTypeInfo(accTypes: _*)
+    val aggMapping = getAdjustedMapping(aggregateMetadata.getAggregateCallsCount, groupings.length)


Actually I thought that this function can be part of AggregateMetadata? Because you always apply it on the calls count.

twalthr · 2018-12-14T12:43:41Z

docs/dev/table/streaming/match_recognize.md

+
+<span class="label label-info">Note</span> Aggregations can be applied to expressions, but only if they reference a single pattern variable. Thus `SUM(A.price * A.tax)` is a valid one, but `AVG(A.price * B.tax)` is not.
+
+<span class="label label-danger">Attention</span> `DISTINCT` aggregations are not supported.


Can you open a follow up issue for this if it does not exist yet?

twalthr · 2018-12-14T15:17:30Z

docs/dev/table/streaming/match_recognize.md

+    PATTERN (A+ B)
+    DEFINE
+        A AS AVG(A.price) < 15
+    ) MR;


@dawidwys what is MR actually doing? Do we have to call SELECT MR.start_tstamp?

It is optional, but it is needed if we want to e.g. join the result of MR with other table. For example in a query:

SELECT MR.aid as mId, T.id as tId FROM MyTable MATCH_RECOGNIZE ( ORDER BY proctime MEASURES A.id AS aid, A.name as aName PATTERN (A) DEFINE A as A.id > 0 ) MR, MyTable T WHERE MR.aName = T.name

…tions

…companion object

…tions

…TCH_RECOGNIZE

dawidwys · 2018-12-16T12:03:00Z

Thank you very much @twalthr for your help and review! Will merge it once travis gives green (as I had to rebase)

dawidwys force-pushed the match_aggregates branch from 8e3215f to 134abbc Compare November 27, 2018 09:09

twalthr changed the title ~~[FLINK-7599] Support for aggregates in MATCH_RECOGNIZE~~ [FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE Nov 27, 2018

dawidwys force-pushed the match_aggregates branch 2 times, most recently from e208911 to 000c2ad Compare November 28, 2018 14:10

twalthr self-assigned this Nov 28, 2018

dawidwys force-pushed the match_aggregates branch from 000c2ad to 05710d0 Compare November 28, 2018 14:51

twalthr requested changes Nov 29, 2018

View reviewed changes

twalthr reviewed Nov 29, 2018

View reviewed changes

flink-libraries/flink-table/src/main/java/org/apache/calcite/sql/validate/SqlValidatorImpl.java Show resolved Hide resolved

dianfu reviewed Nov 30, 2018

View reviewed changes

dawidwys force-pushed the match_aggregates branch from 322a43a to e5dda66 Compare December 3, 2018 11:05

twalthr requested changes Dec 4, 2018

View reviewed changes

dawidwys force-pushed the match_aggregates branch 2 times, most recently from a29aa68 to 779df14 Compare December 5, 2018 15:38

twalthr approved these changes Dec 14, 2018

View reviewed changes

twalthr reviewed Dec 14, 2018

View reviewed changes

dawidwys added 6 commits December 16, 2018 12:52

[FLINK-7599][table] Refactored AggregateUtil#transformToAggregateFunc…

525b237

…tions

[hotfix][table] Move generate runner functions to MatchCodeGenerator …

96fa7cc

…companion object

[FLINK-7599][table, cep] Support for aggregates in MATCH_RECOGNIZE

258a708

[FLINK-7599][docs] Updated MATCH_RECOGNIZE documentation with aggrega…

19fcf65

…tions

[FLINK-7599][table] Throw exception on DISTINCT in aggregations in MA…

0ad1ace

…TCH_RECOGNIZE

[hotfix][docs] Unified indentation

1a92abc

dawidwys force-pushed the match_aggregates branch from 779df14 to 1a92abc Compare December 16, 2018 11:53

dawidwys merged commit 31665d6 into apache:master Dec 17, 2018

rmetzger added component=API/TableSQL component=Library/CEP labels Mar 18, 2019

flinkbot added the component=TableSQL/API label Mar 17, 2022

flinkbot removed the component=API/TableSQL label Mar 17, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE #7177

[FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE #7177

dawidwys commented Nov 26, 2018 •

edited

Loading

twalthr left a comment

twalthr Nov 29, 2018

twalthr Nov 29, 2018

twalthr Nov 29, 2018

twalthr left a comment

twalthr Nov 29, 2018

dawidwys Nov 29, 2018

twalthr Dec 4, 2018

twalthr left a comment

twalthr Nov 29, 2018

twalthr Nov 29, 2018

dawidwys Nov 29, 2018

dianfu left a comment

dianfu Nov 29, 2018

dawidwys Dec 3, 2018

dianfu Nov 30, 2018

dawidwys Dec 3, 2018

dawidwys commented Dec 3, 2018

twalthr left a comment

twalthr Dec 4, 2018

twalthr Dec 4, 2018

twalthr Dec 4, 2018

twalthr Dec 4, 2018

twalthr Dec 4, 2018

twalthr Dec 4, 2018

twalthr Dec 4, 2018

twalthr left a comment

twalthr Dec 14, 2018

twalthr Dec 14, 2018

twalthr Dec 14, 2018

dawidwys Dec 16, 2018

dawidwys commented Dec 16, 2018


		// create aggregate function instances by function type and aggregate field data type.

		@@ -691,12 +795,14 @@ class MatchCodeGenerator(

		class PatternVariableFinder extends RexDefaultVisitor[Option[String]] {

		val ALL_PATTERN_VARIABLE = "*"


		<span class="label label-info">Note</span> Aggregations can be applied to expressions, but only if they reference a single pattern variable. Thus `SUM(A.price * A.tax)` is a valid one, but `AVG(A.price * B.tax)` is not.

		<span class="label label-danger">Attention</span> `DISTINCT` aggregations are not supported.

[FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE #7177

[FLINK-7599] [table] Support for aggregates in MATCH_RECOGNIZE #7177

Conversation

dawidwys commented Nov 26, 2018 • edited Loading

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

twalthr 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

twalthr 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

twalthr 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

dianfu 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

dawidwys commented Dec 3, 2018

twalthr 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

Choose a reason for hiding this comment

twalthr 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

dawidwys commented Dec 16, 2018

dawidwys commented Nov 26, 2018 •

edited

Loading