[jira] [Comment Edited] (FLINK-18405) Add watermark support for unaligned checkpoints

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

    [ https://issues.apache.org/jira/browse/FLINK-18405?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17752324#comment-17752324 ] 

Piotr Nowojski edited comment on FLINK-18405 at 8/9/23 8:37 AM:
----------------------------------------------------------------

To rephrase the problem using the concrete SQL OVER example.

SQL Over ({{RowTimeRowsBoundedPrecedingFunction}}) has this logic:
{code:java}
    @Override
    public void processElement(
            RowData input,
            KeyedProcessFunction<K, RowData, RowData>.Context ctx,
            Collector<RowData> out)
            throws Exception {
        (...)

        // triggering timestamp for trigger calculation
        long triggeringTs = input.getLong(rowTimeIdx);

        Long lastTriggeringTs = lastTriggeringTsState.value();
        if (lastTriggeringTs == null) {
            lastTriggeringTs = 0L;
        }

        // check if the data is expired, if not, save the data and register event time timer
        if (triggeringTs > lastTriggeringTs) {
            (...)
        } else {
            numLateRecordsDropped.inc();
        }
    }
{code}
The problem stated by this ticket can apply both to aligned and unaligned checkpoints. Namely every time after recovery, this operator/function processes any elements, before receiving a watermark, it will consider all of the records late, and it will drop them.

For aligned checkpoints, there is a workaround. After recovery, before source produces any record, it could emit last emitted watermark, thus ensuring that watermarks will be received by downstream operators before any elements.

For unaligned checkpoints, we can't use that trick, as anyway the in-flight data are recovered and processed before anything new produced by the sources :confused:


was (Author: pnowojski):
To rephrase the problem using the concrete SQL OVER example.

SQL Over has this logic:
{code:java}
    @Override
    public void processElement(
            RowData input,
            KeyedProcessFunction<K, RowData, RowData>.Context ctx,
            Collector<RowData> out)
            throws Exception {
        (...)

        // triggering timestamp for trigger calculation
        long triggeringTs = input.getLong(rowTimeIdx);

        Long lastTriggeringTs = lastTriggeringTsState.value();
        if (lastTriggeringTs == null) {
            lastTriggeringTs = 0L;
        }

        // check if the data is expired, if not, save the data and register event time timer
        if (triggeringTs > lastTriggeringTs) {
            (...)
        } else {
            numLateRecordsDropped.inc();
        }
    }
{code}
The problem stated by this ticket can apply both to aligned and unaligned checkpoints. Namely every time after recovery, this operator/function processes any elements, before receiving a watermark, it will consider all of the records late, and it will drop them.

For aligned checkpoints, there is a workaround. After recovery, before source produces any record, it could emit last emitted watermark, thus ensuring that watermarks will be received by downstream operators before any elements.

For unaligned checkpoints, we can't use that trick, as anyway the in-flight data are recovered and processed before anything new produced by the sources :confused:

> Add watermark support for unaligned checkpoints
> -----------------------------------------------
>
>                 Key: FLINK-18405
>                 URL: https://issues.apache.org/jira/browse/FLINK-18405
>             Project: Flink
>          Issue Type: Improvement
>          Components: Runtime / Network
>    Affects Versions: 1.12.0
>            Reporter: Arvid Heise
>            Priority: Not a Priority
>              Labels: auto-deprioritized-major
>
> Currently, Flink generates the watermark as a first step of recovery instead of 
> storing the latest watermark in the operators to ease rescaling. In unaligned 
> checkpoints, that means on recovery, Flink generates watermarks after it 
> restores in-flight data. If your pipeline uses an operator that applies the
> latest watermark on each record, it will produce incorrect results during 
> recovery if the watermark is not directly or indirectly part of the operator 
> state. Thus, SQL OVER operator should not be used with unaligned
> checkpoints, while window operators are safe to use. 
> A possible solution is to store the watermark in the operator state. If rescaling may occur, watermarks should be stored per key-group in a union-state. 



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