Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/5940#discussion_r186036107
--- Diff:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/SqlITCase.scala
---
@@ -52,6 +52,76 @@ class SqlITCase extends StreamingWithStateTestBase {
(8000L, "8", "Hello World"),
(20000L, "20", "Hello World"))
+ @Test
+ 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"),
--- End diff --
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)
```
---
