[jira] [Created] (FLINK-28504) Local-Global aggregation causes watermark alignment (table.exec.source.idle-timeout) of idle partition invalid

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

nyingping created FLINK-28504:
---------------------------------

             Summary: Local-Global aggregation causes watermark alignment (table.exec.source.idle-timeout) of idle partition invalid
                 Key: FLINK-28504
                 URL: https://issues.apache.org/jira/browse/FLINK-28504
             Project: Flink
          Issue Type: Bug
          Components: Table SQL / Runtime
    Affects Versions: 1.14.2
         Environment: flink 1.14.1

kafka 2.4
            Reporter: nyingping
         Attachments: image-2022-07-12-15-11-51-653.png, image-2022-07-12-15-19-29-950.png, image-2022-07-12-15-20-06-919.png

I have a window topN test task, the code is as follows

```

 Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 8082);
        StreamExecutionEnvironment streamExecutionEnvironment =
                StreamExecutionEnvironment.getExecutionEnvironment(configuration);
 
        StreamTableEnvironment st = StreamTableEnvironment.create(streamExecutionEnvironment);
 
        st.getConfig().getConfiguration().setString("table.exec.source.idle-timeout", "10s");

        st.executeSql(
                "CREATE TABLE test (\n"
                        + "  `key` STRING,\n"
                        + "  `time` TIMESTAMP(3),\n"
                        + "  `price` float,\n"
                        + "  WATERMARK FOR `time` AS `time` - INTERVAL '10' SECOND"
                        + ") WITH (\n"
                        + "  'connector' = 'kafka',\n"
                        + "  'topic' = 'test',\n"
                        + "  'properties.bootstrap.servers' = 'testlocal:9092',\n"
                        + "  'properties.group.id' = 'windowGroup',\n"
                        + "  'scan.startup.mode' = 'latest-offset',\n"
                        + "  'format' = 'json'\n"
                        + ")"

    String sqlWindowTopN =
                "select * from (" +
                "  select *, " +
                "   ROW_NUMBER() over (partition by window_start, window_end order by total desc ) as rownum " +
                "     from (" +
                "       select key,window_start,window_end,count(key) as `count`,sum(price) total from table (" +
                "           tumble(TABLE test, DESCRIPTOR(`time`), interval '1' minute)" +
                "        ) group by window_start, window_end, key" +
                "   )" +
                ") where rownum <= 3";
    st.executeSql(sqlWindowTopN).print();

```

 

Run and do not get result on long time after.

Watermark appears as follows on the UI

 

!image-2022-07-12-15-11-51-653.png!

I didn't set the parallelism manually, so it defaults to 12. The data source Kafka has only one partition, so there are free partitions. To align the watermarks for the entire task, I use the `table.exec. source. Idle-timeout` configuration.

 

As above show,I found that the system automatically split window-Topn SQL into local-global aggregation tasks. In the Local phase, watermark didn't work as well as I expected.

 

Manually setting the parallelism to 1 did what I expected.

`streamExecutionEnvironment.setParallelism(1);`

!image-2022-07-12-15-19-29-950.png!

 

I can also manually configure the system not to split into local-global phases. At this point, the `table.exec.source-idle-timeout ` configuration takes effect and the watermark is aligned.

`

st.getConfig().getConfiguration().setString("table.optimizer.agg-phase-strategy", "ONE_PHASE");

`

result:

!image-2022-07-12-15-20-06-919.png!

 

As mentioned above, I hope to be able to aggregate in two stages and align the watermarks at the same time.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)