What is a temporal join in Flink SQL

-- Probe side: transactions with event time
CREATE TABLE forex_transactions (
    txn_id       STRING,
    currency     STRING,
    amount       DECIMAL(15, 2),
    txn_time     TIMESTAMP(3),
    WATERMARK FOR txn_time AS txn_time - INTERVAL '5' SECOND
) WITH ('connector' = 'kafka', 'topic' = 'nbi-forex-txns',
        'format' = 'json',
        'properties.bootstrap.servers' = 'kafka.bank.internal:9092');

-- Build side: versioned exchange rates
CREATE TABLE exchange_rates (
    currency     STRING,
    rate_to_inr  DECIMAL(10, 4),
    update_time  TIMESTAMP(3),
    WATERMARK FOR update_time AS update_time - INTERVAL '5' SECOND,
    PRIMARY KEY (currency) NOT ENFORCED
) WITH ('connector' = 'kafka', 'topic' = 'nbi-exchange-rates',
        'format' = 'json',
        'properties.bootstrap.servers' = 'kafka.bank.internal:9092');

-- Temporal join: get the exchange rate valid at each transaction's time
SELECT
    t.txn_id,
    t.currency,
    t.amount,
    r.rate_to_inr,
    t.amount * r.rate_to_inr AS amount_inr,
    t.txn_time
FROM forex_transactions t
JOIN exchange_rates FOR SYSTEM_TIME AS OF t.txn_time AS r
    ON t.currency = r.currency;

-- A transaction at 10:00 AM gets the rate valid at 10:00 AM,
-- even if the rate changed at 10:01 AM.

package com.bank.flink;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class TemporalJoinDemo {
    public static void main(String[] args) {
        TableEnvironment tEnv = TableEnvironment.create(
            EnvironmentSettings.inStreamingMode());

        // Register source tables (DDL from above)
        // ...

        // Temporal join for point-in-time currency conversion
        tEnv.executeSql(
            "INSERT INTO converted_transactions " +
            "SELECT t.txn_id, t.currency, t.amount, " +
            "       r.rate_to_inr, t.amount * r.rate_to_inr AS amount_inr " +
            "FROM forex_transactions t " +
            "JOIN exchange_rates FOR SYSTEM_TIME AS OF t.txn_time AS r " +
            "  ON t.currency = r.currency"
        );
        // Reprocessing always produces the same result
    }
}

-- Stream-stream join: match trade to confirmation within 5 minutes
SELECT t.trade_id, t.amount, c.confirmed_at
FROM trades t
JOIN confirmations c
    ON t.trade_id = c.trade_id
    AND c.confirmed_at BETWEEN t.trade_time AND t.trade_time + INTERVAL '5' MINUTE;

-- Temporal join: enrich trade with rate valid at trade time
SELECT t.trade_id, t.amount, r.rate_to_inr
FROM trades t
JOIN exchange_rates FOR SYSTEM_TIME AS OF t.trade_time AS r
    ON t.currency = r.currency;

-- Processing-time temporal join: always gets the latest rate
SELECT t.txn_id, t.amount, r.rate_to_inr
FROM forex_transactions t
JOIN exchange_rates FOR SYSTEM_TIME AS OF t.proc_time AS r
    ON t.currency = r.currency;
-- Not deterministic: reprocessing may get different rates
-- Use only when point-in-time correctness is not needed

-- CORRECT: versioned table with primary key and event-time watermark
CREATE TABLE exchange_rates (
    currency     STRING,
    rate_to_inr  DECIMAL(10, 4),
    update_time  TIMESTAMP(3),
    WATERMARK FOR update_time AS update_time - INTERVAL '5' SECOND,
    PRIMARY KEY (currency) NOT ENFORCED
) WITH ('connector' = 'kafka', 'topic' = 'rates', 'format' = 'json',
        'properties.bootstrap.servers' = 'kafka.bank.internal:9092');

-- WRONG: missing primary key or watermark
-- Without primary key: Flink cannot track versions per entity
-- Without watermark: Flink cannot determine version validity

// Senior tip: verify temporal join output determinism
// Run the same input twice → same output each time
// This is guaranteed with event-time temporal joins
// but NOT with processing-time temporal joins