What is Flink SQL and when do you use it

-- Continuous query: sum deposits per branch per hour
SELECT
    branch_id,
    TUMBLE_START(event_time, INTERVAL '1' HOUR) AS window_start,
    SUM(amount) AS hourly_deposits
FROM transactions
WHERE txn_type = 'DEPOSIT'
GROUP BY branch_id, TUMBLE(event_time, INTERVAL '1' HOUR);
-- Runs forever, emitting results as each hour's window closes

-- Dynamic table backed by Kafka
CREATE TABLE transactions (
    txn_id       STRING,
    account_id   STRING,
    branch_id    STRING,
    amount       DECIMAL(15, 2),
    txn_type     STRING,
    event_time   TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic'     = 'nbi-transactions',
    'format'    = 'json',
    'properties.bootstrap.servers' = 'kafka.bank.internal:9092'
);

package com.bank.flink;

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

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

        // Register dynamic table backed by Kafka
        tEnv.executeSql("CREATE TABLE transactions (...) WITH ('connector'='kafka', ...)");

        // Continuous query over dynamic table
        tEnv.executeSql(
            "SELECT branch_id, SUM(amount) AS total " +
            "FROM transactions GROUP BY branch_id"
        );
    }
}

-- Windowed aggregation with TUMBLE
SELECT branch_id, TUMBLE_START(event_time, INTERVAL '1' HOUR), SUM(amount)
FROM transactions GROUP BY branch_id, TUMBLE(event_time, INTERVAL '1' HOUR);

-- Temporal join for point-in-time exchange rate lookup
SELECT t.txn_id, t.amount * r.rate_to_inr AS amount_inr
FROM forex_transactions t
JOIN exchange_rates FOR SYSTEM_TIME AS OF t.event_time AS r
    ON t.currency = r.currency;

-- MATCH_RECOGNIZE for pattern detection
SELECT * FROM stock_prices
MATCH_RECOGNIZE (
    PARTITION BY symbol ORDER BY event_time
    PATTERN (A B+ C)
    DEFINE B AS B.price < A.price, C AS C.price >= A.price
);

-- CDC source for real-time database sync
CREATE TABLE accounts_cdc (
    account_id STRING, balance DECIMAL(15,2),
    PRIMARY KEY (account_id) NOT ENFORCED
) WITH ('connector' = 'mysql-cdc', 'hostname' = 'mysql.bank.internal',
        'database-name' = 'nbi', 'table-name' = 'accounts');

// Flink: continuous query, event-at-a-time processing
tEnv.executeSql("SELECT account_id, SUM(amount) FROM transactions GROUP BY account_id");
// Each transaction updates the aggregate immediately

// Spark: micro-batch, processes accumulated events per trigger
// spark.readStream.format("kafka").load()
//   .groupBy("account_id").agg(sum("amount"))
//   .writeStream.trigger(processingTime="10 seconds").start()
// Updates happen every 10 seconds

-- Set state TTL to 24 hours
SET 'table.exec.state.ttl' = '86400000';

SELECT account_id, SUM(amount)
FROM transactions
GROUP BY account_id;
-- Accounts not seen in 24h have their state cleaned up

// Senior tip: use EXPLAIN to understand the physical plan
tEnv.executeSql("EXPLAIN SELECT branch_id, SUM(amount) FROM transactions GROUP BY branch_id");
// Shows: changelog mode, state operators, optimization choices