[FLINK-10976][table] Add Aggregate operator to Table API by hequn8128 · Pull Request #8311 · apache/flink

hequn8128 · 2019-04-29T12:04:44Z

What is the purpose of the change

This pull request add row-based aggregate function on Table API. Note: this pr based on the #7235. Thank you @dianfu for your excellent work.

Brief change log

Add row-based aggregate function on Table API.
Add tests.
Add docs.

Verifying this change

This change added tests and can be verified as follows:

Added tests in AggregateTest, AggregateITCase, AggregateStringExpressionTest, AggregateValidationTest

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-04-29T12:05:11Z

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 @rmetzger [PMC], @sunjincheng121 [committer]
✅ 2. There is [consensus] that the contribution should go into to Flink.
- Approved by @sunjincheng121 [committer]
❓ 3. Needs [attention] from.
✅ 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

rmetzger · 2019-04-29T14:33:26Z

@flinkbot approve description

dianfu · 2019-05-05T01:27:34Z

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

+	 *   AggregateFunction aggFunc = new MyAggregateFunction()
+	 *   tableEnv.registerFunction("aggFunc", aggFunc);
+	 *   table.aggregate("aggFunc(a, b) as (f0, f1, f2)")
+	 *     .select("key, f0, f1")


select("f0, f1")?

dianfu · 2019-05-05T01:32:53Z

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

 }
+
+/**
+  * The implementation of a [[AggregatedTable]] that has been grouped on a set of grouping keys.


a -> an
that has been performed on an aggregate function.

dianfu · 2019-05-05T01:35:15Z

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

+  * The implementation of a [[AggregatedTable]] that has been grouped on a set of grouping keys.
+  */
+class AggregatedTableImpl(
+  private[flink] val table: Table,


indentation

dianfu · 2019-05-05T01:37:29Z

...nk-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala

  }

+  def rowBasedAggregate(
+    groupingExpressions: JList[Expression],


indentation

dianfu · 2019-05-05T01:47:50Z

...nk-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala

+  : TableOperation = {
+    // resolve for java string case, i.e., turn LookupCallExpression to CallExpression.
+    val resolver = resolverFor(tableCatalog, functionCatalog, child).build
+    val resolvedAggregate = resolveSingleExpression(aggregate, resolver)


use resolveExpression?

dianfu · 2019-05-05T01:59:26Z

docs/dev/table/tableApi.md

+tableEnv.registerFunction("myAggFunc", myAggFunc);
+Table table = input
+  .groupBy("key")
+  .aggregate("myAggFunc(a, b) as (x, y, z)")


dianfu · 2019-05-05T02:04:57Z

docs/dev/table/tableApi.md

+    }
+  }
+
+  def resetAccumulator(acc: MyMinMaxAcc): Unit = {


The Scala example has resetAccumulator, while the Java example has not.

dianfu · 2019-05-05T02:18:37Z

.../test/scala/org/apache/flink/table/api/stream/table/validation/AggregateValidationTest.scala

  }
+
+  @Test(expected = classOf[ValidationException])
+  def testTableFunctionInSelection(): Unit = {


This test failed with the exception:
org.apache.flink.table.api.ValidationException: Given parameters of function 'func' do not match any signature.
Actual: (java.lang.Long)
Expected: (java.lang.String)

dianfu · 2019-05-05T02:21:11Z

.../test/scala/org/apache/flink/table/api/stream/table/validation/AggregateValidationTest.scala

+
+    table
+      .groupBy('a)
+      // must fail. Only AggregateFunction can be used in aggregate


The exception message is not friendly for users:
org.apache.flink.table.api.ValidationException: Invalid arguments [log(b), 'd'] for function: as

dianfu · 2019-05-05T02:24:12Z

.../test/scala/org/apache/flink/table/api/stream/table/validation/AggregateValidationTest.scala

+    util.tableEnv.registerFunction("func", new TableFunc0)
+    table
+      .groupBy('a)
+      // must fail. Only AggregateFunction can be used in aggregate


Only AggregateFunction -> Only one AggregateFunction

dianfu · 2019-05-05T02:29:38Z

@hequn8128 Thanks a lot for the PR. LGTM overall with just a few comments.

sunjincheng121 · 2019-05-05T08:03:12Z

@flinkbot approve-until architecture
I am glad to review this change after you rebase the code. :) @hequn8128
Best, Jincheng

hequn8128 · 2019-05-05T08:34:07Z

@sunjincheng121 @dianfu Thanks a lot for your review. I have updated the PR and rebased to the master.

Best, Hequn

dianfu · 2019-05-05T08:47:45Z

@hequn8128 Thanks for the update. LGTM. +1

sunjincheng121 · 2019-05-06T09:10:25Z

@flinkbot approve-until architecture

sunjincheng121

Thanks for the update @hequn8128
I only left some suggestions.
Best,
Jincheng

sunjincheng121 · 2019-05-06T09:51:42Z

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

+	/**
+	 * Performs an aggregate operation with an aggregate function. Use this before a selection
+	 * to perform the selection operation. The output will be flattened if the output type is a
+	 * composite type.


Use this before a selection to perform the selection operation. -> You have to close the "aggregate" with a select statement ?

sunjincheng121 · 2019-05-06T09:54:11Z

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


+	/**
+	 * Performs a global aggregate operation with an aggregate function. Use this before a selection
+	 * to perform the selection operation. The output will be flattened if the output type is a


Same as GroupedTable.

sunjincheng121 · 2019-05-06T09:58:17Z

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

+    aggregate(ExpressionParser.parseExpression(aggregateFunction))
+  }
+
+  override def aggregate(aggregateFunction: Expression): AggregatedTable = {


Can we consistency the name format for aggregateFunction and tableAggFunction?
I suggest using the complete word, i.e.: Agg->Aggregate? What do you think?

sunjincheng121 · 2019-05-06T10:05:02Z

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

+
+  override def select(fields: Expression*): Table = {
+    new TableImpl(tableImpl.tableEnv,
+      tableImpl.operationTreeBuilder.project(fields,


I think we can unify aggregate and flatAggregate implementations. i.e. both AggregatedTableImpl#select and FlatAggregateTableImpl#select can usingoperationTreeBuilder.project or xxTable.select(fields: _*). What do you think?

sunjincheng121 · 2019-05-06T10:12:38Z

...nk-table-planner/src/main/scala/org/apache/flink/table/operations/OperationTreeBuilder.scala

    aggregateOperationFactory.createAggregate(resolvedGroupings, resolvedAggregates, child)
  }

+  def rowBasedAggregate(


I think we can rowBasedAggregate -> aggregate and add some comments to illustrate the difference between the two aggregate methods, What do you think?

sunjincheng121 · 2019-05-06T10:24:10Z

...scala/org/apache/flink/table/api/stream/table/stringexpr/AggregateStringExpressionTest.scala

    verifyTableEquals(resJava, resScala)
  }
+
+  def testNonGroupedAggregate2(): Unit = {


testNonGroupedAggregate2->testRowBasedNonGroupedAggregate?

hequn8128 · 2019-05-06T13:57:51Z

@sunjincheng121 Thanks a lot for your review and suggestions. I have updated the PR according to your comments.

Best, Hequn

sunjincheng121 · 2019-05-06T23:28:12Z

LGTM. +1 to merged
@flinkbot approve all

This closes apache#8311.

rmetzger added review=description? component=API/TableSQL labels Apr 29, 2019

rmetzger added review=consensus? component=TableSQL/API and removed review=description? component=API/TableSQL labels Apr 29, 2019

dianfu reviewed May 5, 2019

View reviewed changes

dianfu added 2 commits May 5, 2019 15:58

[FLINK-10976] [table] Add support for aggregate to table API

f88a3bc

address review comments

710b006

sunjincheng121 self-requested a review May 5, 2019 08:03

sunjincheng121 self-assigned this May 5, 2019

hequn8128 added 2 commits May 5, 2019 16:06

[FLINK-10976][table] Add Aggregate operator to Table API

d4f33ab

address comments

09e3e73

hequn8128 force-pushed the flip29_aggregate2 branch from 2ccf79b to 09e3e73 Compare May 5, 2019 08:32

sunjincheng121 reviewed May 6, 2019

View reviewed changes

rmetzger added review=quality? and removed review=consensus? labels May 6, 2019

address comments

579cfdc

rmetzger added review=approved ✅ and removed review=quality? labels May 6, 2019

sunjincheng121 pushed a commit to sunjincheng121/flink that referenced this pull request May 7, 2019

[FLINK-10976] [table] Add support for aggregate to table API

fe6f366

This closes apache#8311.

asfgit closed this in 560b2bd May 7, 2019

Conversation

hequn8128 commented Apr 29, 2019

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

Uh oh!

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

Uh oh!

Review Progress

Uh oh!

rmetzger commented Apr 29, 2019

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!

dianfu commented May 5, 2019

Uh oh!

sunjincheng121 commented May 5, 2019

Uh oh!

hequn8128 commented May 5, 2019

Uh oh!

dianfu commented May 5, 2019

Uh oh!

sunjincheng121 commented May 6, 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!

hequn8128 commented May 6, 2019

Uh oh!

sunjincheng121 commented May 6, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

flinkbot commented Apr 29, 2019 •

edited

Loading