[FLINK-8478] Add support for time-bounded joins in DataStream API

• Type: New Feature
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Component/s: API / DataStream, API / Scala
• Labels: auto-deprioritized-major auto-deprioritized-minor auto-unassigned
• Assignee: Unassigned
• Reporter: Florian Schmidt
• Created: 22/Jan/18 15:52
• Updated: 18/Feb/22 10:38

Description

The goal is to implement a DataStream API to join together two streams based on a key and user defined time boundaries. 

This feature should mirror what is already possible in Flink SQL:

SELECT s1.value, s2.value FROM Table s1 JOIN Table s2 ON o.key = s.key AND o.rowtime BETWEEN s.rowtime AND s.rowtime + INTERVAL '1' HOUR;

Sub-Tasks

• Implement Java API to expose join functionality of TimeBoundedStreamJoinOperator: Closed
• Implement time-bounded inner join of streams as a TwoInputStreamOperator: Closed
• Implement Scala API to expose join functionality of TimeBoundedStreamJoinOperator: Closed
• Implement and expose option to use min / max / left / right timestamp for joined streamrecords: Open
• Implement and expose outer joins: Open
• Bucket MapState of TimeBoundedStreamJoinOperator for improved performance: Open
• Add separate docs page for different join types in DataStream API: Closed
• Add support for processing time to interval join: Open