What are Flink SQL hints and how are they used to override options or control join behavior

-- OPTIONS hint: override Kafka startup mode for this query only
SELECT *
FROM transactions /*+ OPTIONS('scan.startup.mode'='earliest-offset') */
WHERE amount > 1000000;
-- All other consumers of 'transactions' are unaffected

-- Replay from a specific timestamp for backfill
SELECT account_id, SUM(amount) AS total
FROM transactions /*+ OPTIONS(
    'scan.startup.mode'='timestamp',
    'scan.startup.timestamp-millis'='1704067200000'
) */
GROUP BY account_id;
-- Replays from Jan 1, 2024 00:00:00 UTC
-- Other jobs reading 'transactions' stay at latest-offset

package com.bank.flink;

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

public class OptionsHintDemo {
    public static void main(String[] args) {
        Configuration config = new Configuration();
        // Must enable dynamic table options for hints to work
        config.setBoolean("table.dynamic-table-options.enabled", true);

        TableEnvironment tEnv = TableEnvironment.create(
            EnvironmentSettings.newInstance()
                .inStreamingMode()
                .withConfiguration(config)
                .build());

        // Use hint to replay from earliest offset
        tEnv.executeSql(
            "SELECT * FROM transactions " +
            "/*+ OPTIONS('scan.startup.mode'='earliest-offset') */ " +
            "WHERE amount > 1000000"
        );
    }
}

// Enable OPTIONS hints (required)
Configuration config = new Configuration();
config.setBoolean("table.dynamic-table-options.enabled", true);

// Without this flag, using OPTIONS hint produces an error:
// "Dynamic table options is not allowed"

-- Lookup join with async hint for enrichment
SELECT t.txn_id, t.amount, a.risk_score
FROM transactions t
JOIN account_risk /*+ LOOKUP(
    'table'='account_risk',
    'async'='true',
    'output-mode'='allow_unordered',
    'capacity'='100'
) */ FOR SYSTEM_TIME AS OF t.proc_time AS a
    ON t.account_id = a.account_id;

-- Without async: 1 lookup at a time → 100 lookups/sec
-- With async:    100 concurrent lookups → 10,000 lookups/sec

package com.bank.flink;

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

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

        // Async lookup join for fraud enrichment
        tEnv.executeSql(
            "SELECT t.txn_id, t.amount, r.risk_score " +
            "FROM transactions t " +
            "JOIN risk_scores /*+ LOOKUP('table'='risk_scores', " +
            "  'async'='true', 'capacity'='200') */ " +
            "  FOR SYSTEM_TIME AS OF t.proc_time AS r " +
            "  ON t.account_id = r.account_id"
        );
    }
}

-- Retry on lookup miss: dimension row may not exist yet
SELECT t.txn_id, a.holder_name
FROM transactions t
JOIN accounts /*+ LOOKUP(
    'table'='accounts',
    'retry-predicate'='lookup_miss',
    'retry-strategy'='fixed_delay',
    'fixed-delay'='1s',
    'max-attempts'='3'
) */ FOR SYSTEM_TIME AS OF t.proc_time AS a
    ON t.account_id = a.account_id;
-- If account not found, retries 3 times with 1s delay

-- State TTL hint (per-query override)
SELECT account_id, COUNT(*) AS txn_count
FROM transactions /*+ OPTIONS('table.exec.state.ttl'='3600000') */
GROUP BY account_id;
-- State TTL = 1 hour for this query only

// Senior tip: measure throughput before and after async LOOKUP
// Before: synchronous → 100 records/sec (blocked by external DB latency)
// After:  async, capacity=200 → 15,000 records/sec
// The improvement is proportional to the external lookup latency