Write Flink SQL temporal joins and lookup joins for stream enrichment

Steps

1. Define the build (dimension) table with a PRIMARY KEY and either set it as a lookup source (connector = 'jdbc' or similar) or define it as a versioned table using a rowtime attribute.
2. For a lookup join, write SELECT ... FROM stream_table AS s LEFT JOIN dim_table FOR SYSTEM_TIME AS OF s.proc_time ON s.key = d.key, where proc_time is a processing-time attribute.
3. For a temporal join (event-time), ensure the dimension table is versioned (append-only with timestamps) and use FOR SYSTEM_TIME AS OF s.event_time.
4. Validate join correctness by checking that the dimension table is being cached or refreshed at the expected rate.
5. Tune lookup cache settings (cache.max-rows, cache.ttl in connector options) for the lookup join path.
6. Monitor enrichment hit rate by counting NULL dimension fields post-join.