[FLINK-12098] [table-planner-blink] Add support for generating optimized logical plan for simple group aggregate on stream #8110
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community Review Progress
Please see the Pull Request Review Guide for a full explanation of the review process. 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 commandsThe @flinkbot bot supports the following commands:
|
KurtYoung
reviewed
Apr 4, 2019
| import java.sql.{Date, Time, Timestamp} | ||
|
|
||
| /** The initial accumulator for Max with retraction aggregate function */ | ||
| class MaxWithRetractAccumulator[T] { |
There was a problem hiding this comment.
ok, should implements MinWithRetractAggFunctionTest as java in this PR?
| import java.sql.{Date, Time, Timestamp} | ||
|
|
||
| /** The initial accumulator for Min with retraction aggregate function */ | ||
| class MinWithRetractAccumulator[T] { |
| groupSize: Int, | ||
| needRetraction: Boolean, | ||
| aggs: Seq[AggregateCall]): Array[Boolean] = { | ||
|
|
| inputRowType: RelDataType, | ||
| groupSet: Array[Int], | ||
| typeFactory: FlinkTypeFactory): RelDataType = { | ||
|
|
| * Derives accumulators names from aggregate | ||
| */ | ||
| def inferAggAccumulatorNames(aggInfoList: AggregateInfoList): Array[String] = { | ||
|
|
KurtYoung
reviewed
Apr 8, 2019
| /** The initial accumulator for Max with retraction aggregate function. */ | ||
| public static class MaxWithRetractAccumulator<T> { | ||
| public T max; | ||
| public Long distinctCount; |
There was a problem hiding this comment.
just use mapSize? the name is a little bit confusing with distinct agg
| @Override | ||
| public MaxWithRetractAccumulator<T> createAccumulator() { | ||
| MaxWithRetractAccumulator<T> acc = new MaxWithRetractAccumulator<>(); | ||
| acc.max = getInitValue(); // max |
There was a problem hiding this comment.
Why not just use NULL to represent init value for all sub-classes
There was a problem hiding this comment.
I does not change any logic when porting to java.
for java, using NULL as init value is ok.
| hasMax = true; | ||
| } | ||
| } | ||
| if (!hasMax) { |
There was a problem hiding this comment.
The behavior of deleting expired data in the state backend is uncertain.
so mapSize data may exist while map data may have been deleted when both of them are expired.
I have added some comments.
f980267
to
bf9ee8c
Compare
…zed logical plan for simple group aggregate on stream
… java, and move agg functions to org.apache.flink.table.functions.aggfunctions
fix NPE in MaxWithRetractAggFunction and MinWithRetractAggFunction
…ctAggFunction from String to BinaryString
|
LGTM, +1 to merge |
HuangZhenQiu
pushed a commit
to HuangZhenQiu/flink
that referenced
this pull request
Apr 22, 2019
…ed logical plan for simple group aggregate on stream (apache#8110)
sunhaibotb
pushed a commit
to sunhaibotb/flink
that referenced
this pull request
May 8, 2019
…ed logical plan for simple group aggregate on stream (apache#8110)
tianchen92
pushed a commit
to tianchen92/flink
that referenced
this pull request
May 13, 2019
…ed logical plan for simple group aggregate on stream (apache#8110)
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
Add support for generating optimized logical plan for simple group aggregate on stream
Brief change log
Verifying this change
This change added tests and can be verified as follows:
Does this pull request potentially affect one of the following parts:
@Public(Evolving): ( no)Documentation