@Bugen Zhao @Tianyi Zhuang @Tianshuo Shi

Updates

• July 16, 2024 The assumption that the very initial messages are always Watermark does not hold anymore (since https://github.com/risingwavelabs/risingwave/pull/12462). However, we don’t find it problematic, given that watermark is generally for optimization purposes.

In RFC: The WatermarkFilter and StreamSort Operator, we showed how watermark will be used in RisingWave. This doc will introduce how the key operators should be implemented in our system.

Overview of the design. Image explained in detail below

Basically, the general idea is similar to the design of Flink like above, while one should note that there’re several major differences in our system:

• Watermark can be generated everywhere, not associated with or to be a property of the sources.
• We’re able to scale the stream graph online. Therefore, the state persistence and recovery should be considered.

We’ll introduce 3 operators one by one in the following.

Watermark Filter (Stateful)

The Watermark Filter maintains the timestamp-like column as the watermark while taking input records, filtering out the outdated records based on the watermark and a fixed timeout. Thus, the Watermark Filter is the source of truth of the Watermark messages, and it does not need to reorder or buffer the input chunks.

https://viewer.diagrams.net/?border=0&tags={}&highlight=0000ff&edit=_blank&layers=1&nav=1&title=watermark-1.drawio&open=R5VdNc5swEP01nkkP6fBhsHN0bCc9tDOd8XTSnjoKyKBGICJEwP31XSHJIAP5bi%2FNRehptSvt27dyZv46a645KtIvLMZ05jlxM%2FM3M89z514Ig0QOClnM5wpIOIm1UQfsyG%2BsQUejFYlxaRkKxqgghQ1GLM9xJCwMcc5q22zPqB21QAkeALsI0SF6Q2KRKnTpLTr8EyZJaiK74YVayZAx1jcpUxSzugf525m%2F5owJ9ZU1a0xl8kxe1L6ridXjwTjOxXM2OM0dC%2FNdk9xef7svdouidn%2BeB8rLA6KVvrA%2BrDiYDOAYEqKnjIuUJSxHdNuhl5xVeYxlGAdmnc1nxgoAXQB%2FYSEOml1UCQZQKjKqV%2FcsF2tGGW8j%2BldXDvwBPryivnXJKh7hR%2B6l7eTZext1Yq4xy7DgBzDgmCJBHmyyka6Z5Gh33PqVETgKxFQWQajJ1dU9dxzbhUA8wULv6siBj94xOqil7AX0uSP0hVTolFo8hvcVMwvnZcvECgzcoGjaTJt1%2BErkeIME5hnid9sGR8AYPxMkw6ySXluHjv%2FBBIOzq3hqq4FvuUFq40ztXAyLjFJQtCymOiUC7wrU0ltDU7FLZbIkHjAXuHmUbL3qLW3SfENi3QkcmFRY2hP30pmuD4vZl9LoPUOFebyS7QxmEUVlSSI7L7YGITn88L0%2F%2BSEnHwMz3TT9xc1Bzyazq%2Br4HQTXy3EwkmKDvVGXR%2FqmdKkayECXA0fzE0f%2Bxb8VuD%2BoDMgFuHI8NYRqOFF9dOylnar9%2Fb7tqgOh%2B5Mqlp7nagjUsFSDfJzfGnYxGfak8kHVwq51REmSSyFAlWIIeCm1T%2BDNXumFjMSxepkwdDp027qS9V1InlrmgstZsJG%2BoLWpbti6LgVnd9i8RTnLpZc9ofQEeoc25DonpbUMBm3IHWtD%2Ft9qQ%2BFTr8kowY4ThmME35w50w9EW0WvreTpiI88SZPF%2FLaIwRN3fIFSpoMspoP8J2I5%2FaE1JpbRB%2BUVYoFp96tcNfLufxt%2F%2Bwc%3D

There’re two notable facts:

• The watermark filters run independently on each partition (parallel unit), so they’ll emit Watermark messages based on the local input like the specific splits assigned to the upstream source executor, which can be slightly different from others.