[FLINK-9717] Flush state of one side of the join if other side is bounded

• Type: Improvement
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.5.0
• Component/s: Table SQL / Planner
• Labels: auto-deprioritized-major auto-deprioritized-minor
• Assignee: Unassigned
• Reporter: Piotr Nowojski
• Created: 03/Jul/18 09:53
• Updated: 28/Dec/21 22:38

Description

Whenever one side of join receives MAX_WATERMARK, other side in joins (both normal and versioned joins) could flush the state from other side.

This highly useful optimisation that would speed up versioned joins and would allow normal joins of large unbounded streams with bounded tables (for example some static data).

edit:
Currently problem is that Flink doesn't keep & restore the last previous watermark after restoring from checkpoint and this is hard to workaround.

In other words, now we can easily "flush" one side of the join when we receive MAX_WATERMARK, but what should happen after restoring from checkpoint? There is no easy way to store the information that MAX_WATERMARK was previously reached. As far as I have thought about this, it can not be stored on the state of the Join operator and even if it could be done this way, it's probably not the proper/elegant solution. Probably the correct solution is to store MAX_WATERMARK in the state around watermark emitter/source operator and the last previously emitted watermark should be re-emitted when the job is restored.