[jira] [Created] (FLINK-13446) Row count sliding window outputs incorrectly in blink planner

["Hequn Cheng (JIRA)" <ji...@apache.org>]

Hequn Cheng created FLINK-13446:
-----------------------------------

             Summary: Row count sliding window outputs incorrectly in blink planner
                 Key: FLINK-13446
                 URL: https://issues.apache.org/jira/browse/FLINK-13446
             Project: Flink
          Issue Type: Bug
          Components: Table SQL / Runtime
    Affects Versions: 1.9.0
            Reporter: Hequn Cheng


For blink planner, the Row count sliding window outputs incorrectly. The window assigner assigns less window than what expected. This means the window outputs fewer data. The bug can be reproduced by the following test:
{code:java}
  @Test
  def testGroupWindowWithoutKeyInProjection(): Unit = {
    val data = List(
      (1L, 1, "Hi", 1, 1),
      (2L, 2, "Hello", 2, 2),
      (4L, 2, "Hello", 2, 2),
      (8L, 3, "Hello world", 3, 3),
      (16L, 3, "Hello world", 3, 3))

    val stream = failingDataSource(data)
    val table = stream.toTable(tEnv, 'long, 'int, 'string, 'int2, 'int3, 'proctime.proctime)

    val weightAvgFun = new WeightedAvg
    val countDistinct = new CountDistinct

    val windowedTable = table
      .window(Slide over 2.rows every 1.rows on 'proctime as 'w)
      .groupBy('w, 'int2, 'int3, 'string)
      .select(weightAvgFun('long, 'int), countDistinct('long))

    val sink = new TestingAppendSink
    windowedTable.toAppendStream[Row].addSink(sink)
    env.execute()

    val expected = Seq("12,2", "8,1", "2,1", "3,2", "1,1")
    assertEquals(expected.sorted, sink.getAppendResults.sorted)
  }
{code}
The expected output is Seq("12,2", "8,1", "2,1", "3,2", "1,1") while the actual output is Seq("12,2", "3,2")

To fix the problem, we can correct the assign logic in CountSlidingWindowAssigner.assignWindows.



--
This message was sent by Atlassian JIRA
(v7.6.14#76016)