[FLINK-24501] Unexpected behavior of cumulate window aggregate for late event after recover from sp/cp

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.15.0
• Component/s: Table SQL / Runtime
• Labels: pull-request-available
• Assignee: Jing Zhang
• Reporter: Jing Zhang
• Created: 11/Oct/21 08:50
• Updated: 10/Nov/21 07:16
• Resolved: 10/Nov/21 07:16

Description

Problem description

After recover from savepoint or checkpoint, unexpected behavior of cumulate window aggregate for late event may happened.

Bug analyze

Currently, for cumulate window aggregate, late events belongs to the cleaned slice would be merged into the merged window state, and would be counted into the later slice.

For example, for a CUMULATE window, step is 1 minute, size is 1 day.

SELECT window_start, window_end, COUNT(USER_ID)
  FROM TABLE(
    CUMULATE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '1' MINUTES, INTERVAL '1' DAY))
  GROUP BY window_start, window_end;

When the watermark already comes to 11:01, result of window [00:00, 11:01) would be emitted. Let's assume the result is INSERT (00:00, 11:01, 4)

Then if a late record which event time is 11:00 comes, it would be merged into merged state, and would be counted into the later slice, for example, for window [00:00, 11:02), [00:00, 11:03)... But the emitted window result INSERT (00:00, 11:01, 4) would not be retracted and updated.

The behavior would be different if the job recover from savepoint/checkpoint.

Let's do a savepoint after watermark comes to 11:01 and emit (00:00, 11:01, 4).

Then recover the job from savepoint. Watermarks are not checkpointed and they need to be repopulated again. So after recovered, the watermark may rollback to 11:00, then if a record which event time is 11:00 comes, it would not be processed as late event, after watermark comes to 11:01 again, a window result INSERT (00:00, 11:01, 5)  would be emitted to downstream.

So the downstream operator would receive two INSERT record for WINDOW (00:00, 11:01) which may leads to wrong result.

 

Solution

There are two solutions for the problem:

1. save watermark to state in slice shared operator. (Prefered)
2. update the behavior for late event. For example, retract the emitted result and send the updated result. It needs to change the behavior of slice state clean mechanism because we clean the slice state after watermark exceeds the slice end currently.

 

Issue Links

• relates to: FLINK-21305 Fix Cumulative and Hopping window should accumulate late events belonging to the cleaned slice
• links to: GitHub Pull Request #17509