[FLINK-11051][table] Add streaming window FlatAggregate to Table API #8359

hequn8128 · 2019-05-07T12:44:10Z

What is the purpose of the change

Goal

This PR supports (streaming, window) flatAggregate on Table API.

Background

FLINK-10977 add streaming non-window flatAggregate on TableAPI. The behavior of table aggregates is most like GroupReduceFunction did, which computed for a group of elements, and output a group of elements. The TableAggregateFunction can be applied on Table/GroupedTable.

Motivation

For group window cases, it would be nice if we can also support flatAggregate. This means we can perform flatAggregate on Tumble/Slide/Session Windows. The API looks like:

table.window(Tumble.over("5.minutes").on("rowtime").as("w")) // define window
    .groupBy("a, w") // group by key and window
    .flatAggregate("tableAggFunc(b) as (x, y)")

Solution

This PR adds flatAggregate API on WindowGroupedTable and adds window table aggregate logical node for the plan optimization. For the runtime code, window table aggregate reused most of the code with window aggregate and non-window table aggregate.

The difference between window aggregate and window table aggregate is, for window table aggregate, the runtime aggregate function(org.apache.flink.table.runtime.aggregate.AggregateAggFunction) returns both accumulator and function to the window function and emit value in it while for window aggregate, the runtime aggregate function returns the final results to the window function. This is because we can't emit results in the runtime aggregate function for table aggregate as there is no collector to be used.

Brief change log

494d356 Add flatAggregate API on WindowGroupedTable. Add java and scala API. In this commit, the implementation has not been implemented.
9990a47 Resolve expression and build LogicalWindowTableAggregate. TableAggregate is added to make it different from Aggregate. The two kinds of RelNode contain different semantics. For Aggregate, it represents a relational operator that eliminates duplicates and computes totals while for TableAggregate, it can output 0 or more records for a group.
3b2e2bb Add Plan support for window table aggregate. Logical&Physical nodes and rules are added in the commit.
e1a3ed4 Add runtime support for window table aggregate. This commit reuses the code of window aggregate and adds support for window table aggregate.
031a6e7 Add docs for window table aggregate. Also rename Table Aggregate to FlatAggregate as it is more consistent with the API for users.

Verifying this change

This change added tests and can be verified as follows:

Added plan test(GroupWindowTableAggregateTest) to verify plan
Add IT case(GroupWindowTableAggregateITCase) to verify code gen and runtime logic
Add GropuWindowTableAggregateStringExpressionTest to test java.

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

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

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (docs)

flinkbot · 2019-05-07T12:46:57Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Review Progress

✅ 1. The [description] looks good.
- Approved by @sunjincheng121 [committer]
✅ 2. There is [consensus] that the contribution should go into to Flink.
- Approved by @sunjincheng121 [committer]
❗ 3. Needs [attention] from.
- Needs attention by @aljoscha [PMC], @dawidwys [committer], @twalthr [PMC]
✅ 4. The change fits into the overall [architecture].
- Approved by @sunjincheng121 [committer]
✅ 5. Overall code [quality] is good.
- Approved by @sunjincheng121 [committer]

Please see the Pull Request Review Guide for a full explanation of the review process.

Details

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

sunjincheng121 · 2019-05-09T07:47:50Z

@flinkbot approve consensus

sunjincheng121 · 2019-05-10T00:32:13Z

@flinkbot attention @aljoscha @twalthr @dawidwys

sunjincheng121

Thanks for your great Job! @hequn8128
I just finish 2 commits review and left a few comments.
please let me know what do you think about the suggestions.

Thanks,
Jincheng

sunjincheng121 · 2019-05-10T00:55:14Z

...-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/WindowGroupedTable.java

+	 * }
+	 * </pre>
+	 */
+	FlatAggregateTable flatAggregate(String tableAggregateFunction);


We should add the JavaDoc for WindowGroupedTable in FlatAggregateTable. e.g.:
A table that performs flatAggregate on a {@link Table}, a {@link GroupedTable} or a {@link WndowGroupedTable}.

One more comment, we should make the change: Similar to an SQL SELECT -> Similar to a SQL SELECT in FlatAggregateTable JavaDoc as well.

sunjincheng121 · 2019-05-10T01:15:48Z

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/tableImpl.scala

+  private val tableImpl = table.asInstanceOf[TableImpl]
+
+  override def select(fields: String): Table = {
+    select(ExpressionParser.parseExpressionList(fields).asScala: _*)


Can we remove .asScala?

sunjincheng121 · 2019-05-10T01:28:10Z

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/tableImpl.scala

+  }
+
+  override def select(fields: Expression*): Table = {
+    val expressionsWithResolvedCalls = fields.map(_.accept(tableImpl.callResolver)).asJava


asJava is very important for IDE check, but we do not have to add asJava for the functionality, is that correct? due to we already import the JavaConversions._.

You are right. I think we can remove it.

sunjincheng121 · 2019-05-10T01:29:39Z

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/tableImpl.scala

+      tableImpl.getUniqueAttributeSupplier)

-  override def flatAggregate(tableAggregateFunction: Expression): FlatAggregateTable = ???
+    if (!extracted.getAggregations.isEmpty) {


extracted.getAggregations.nonEmpty is better than !extracted.getAggregations.isEmpty, right?

sunjincheng121 · 2019-05-10T02:20:56Z

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/tableImpl.scala

-  override def flatAggregate(tableAggregateFunction: Expression): FlatAggregateTable = ???
+    if (!extracted.getAggregations.isEmpty) {
+      throw new ValidationException("The select statement after flatAggregate does not support " +
+        "aggregate functions.")


aggregate functions cannot be used in the select statement right after flatAggregate ?

sunjincheng121 · 2019-05-10T07:30:16Z

...table-planner/src/main/java/org/apache/flink/table/operations/AggregateOperationFactory.java

+			aggregates.stream().flatMap(expr -> {
+				if (isTableAggregate) {
+					return Stream.of(UserDefinedFunctionUtils.getFieldInfo(
+						((AggregateFunctionDefinition) ((CallExpression) expr).getFunctionDefinition())


I suggest defining a new method for extract user-defined table aggregate function expression names, so we can share the logic for both createWindowAggregate and createAggregate, or we can extract a method for both extractName and extractudtafNames, such as:

extractNames/extractTypes(expr){ if (isTableAggregate) { return Stream.of(extractUdtafName(expr)); } else { return Stream.of(extractName(expr).orElseGet(expr::toString)); } }

what do you think?

sunjincheng121 · 2019-05-10T08:21:16Z

...e/flink-table-planner/src/main/java/org/apache/flink/table/plan/TableOperationConverter.java

-				.map(expr -> expr.accept(aggregateVisitor))
-				.collect(toList());
+				.map(expr -> {
+					if (isTableAggregate) {


if(isTableAggFunctionCall(expr)){...} ?

sunjincheng121 · 2019-05-10T08:22:24Z

...e/flink-table-planner/src/main/java/org/apache/flink/table/plan/TableOperationConverter.java

 			GroupKey groupKey = relBuilder.groupKey(groupings);
 			LogicalWindow logicalWindow = toLogicalWindow(windowAggregate.getGroupWindow());
-			return flinkRelBuilder.aggregate(logicalWindow, groupKey, windowProperties, aggregations).build();
+			return flinkRelBuilder.aggregate(logicalWindow, groupKey, windowProperties, aggregations, isTableAggregate).build();


And put logic of isTableAggregate into flinkRelBuilder.aggregate?

BTW: It's better to Unified naming aggregate and windowAggregate， in AggregateOperationFactory and AggregateOperationFactory. What do you think?

AggregateOperationFactory#createAggregate AggregateOperationFactory#createWindowAggregate FlinkRelBuilder#aggregate(window: LogicalWindow) FlinkRelBuilder#aggregate(...)

Good idea! I think it would be nice if we can make these names consistent.

sunjincheng121 · 2019-05-10T08:38:53Z

...main/java/org/apache/flink/table/plan/rules/logical/ExtendedAggregateExtractProjectRule.java

-			LogicalAggregate logicalAggregate =
-				LogicalTableAggregate.getCorrespondingAggregate((LogicalTableAggregate) relNode);
-			RelNode newAggregate = performExtract(logicalAggregate, input, relBuilder);
+			Aggregate aggregate =


I think it's better to add a new performExtract for TableAggregate, in this way the code is more clearly.
We can not use the current preformExtract because of the Aggregate and TableAggregate if different logic node. What do you think?

sunjincheng121 · 2019-05-10T08:43:50Z

...able/flink-table-planner/src/main/scala/org/apache/flink/table/calcite/FlinkRelBuilder.scala

      groupKey: GroupKey,
      windowProperties: JList[PlannerExpression],
-      aggCalls: Iterable[AggCall])
+      aggCalls: Iterable[AggCall],


I think we can rename aggregate to windowAggregate due to AggregateOperationFactory#createAggregate and AggregateOperationFactory#createWindowAggregate. I am sorry for the aggregate PR not see this issue.

I think you make a good point here. I will make them consistent.

hequn8128 · 2019-05-10T14:21:59Z

@sunjincheng121 Thank you very much for your suggestions and meticulous review. I have addressed all your comments and updated the PR. Would be great if you can take another look. Thank you.

Best, Hequn

sunjincheng121

Thanks for the update! @hequn8128
only left a few suggestions!
Best, Jincheng

sunjincheng121 · 2019-05-13T00:13:34Z

...table-planner/src/main/java/org/apache/flink/table/operations/AggregateOperationFactory.java

+	 * Get result types for the aggregate or the table aggregate expression. For a table aggregate,
+	 * it may return multi result types when the composite return type is flattened.
+	 */
+	private Stream<TypeInformation<?>> getAggregateResultTypes(PlannerExpression plannerExpression) {


How about getAggregateResultTypes ->extractAggregateResultTypes, That's because for tableAgg we should flatten the tableAgg expression's type, furthermore, it's more consistence with extractAggregateNames, what do you think?

sunjincheng121 · 2019-05-13T00:20:37Z

...e/flink-table-planner/src/main/java/org/apache/flink/table/plan/TableOperationConverter.java

+				return aggregateExpression.accept(tableAggregateVisitor);
+			} else {
+				return aggregateExpression.accept(aggregateVisitor);
+			}


In general, we also need to check the isAggFunctionCall and throw an exception if false returned from isAggFunctionCall. but I agree it works for the current use case.

sunjincheng121 · 2019-05-13T00:43:59Z

...main/java/org/apache/flink/table/plan/rules/logical/ExtendedAggregateExtractProjectRule.java


-	private int getWindowTimeFieldIndex(LogicalWindowAggregate aggregate, RelNode input) {
-		LogicalWindow logicalWindow = aggregate.getWindow();
+	private int getWindowTimeFieldIndex(LogicalWindow logicalWindow, RelNode input) {


Why we should do this change? I can see one reason is for LogicalWindowTableAggregate, but I do not find out the change about LogicalWindowTableAggregate deal with the time field. Can explain more about this?

Only do some minor refactor. LogicalWindowTableAggregate.getCorrespondingAggregate returns a LogicalWindowAggregate and the time field will be processed same like LogicalWindowAggregate.

sunjincheng121 · 2019-05-13T00:50:27Z

.../scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregateBase.scala

+
+  override def explainTerms(pw: RelWriter): RelWriter = {
+    super.explainTerms(pw)
+      .itemIf("groupBy", groupingToString(inputSchema.relDataType, grouping), !grouping.isEmpty)


!grouping.isEmpty -> grouping.nonEmpty ? just a suggestion. because the implantion of noEmpty is !isEmpty

sunjincheng121 · 2019-05-13T00:55:47Z

.../scala/org/apache/flink/table/plan/rules/datastream/DataStreamGroupWindowAggregateRule.scala

      agg.getGroupSet.toArray)
-    }
  }
+}


Useless change?

sunjincheng121 · 2019-05-13T01:20:36Z

docs/dev/table/tableApi.md

+
+    <tr>
+      <td>
+        <strong>GroupBy Window TableAggregation</strong><br>


GroupBy Window TableAggregation -> FlatAggregate, and I think it again It's better to describe the document from the perspective of an operator? such as: `FlatAggregate and FlatAggregate(Windowed). What do you think?

So did the GroupBy TableAggregation section.

sunjincheng121 · 2019-05-13T01:59:33Z

...k-table-planner/src/main/scala/org/apache/flink/table/codegen/AggregationCodeGenerator.scala

  val constantFields = constantExprs.map(addReusableBoxedConstant)

+  val isTableAggregate = aggregates.length == 1 &&
+    aggregates(0).isInstanceOf[TableAggregateFunction[_, _]]


I think aggregates.length == 1 is the requirement of aggreate operator, and there a so many places using the aggregates(0).isInstanceOf[TableAggregateFunction[_, _]], so I suggest add a util method in UserDefinedFunctionUtils. such as: containsTableAggFunction(aggregationList: Seq[UserDefinedAggregateFunction]). which can be using in isTableAggregate. It's that make sense to you?

BTW: I think that as long as the TableAggregateFunciton is included, it is the TableAggregate operation, so there is no need to judge the size of Aggregates here. The judgment of size==1 should be the check done by the logic level(validate step).

Make sense, I think we don't need to check the size after validation.

sunjincheng121 · 2019-05-13T02:54:53Z

...e-planner/src/main/scala/org/apache/flink/table/runtime/ConcatKeyAndAggResultCollector.scala

+/**
+  * The collector is used to concat group key and table aggregate function output.
+  */
+class ConcatKeyAndAggResultCollector(val keyNum: Int) extends CRowWrappingCollector {


This class is only using in Table Aggregate right? if so It's better to ConcatKeyAndAggResultCollector -> TableAggregateCollector, and the keyNum -> groupKeySize or groupKeyCount, What do you think?

sunjincheng121 · 2019-05-13T02:55:38Z

...e-planner/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateAggFunction.scala

+  * Aggregate Function used for the aggregate or table aggregate operator in
+  * [[org.apache.flink.streaming.api.datastream.WindowedStream]].
  *
  * @param genAggregations Generated aggregate helper function


aggregate or table aggregate?

sunjincheng121 · 2019-05-13T03:00:21Z

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

    * @param namedAggregates    List of calls to aggregate functions and their output field names
    * @param inputRowType       Input row type
    * @param inputFieldTypes    Types of the physical input fields
+    * @param outputType         Output type of the aggregate node


Output type of the aggregate node -> Output type of the (table)aggregate node ？ please check all the JavaDoc，for aggregate and tableAggregate， and make the consistency of the descript.

- Add java and scala api. In this commit, the implementation has not been implemented.

…ggregate - `TableAggregate` is added to make it different from `Aggregate`. The two kinds of RelNode contain different semantics. For `Aggregate`, it represents a relational operator that eliminates duplicates and computes totals while for `TableAggregate`, it can output 0 or more records for a group.

- Logical&Physical nodes and rules are added in the commit.

- This commit reuses the code of window aggregate and add support for window table aggregate. The difference between window aggregate and window table aggregate is, for window table aggregate, the runtime aggregate function(org.apache.flink.table.runtime.aggregate.AggregateAggFunction) returns both accumulator and function to the window function and emit value in it while for window aggregate, the runtime aggregate function returns the final results to the window function. This is because we can't emit results in the runtime aggregate function for table aggregate as there is no collector to be used.

- Also rename `Table Aggregate` to `FlatAggregate` as it is more consistent with the api for users.

hequn8128 · 2019-05-14T13:09:38Z

@sunjincheng121 Hi, thanks a lot for your review and suggestions. I have updated the PR according to your comments. Could you kindly take another look. Thank you.

Best, Hequn

sunjincheng121 · 2019-05-17T01:59:09Z

LGTM. +1 to merged

sunjincheng121 · 2019-05-17T02:21:46Z

@flinkbot approve all

Brief change log: - Add flatAggregate API on WindowGroupedTable. Add java and scala API. In this commit, the implementation has not been implemented. - Resolve expression and build LogicalWindowTableAggregate. TableAggregate is added to make it different from Aggregate. The two kinds of RelNode contain different semantics. For Aggregate, it represents a relational operator that eliminates duplicates and computes totals while for TableAggregate, it can output 0 or more records for a group. - Add Plan support for window table aggregate. Logical&Physical nodes and rules are added in the commit. - Add runtime support for window table aggregate. This commit reuses the code of window aggregate and adds support for window table aggregate. - Add docs for window table aggregate. Also rename Table Aggregate to FlatAggregate as it is more consistent with the API for users. This closes apache#8359

rmetzger added review=description? component=TableSQL/API labels May 7, 2019

sunjincheng121 reviewed May 10, 2019

View reviewed changes

hequn8128 force-pushed the flip29_flatagg_window_split branch from ed284d5 to ef5014d Compare May 10, 2019 14:18

rmetzger requested review from aljoscha, dawidwys and twalthr May 11, 2019 11:43

hequn8128 force-pushed the flip29_flatagg_window_split branch from ef5014d to 5a1db3f Compare May 12, 2019 15:38

sunjincheng121 reviewed May 13, 2019

View reviewed changes

hequn8128 added 5 commits May 14, 2019 20:55

[FLINK-11051][table] Add flatAggregate api on WindowGroupedTable

5000809

- Add java and scala api. In this commit, the implementation has not been implemented.

[FLINK-11051][table] Add Plan support for window table aggregate

3528007

- Logical&Physical nodes and rules are added in the commit.

[FLINK-11051][table] Add docs for table aggregate

f1a68be

- Also rename `Table Aggregate` to `FlatAggregate` as it is more consistent with the api for users.

hequn8128 force-pushed the flip29_flatagg_window_split branch from 5a1db3f to f1a68be Compare May 14, 2019 12:59

rmetzger added review=approved ✅ and removed review=description? labels May 17, 2019

asfgit closed this in 142462f May 19, 2019

[FLINK-11051][table] Add streaming window FlatAggregate to Table API #8359

[FLINK-11051][table] Add streaming window FlatAggregate to Table API #8359

Uh oh!

Conversation

hequn8128 commented May 7, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Goal

Background

Motivation

Solution

Brief change log

Verifying this change

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

Documentation

Uh oh!

flinkbot commented May 7, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Review Progress

Uh oh!

sunjincheng121 commented May 9, 2019

Uh oh!

sunjincheng121 commented May 10, 2019

Uh oh!

sunjincheng121 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hequn8128 commented May 10, 2019

Uh oh!

sunjincheng121 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hequn8128 commented May 7, 2019 •

edited

Loading

flinkbot commented May 7, 2019 •

edited

Loading