[jira] [Closed] (FLINK-17553) Constant exists in group window key leads to error: Unsupported call: TUMBLE_END(TIMESTAMP(3) NOT NULL)

["Jark Wu (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-17553?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Jark Wu closed FLINK-17553.
---------------------------
    Resolution: Fixed

- master (1.12.0): c67109e99437d1dcca9fb6fca9f2741747c82a04
- 1.11.0: 3fe471ffaccadf674f310019db5ccdc9a3ae94e4

> Constant exists in group window key leads to  error:  Unsupported call: TUMBLE_END(TIMESTAMP(3) NOT NULL)
> ---------------------------------------------------------------------------------------------------------
>
>                 Key: FLINK-17553
>                 URL: https://issues.apache.org/jira/browse/FLINK-17553
>             Project: Flink
>          Issue Type: Bug
>          Components: Table SQL / Planner
>            Reporter: Terry Wang
>            Assignee: Terry Wang
>            Priority: Critical
>              Labels: pull-request-available
>             Fix For: 1.11.0
>
>         Attachments: temp.png
>
>
> Exception stack is as following:
>  !temp.png! 
> We can reproduce this problem by add following test in org.apache.flink.table.planner.runtime.stream.sql#TimeAttributeITCase
> {code:scala}
> // Some comments here
>   @Test
>   def testWindowAggregateOnConstantValue(): Unit = {
>     val ddl1 =
>       """
>         |CREATE TABLE src (
>         |  log_ts STRING,
>         |  ts TIMESTAMP(3),
>         |  a INT,
>         |  b DOUBLE,
>         |  rowtime AS CAST(log_ts AS TIMESTAMP(3)),
>         |  WATERMARK FOR rowtime AS rowtime - INTERVAL '0.001' SECOND
>         |) WITH (
>         |  'connector' = 'COLLECTION',
>         |  'is-bounded' = 'false'
>         |)
>       """.stripMargin
>     val ddl2 =
>       """
>         |CREATE TABLE dst (
>         |  ts TIMESTAMP(3),
>         |  a BIGINT,
>         |  b DOUBLE
>         |) WITH (
>         |  'connector.type' = 'filesystem',
>         |  'connector.path' = '/tmp/1',
>         |  'format.type' = 'csv'
>         |)
>       """.stripMargin
>     val query =
>       """
>         |INSERT INTO dst
>         |SELECT TUMBLE_END(rowtime, INTERVAL '0.003' SECOND), COUNT(ts), SUM(b)
>         |FROM src
>         | GROUP BY 'a', TUMBLE(rowtime, INTERVAL '0.003' SECOND)
>         |-- GROUP BY TUMBLE(rowtime, INTERVAL '0.003' SECOND)
>       """.stripMargin
>     tEnv.sqlUpdate(ddl1)
>     tEnv.sqlUpdate(ddl2)
>     tEnv.sqlUpdate(query)
>     println(tEnv.explain(true))
>   }
> {code}
> I spent lots of work digging into this bug, and found the problem may be caused by AggregateProjectPullUpConstantsRule which doesn't generate proper project items correctly.
> After I remove AggregateProjectPullUpConstantsRule from FlinkStreamRuleSets#DEFAULT_REWRITE_RULES, the test passed as expect.
> The problem is that WindowPropertiesRule can not match RelNodeTree after the transformation of AggregateProjectPullUpConstantsRule, we also can add ProjectMergeRule.INSTANCE in FlinkStreamRuleSets#DEFAULT_REWRITE_RULES after AggregateProjectPullUpConstantsRule.INSTANCE to solve this problem.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)