[FLINK-13742] Fix code generation when aggregation contains both distinct aggregate with and without filter

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.9.1, 1.10.0
• Component/s: Table SQL / Planner
• Labels: pull-request-available
• Assignee: Shuo Cheng
• Reporter: Jark Wu
• Created: 16/Aug/19 03:49
• Updated: 27/Aug/19 04:17
• Resolved: 19/Aug/19 02:34

Description

The following test will fail when the aggregation contains COUNT(DISTINCT c) and COUNT(DISTINCT c) filter ....

	  @Test
  def testDistinctWithMultiFilter(): Unit = {
    val sqlQuery =
      "SELECT b, " +
        "  SUM(DISTINCT (a * 3)), " +
        "  COUNT(DISTINCT SUBSTRING(c FROM 1 FOR 2))," +
        "  COUNT(DISTINCT c)," +
        "  COUNT(DISTINCT c) filter (where MOD(a, 3) = 0)," +
        "  COUNT(DISTINCT c) filter (where MOD(a, 3) = 1) " +
        "FROM MyTable " +
        "GROUP BY b"
    val t = failingDataSource(StreamTestData.get3TupleData).toTable(tEnv).as('a, 'b, 'c)
    tEnv.registerTable("MyTable", t)
    val result = tEnv.sqlQuery(sqlQuery).toRetractStream[Row]
    val sink = new TestingRetractSink
    result.addSink(sink)
    env.execute()
    val expected = List(
      "1,3,1,1,0,1",
      "2,15,1,2,1,0",
      "3,45,3,3,1,1",
      "4,102,1,4,1,2",
      "5,195,1,5,2,1",
      "6,333,1,6,2,2")
    assertEquals(expected.sorted, sink.getRetractResults.sorted)
  }

Issue Links

• links to: GitHub Pull Request #9459