Implement Flink temporal joins and interval joins for enriching event streams with dimension tables

Steps

1. For a temporal join against a slowly-changing dimension table in a Flink catalog or Hive table, use the FOR SYSTEM_TIME AS OF syntax in SQL: join the event table to the dimension table at the version corresponding to each event's event_time, ensuring point-in-time correctness.
2. For event-to-event interval joins (two append-only streams where you want to match events within a time bound), use the DataStream API intervalJoin().between() or the SQL INTERVAL predicate in the WHERE clause to restrict the time window.
3. Register the dimension table as a lookup source with a LookupTableSource implementation (or use the built-in JDBC or HBase connectors) for synchronous enrichment; set a lookup cache TTL to avoid hitting the backend on every record.
4. Assign watermarks and event-time attributes to both streams before the join; Flink uses the minimum watermark of the two streams to advance join state cleanup.
5. Set the state TTL (table.exec.state.ttl) to bound the amount of unmatched records retained in state for interval joins in high-throughput scenarios.
6. Test the join correctness with out-of-order test data using Flink's MiniClusterExtension and AssertCollect utilities.