[FLINK-8690][table]Support group window distinct aggregation on DataStream #5940

walterddr · 2018-04-28T16:07:36Z

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.

suez1224 · 2018-04-28T22:03:39Z

...k-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


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

suez1224 · 2018-04-28T22:12:20Z

...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) {


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

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

fhueske

Thanks for the PR @walterddr!

I noticed that some of the changes are based on @haohui's PR #3764. I think it would be good to put your changes on top of his commit.
I left a few comments that should be easy to address.

FYI, I also created a JIRA issue to support distinct aggregates also in non-windowed grouping queries.
Given the runtime support, this should be very easy to achieve and basically just require a few tests for validation.

Thanks, Fabian

fhueske · 2018-05-03T22:43:15Z

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

@@ -641,7 +641,8 @@ class AggregationCodeGenerator(
               |      java.util.Map.Entry entry = (java.util.Map.Entry) mergeIt$i.next();
               |      Object k = entry.getKey();
               |      Long v = (Long) entry.getValue();
-               |      if (aDistinctAcc$i.add(k, v)) {
+               |      if (aDistinctAcc$i.add(


The key in the entry is a Row already

fhueske · 2018-05-04T08:50:47Z

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

@@ -641,7 +641,8 @@ class AggregationCodeGenerator(
               |      java.util.Map.Entry entry = (java.util.Map.Entry) mergeIt$i.next();
               |      Object k = entry.getKey();


Change this line to ${classOf[Row].getCanonicalName} k = (${classOf[Row].getCanonicalName}) entry.getKey();

fhueske · 2018-05-04T09:23:34Z

...flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/DistinctAggregateTest.scala

+  }
+
+  @Test
+  def testDistinctAggregateWithNonDistinctAndGrouping(): Unit = {


I think this test can be removed

fhueske · 2018-05-04T09:37:43Z

...braries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/SqlITCase.scala

+  def testDistinctAggWithMergeOnEventTimeSessionGroupWindow(): Unit = {
+    // create a watermark with 10ms offset to delay the window emission by 10ms to verify merge
+    val sessionWindowTestdata = List(
+      (1L, 1, "Hello"),


The test is not checking for DISTINCT semantics since all aggregated values are distinct. We could do COUNT(DISTINCT num) (int has to be renamed to num because its a SQL keyword).

To check the correct merge behavior, we need two windows which aggregate the same value that is than deduplicated in merge.

Some data like:

(1L, 2, "Hello"), // 1. Hello window (2L, 2, "Hello"), // 1. Hello window, deduped (8L, 2, "Hello"), // 2. Hello window, deduped during merge (10L, 3, "Hello"), // 2. Hello window, forwarded during merge (9L, 9, "Hello World"), // 1. Hello World window (4L, 1, "Hello"), // 1. Hello window, triggering merge of 1. and 2. Hello windows (16L, 16, "Hello")) // 3. Hello window (not merged)

fhueske · 2018-05-04T09:39:03Z

...braries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/SqlITCase.scala

+    tEnv.registerTable("MyTable", table)
+
+    val sqlQuery = "SELECT string, " +
+      "  COUNT(DISTINCT long) " +


It would be good to add the end timestamp of the windows (SESSION_END(rowtime, INTERVAL '0.005' SECOND)) to make it easier to eyeball the expected test results.

walterddr · 2018-05-04T15:44:01Z

Thanks @suez1224 @fhueske for the comments. I will change them accordingly.

Yes I copied a lot of test cases from @haohui's PR for my own testing. I can definitely put it on top given the runtime support is already merged in #5555. Procedure-wise question: should I rebase his commit then add my change on top, then attached to this PR? I am not sure if there's a clever way to both preserve the discussion in this thread and rebase on top of his change.

fhueske · 2018-05-04T16:03:54Z

Hmmm, good point. The discussion would be lost.
How about I put your changes on top of Haohui's changes before merging?

walterddr · 2018-05-04T16:07:43Z

LOL. I think I found a way:

Rebase [FLINK-6335] Parse DISTINCT over grouped windows in stream SQL. #3764 over to current master;
Rebase this branch to the rebased [FLINK-6335] Parse DISTINCT over grouped windows in stream SQL. #3764;
Make changes on top

:-)

…ES, and enabled distinct aggregate support for window aggregate over datastream

fhueske · 2018-05-07T15:43:33Z

Thanks for the update @walterddr.
The PR is good to merge.

fhueske · 2018-05-07T15:53:43Z

merging

…eaming tables. This closes apache#5940.

walterddr changed the title ~~[FLINK-8690][table]Support DistinctAgg on DataStream~~ [FLINK-8690][table]Support group window distinct aggregation on DataStream Apr 28, 2018

suez1224 reviewed Apr 28, 2018

View reviewed changes

fhueske reviewed May 4, 2018

View reviewed changes

Haohui Mai and others added 2 commits May 4, 2018 09:16

[FLINK-6335] Parse DISTINCT over grouped windows in stream SQL.

5de5eb1

moving AggregateExpandDistinctAggregatesRule.JOIN to DATASET_NORM_RUL…

6a17cd8

…ES, and enabled distinct aggregate support for window aggregate over datastream

walterddr force-pushed the FLINK-8690 branch from c517821 to 6a17cd8 Compare May 4, 2018 16:28

address diff comments, adding session end time for identifying merge

5599a94

fhueske mentioned this pull request May 7, 2018

[FLINK-6335] Parse DISTINCT over grouped windows in stream SQL. #3764

Closed

fhueske pushed a commit to fhueske/flink that referenced this pull request May 7, 2018

[FLINK-8690] [table] Add DISTINCT aggregates for group windows on str…

b4fb749

…eaming tables. This closes apache#5940.

asfgit closed this in 53610c3 May 7, 2018

sampathBhat pushed a commit to sampathBhat/flink that referenced this pull request Jul 26, 2018

[FLINK-8690] [table] Add DISTINCT aggregates for group windows on str…

dc5d1ae

…eaming tables. This closes apache#5940.

rmetzger added the component=API/TableSQL label Mar 18, 2019

flinkbot added component=TableSQL/API and removed component=API/TableSQL labels Mar 17, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-8690][table]Support group window distinct aggregation on DataStream #5940

[FLINK-8690][table]Support group window distinct aggregation on DataStream #5940

walterddr commented Apr 28, 2018

suez1224 Apr 28, 2018

suez1224 Apr 28, 2018

fhueske left a comment

fhueske May 3, 2018

fhueske May 4, 2018

fhueske May 4, 2018

fhueske May 4, 2018

fhueske May 4, 2018

fhueske May 4, 2018

walterddr commented May 4, 2018

fhueske commented May 4, 2018

walterddr commented May 4, 2018

fhueske commented May 7, 2018

fhueske commented May 7, 2018

		@@ -641,7 +641,8 @@ class AggregationCodeGenerator(
		\| java.util.Map.Entry entry = (java.util.Map.Entry) mergeIt$i.next();
		\| Object k = entry.getKey();

[FLINK-8690][table]Support group window distinct aggregation on DataStream #5940

[FLINK-8690][table]Support group window distinct aggregation on DataStream #5940

Conversation

walterddr commented Apr 28, 2018

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fhueske left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

walterddr commented May 4, 2018

fhueske commented May 4, 2018

walterddr commented May 4, 2018

fhueske commented May 7, 2018

fhueske commented May 7, 2018