[FLINK-15803][table] Update AggregateFunction and TableAggregateFunction to the new type system #13007

twalthr · 2020-07-28T12:26:39Z

What is the purpose of the change

This updates imperative aggregate functions AggregateFunction and TableAggregateFunction to the new type system with new type inference. The new stack is activated when using TableEnvironment.createTemporarySystemFunction or call(Class, ...) in Table API. Other locations (SQL DDL, Table API Scala implicits) will be updated once we support the concept of DataView as well. Currently, DataViews are not supported.

Brief change log

Update the base classes AggregateFunction and TableAggregateFunction
Enable them in createTemporarySystemFunction
Update the entire stack from SQL/Table API until AggregateUtil
Update some tests in different areas

Verifying this change

This change is already covered by existing tests.

The following functions and their tests have been updated:
CountNullNonNull, VarArgsAggFunction, Top3, WeightedAvgWithMergeAndReset

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): yes
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/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? JavaDocs

flinkbot · 2020-07-28T12:28:54Z

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.

Automated Checks

Last check on commit b597d2d (Tue Jul 28 12:28:54 UTC 2020)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

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

flinkbot · 2020-07-28T12:50:57Z

CI report:

c93cd7e Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

aljoscha

I left some comments inline. And here are some other comments:

I think [hotfix][table] Rename UserDefinedAggregateFunction to ImperativeAggregateFunction would be a better title for the first commit. And for future readers it might help if there were a short description, sth like “This is a pure rename. We need the rename because …”

[hotfix][table-api-java] Fix JavaDocs in TableEnvironment also drops @Experimental from a lot of functions. That’s a bit surprising when you consider the title and a description would also help.

In general, I think this is quite nice and also actually quite understandable if you take some time. 😅

...able/flink-table-api-java/src/main/java/org/apache/flink/table/typeutils/FieldInfoUtils.java

...java-bridge/src/main/java/org/apache/flink/table/api/bridge/java/StreamTableEnvironment.java

aljoscha · 2020-07-28T14:38:02Z

...link-table-common/src/main/java/org/apache/flink/table/functions/TableAggregateFunction.java

+ * the aggregated values until a final aggregation result is computed.
+ *
+ * <p>For each set of rows that needs to be aggregated, the runtime will create an empty accumulator
+ * by calling the {@link #createAccumulator()}. Subsequently, the {@code accumulate()} method of the


Suggested change

* by calling the {@link #createAccumulator()}. Subsequently, the {@code accumulate()} method of the

* by calling {@link #createAccumulator()}. Subsequently, the {@code accumulate()} method of the

...link-table-common/src/main/java/org/apache/flink/table/functions/TableAggregateFunction.java

aljoscha · 2020-07-28T14:41:28Z

...table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeChecks.java

 			return distinctType.getSourceType().accept(this);
 		}
+
+		@Override


Stumpled across this? What does defaultMethod actually do?

The defaultMethod handles all types which methods have not been overridden. See LogicalTypeDefaultVisitor.

...table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/DataTypeUtils.java

aljoscha · 2020-07-28T14:52:00Z

...blink/src/main/java/org/apache/flink/table/planner/catalog/FunctionCatalogOperatorTable.java

+		final FunctionKind kind = definition.getKind();

-		if (definition.getKind() == FunctionKind.SCALAR) {
+		if (kind == FunctionKind.TABLE) {


Not to ask timo, what exactly is this verifying?

I added a JavaDoc to this method. It basically verifies which kinds of functions are allowed to be returned from the catalog given the context information. SqlFunctionCategory can help for checking but is not always helpful. In any case non user-defined functions or FunctionKind.OTHER are definitely not supported at this location.

aljoscha · 2020-07-28T15:15:41Z

...er-blink/src/main/java/org/apache/flink/table/planner/expressions/SqlAggFunctionVisitor.java

+	private SqlAggFunction createSqlAggFunction(@Nullable FunctionIdentifier identifier, FunctionDefinition definition) {
+		// legacy
+		if (definition instanceof AggregateFunctionDefinition) {
+			final AggregateFunctionDefinition aggDef = (AggregateFunctionDefinition) definition;


That's the clean code nerd in me but I think all the if-else branches could become function calls, including the new code outside the if. I.e:

if (definition instanceof AggregateFunctionDefinition) { return translateLegacyAggregateFunction(...) } else if (...) { return translateLegacyTableAggregateFunction(...) } return translateFunction(...)

…egateFunction Renames this base class because users can easily mix up AggregateFunction and UserDefinedAggregateFunction. The new naming also prepares for the future DeclarativeAggregateFunction. ImperativeAggregateFunction has already been introduced as a concept in the Blink planner.

… methods in TableEnvironment

…printing

…ion to the new type system

aljoscha

Perfect! This looks good to merge now.

rmetzger added the review=description? label Jul 28, 2020

rmetzger added component=TableSQL/Planner component=TableSQL/API labels Jul 28, 2020

twalthr force-pushed the FLINK-15803_4 branch from b597d2d to 124c9dc Compare July 28, 2020 14:17

aljoscha reviewed Jul 28, 2020

View reviewed changes

twalthr added 7 commits July 29, 2020 09:38

[hotfix][table-api-java] Remove @experimental annotation for function…

05e24a6

… methods in TableEnvironment

[hotfix][table-api-java] Fix JavaDocs for TableEnvironment.fromValues

861c3f1

[hotfix][table] Remove deprecated AggregateFunction.requiresOver()

33d212a

[hotfix][table-common] Add CallExpression.getFunctionName for easier …

dd26620

…printing

[FLINK-15803][table] Update AggregateFunction and TableAggregateFunct…

dd74735

…ion to the new type system

Feedback addressed

c93cd7e

twalthr force-pushed the FLINK-15803_4 branch from 124c9dc to c93cd7e Compare July 29, 2020 08:12

aljoscha self-assigned this Jul 29, 2020

aljoscha self-requested a review July 29, 2020 08:18

aljoscha approved these changes Jul 29, 2020

View reviewed changes

twalthr closed this in b7e1574 Jul 29, 2020

	* by calling the {@link #createAccumulator()}. Subsequently, the {@code accumulate()} method of the
	* by calling {@link #createAccumulator()}. Subsequently, the {@code accumulate()} method of the

[FLINK-15803][table] Update AggregateFunction and TableAggregateFunction to the new type system #13007

[FLINK-15803][table] Update AggregateFunction and TableAggregateFunction to the new type system #13007

Uh oh!

Conversation

twalthr commented Jul 28, 2020

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 Jul 28, 2020

Automated Checks

Review Progress

Uh oh!

flinkbot commented Jul 28, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

aljoscha left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

aljoscha Jul 28, 2020

Choose a reason for hiding this comment

Uh oh!

Uh oh!

aljoscha Jul 28, 2020

Choose a reason for hiding this comment

Uh oh!

twalthr Jul 29, 2020

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

aljoscha Jul 28, 2020

Choose a reason for hiding this comment

Uh oh!

twalthr Jul 29, 2020

Choose a reason for hiding this comment

Uh oh!

aljoscha Jul 28, 2020

Choose a reason for hiding this comment

Uh oh!

aljoscha left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

flinkbot commented Jul 28, 2020 •

edited

Loading