[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16457821#comment-16457821
 ] 

ASF GitHub Bot commented on FLINK-8690:
---

Github user suez1224 commented on a diff in the pull request:

https://github.com/apache/flink/pull/5940#discussion_r184866337
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/CommonAggregate.scala
 ---
@@ -49,11 +49,16 @@ trait CommonAggregate {
 
 val aggs = namedAggregates.map(_.getKey)
 val aggStrings = aggs.map( a => s"${a.getAggregation}(${
-  if (a.getArgList.size() > 0) {
+  val prefix = if (a.isDistinct) {
--- End diff --

I think one line should be fine here, slightly more compact IMO.

val prefix = if (a.isDistinct) "DISTINCT " else ""


> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16457820#comment-16457820
 ] 

ASF GitHub Bot commented on FLINK-8690:
---

Github user suez1224 commented on a diff in the pull request:

https://github.com/apache/flink/pull/5940#discussion_r184866214
  
--- Diff: 
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalAggregate.scala
 ---
@@ -82,7 +82,7 @@ private class FlinkLogicalAggregateConverter
   case _ => true
 }
 
-!agg.containsDistinctCall() && supported
+supported
--- End diff --

I dont think we need this extra local variable here after your change.


> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[Rong Rong (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16457679#comment-16457679
 ] 

Rong Rong commented on FLINK-8690:
--

Thanks for the suggestion [~fhueske]. I think this is much cleaner than having 
2 logical plans since they are suppose to be "logical" LOL.

I created the PR based on your suggestions and it looks pretty darn good 
covering all the cases (and potentially it also clears the way to support 
distinct group aggregation w/o time window in the future). Please take a look 
when you have time :-)

> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16457676#comment-16457676
 ] 

ASF GitHub Bot commented on FLINK-8690:
---

GitHub user walterddr opened a pull request:

https://github.com/apache/flink/pull/5940

[FLINK-8690][table]Support DistinctAgg on DataStream

## What is the purpose of the change

* Allow FlinkLogicalAggregate to support distinct aggregations on 
DataStream, while keeping DataSet to decompose distinct aggs into GROUP BY 
follow by normal aggregates.

## Brief change log

  - Moved `AggregateExpandDistinctAggregatesRule.JOIN` to 
`DATASET_NORM_RULES`
  - Enabled `DataStreamGroupWindowAggregate` to support distinct agg while 
maintaining unsupported for `[DataStream/DataSet]GroupAggregate`.
  - Fixed typo in codegen for distinct aggregate when merge
  - Fixed a possible codegen test error for `UNION ALL`.

## Verifying this change

  - Unit-test are added for `DistinctAggregateTest`
  - Added ITCase for distinct group window agg

## 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): yes (codegen)
  - Anything that affects deployment or recovery: no
  - The S3 file system connector: no

## Documentation

  - Does this pull request introduce a new feature? yes
  - If yes, how is the feature documented? not yet, should we put in 
`Aggregate` section or `Group Window` section? inputs are highly appreciated. 
Also distinct over aggregate is bug-fixed in FLINK-8689 but not documented.


You can merge this pull request into a Git repository by running:

$ git pull https://github.com/walterddr/flink FLINK-8690

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/flink/pull/5940.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #5940


commit c517821d13341ae10b5d47acdbd0cc7d5bbe38b7
Author: Rong Rong 
Date:   2018-04-28T15:59:12Z

moving AggregateExpandDistinctAggregatesRule.JOIN to DATASET_NORM_RULES, 
and enabled distinct aggregate support for window aggregate over datastream




> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[Fabian Hueske (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16445682#comment-16445682
 ] 

Fabian Hueske commented on FLINK-8690:
--

Hi [~walterddr] and [~hequn8128],

Thanks for the good discussion!
I think there are two ways to solve this issue:

1. Separate logical optimization rule sets for batch and streaming as you did 
in your branch.
2. Move the {{AggregateExpandDistinctAggregatesRule.JOIN}} rule to the existing 
batch normalization rules, generalize {{FlinkLogicalAggregateConverter}} to 
accept {{DISTINCT}} aggregates, and guard for {{DISTINCT}} aggregations in the 
batch physical translation. 

Both approaches will work, but I think the second approach is a bit cleaner 
because doesn't rely on the cost pruning.

What do you think?


> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[Rong Rong (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16394806#comment-16394806
 ] 

Rong Rong commented on FLINK-8690:
--

Sweet. Thank you [~hequn8128]!
I am still a bit skeptical putting a "LOGICAL_RULESET" which is not entirely 
logical, I guess I will receive more feedback once the PR is out. :-P
Much appreciate the feedback and suggestion Hequn.

> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[Hequn Cheng (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16394404#comment-16394404
 ] 

Hequn Cheng commented on FLINK-8690:


Looks good for the rule changes :D

> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[Rong Rong (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16394379#comment-16394379
 ] 

Rong Rong commented on FLINK-8690:
--

Thanks [~hequn8128] for the valuable feedback :-)
I did some research and came up with a change somewhat close to what you 
expected. Please see:
https://github.com/walterddr/flink/commit/6c25d178e7acff57c598e70a8f4a2c0f3b74332e

I had a hard time trying to find a place to put the two different 
LogicalAggregateConverter(s), so I put them separately in 2 different 
LOGICAL_OPT_RULESET, for Stream and Batch each. 

Any comments or suggestions are highly appreciated :-)

> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[Hequn Cheng (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16394069#comment-16394069
 ] 

Hequn Cheng commented on FLINK-8690:


Rename current FlinkLogicalAggregateConverter to 
FlinkLogicalAggregateDataSetConverter will bring no change for dataset. Calcite 
can only find a plan generated by AggregateExpandDistinctAggregatesRule.

> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[Rong Rong (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16394068#comment-16394068
 ] 

Rong Rong commented on FLINK-8690:
--

You are right. Haven't thought about that. I will verify the plan cost. However 
I am less worry about the DataStream side as the Calcite rule will generate 2 
operations that requires state backend while using MapView will generate only 
1. My worry is on DataSet, as per my initial implementation of splitting 
FlinkLogicalAggregate into 2, both of them ignores the plan generated by 
{{AggregateExpandDistinctAggregatesRule}}

> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[Hequn Cheng (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16394064#comment-16394064
 ] 

Hequn Cheng commented on FLINK-8690:


No need to worry about the {{AggregateExpandDistinctAggregatesRule}}, the cost 
of plan generated by it will be bigger than the cost of the plan generated by 
{{FlinkLogicalAggregateDataStreamConverter}} and calcite will choose the best 
plan(i.e., the plan with the smallest cost).

> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[Rong Rong (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16394046#comment-16394046
 ] 

Rong Rong commented on FLINK-8690:
--

That should resolve our problem partially. The real reason why I introduced 
another node before logical plan is because 
**AggregateExpandDistinctAggregatesRule** is actually calcite specific and will 
apply globally regardless of whether it is on DataSet or DataStream and that's 
the rule we want to avoid applying in DataStream API.

Basically it converts 

{code:java}
/COUNT (DISTINCT f1)
{code}

Into 

{code:java}
COUNT (DIST_A)
FROM (SELECT A AS DIST_A GROUP BY A)
{code}

There's always a way to unite these two operators together using 
FlinkLogicalAggregateConverter. But it seems counter-intuitive first split then 
unite them together. 

> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[Hequn Cheng (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16394002#comment-16394002
 ] 

Hequn Cheng commented on FLINK-8690:


Hi, we don't have to create a new logical RelNode. As I said in the design doc, 
we can rename current FlinkLogicalAggregateConverter to 
FlinkLogicalAggregateDataSetConverter and add another 
FlinkLogicalAggregateDataStreamConverter which supports distinct aggregates. 
Both FlinkLogicalAggregateDataSetConverter and 
FlinkLogicalAggregateDataStreamConverter convert a LogicalAggregate to a 
FlinkLogicalAggregate. This may be more neat. What do you think?

> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[Rong Rong (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16390467#comment-16390467
 ] 

Rong Rong commented on FLINK-8690:
--

I created one of the initial (not at all perfect) support trying to create 
special treatment by adding a distinct rule on DataStreamNormRuleSet, so 
logical plan for DataStream directly coverts distinct logicalAggregate to 
FlinkLogicalAggregates, while DataSet still goes through 
AggregateExpandDistinctAggregatesRule.  Please see them here:
https://github.com/walterddr/flink/commit/ef9777cd8859180f38900393967deaabf39b7453
 

[~hequn8128] mentioned that we can override the match() function to achieve 
this. However, in order to separately generate logical plan for DataSet and 
DataStream, current solution I can think off is to add new rule with the 
NormRuleSet. Do you think this is the right path?

Any comments or suggestions are highly appreciate :-)

> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[Rong Rong (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16376938#comment-16376938
 ] 

Rong Rong commented on FLINK-8690:
--

Yes, changed the description

> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> Currently, *FlinkLogicalAggregate / FlinkLogicalWindowAggregate* does not 
> allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate*, to support unbounded distinct aggregation on 
> datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Commented] (FLINK-8690) Update logical rule set to generate FlinkLogicalAggregate explicitly allow distinct agg on DataStream

[Hequn Cheng (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-8690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16375920#comment-16375920
 ] 

Hequn Cheng commented on FLINK-8690:


Hi, currently, {{FlinkLogicalWindowAggregate}} doesn't support distinct 
aggregate either. Distinct aggregates are filtered in 
\{{LogicalWindowAggregateRule}}, so this issue should also take it into 
consideration.

> Update logical rule set to generate FlinkLogicalAggregate explicitly allow 
> distinct agg on DataStream
> -
>
> Key: FLINK-8690
> URL: https://issues.apache.org/jira/browse/FLINK-8690
> Project: Flink
>  Issue Type: Sub-task
>Reporter: Rong Rong
>Assignee: Rong Rong
>Priority: Major
>
> **Currently, *FlinkLogicalAggregate* does not allow distinct aggregate.
> We are proposing to reuse distinct aggregate codegen work designed for 
> *FlinkLogicalOverAggregate* / *FlinkLogicalWindowAggregate*, to support 
> unbounded distinct aggregation on datastream as well.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)