[FLINK-21305] Fix Cumulative and Hopping window should accumulate late events belonging to the cleaned slice

• Type: Sub-task
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Fix Version/s: 1.13.0
• Component/s: Table SQL / Planner
• Labels: pull-request-available
• Assignee: Jark Wu
• Reporter: Jark Wu
• Created: 05/Feb/21 11:31
• Updated: 11/Oct/21 08:52
• Resolved: 14/Apr/21 01:58

Description

Currently, the CUMULATE window drops elements belonging to the cleaned slices. This will lead to more inaccurate result than without slicing optimization.

Hopping window has the similar problem, you can reproduce the problem with the following test in WindowAggregateJsonITCase.

    @Test
    public void testEventTimeHopWindow2() throws Exception {
        createTestValuesSinkTable("MySink", "name STRING", "cnt BIGINT");
        String jsonPlan =
                tableEnv.getJsonPlan(
                        "insert into MySink select\n"
                                + "  name,\n"
                                + "  COUNT(*)\n"
                                + "FROM TABLE(\n"
                                + "   HOP(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '5' SECOND, INTERVAL '10' SECOND))\n"
                                + "GROUP BY name, window_start, window_end");
        tableEnv.executeJsonPlan(jsonPlan).await();

        List<String> result = TestValuesTableFactory.getResults("MySink");
        assertResult(
                Arrays.asList(
                        "+I[a, 1]",
                        "+I[a, 4]",
                        "+I[a, 6]",
                        "+I[b, 1]",
                        "+I[b, 1]",
                        "+I[b, 1]",
                        "+I[b, 1]",
                        "+I[b, 2]",
                        "+I[b, 2]",
                        "+I[null, 1]",
                        "+I[null, 1]"),
                result);
    }

Issue Links

• is related to: FLINK-24501 Unexpected behavior of cumulate window aggregate for late event after recover from sp/cp
• links to: GitHub Pull Request #15495