What is MATCH_RECOGNIZE in Flink SQL

-- Detect a drop-low-recovery pattern in stock prices
SELECT *
FROM stock_prices
MATCH_RECOGNIZE (
    PARTITION BY symbol
    ORDER BY event_time
    MEASURES
        A.price AS start_price,
        LAST(B.price) AS lowest_price,
        C.price AS recovery_price,
        A.event_time AS pattern_start,
        C.event_time AS pattern_end
    ONE ROW PER MATCH
    AFTER MATCH SKIP PAST LAST ROW
    PATTERN (A B+ C)
    DEFINE
        A AS A.price > 100,
        B AS B.price < A.price,
        C AS C.price >= A.price
);

-- Breaking down each clause:
SELECT * FROM loan_defaults
MATCH_RECOGNIZE (
    -- Split into independent sequences per loan
    PARTITION BY loan_id

    -- Order by time (required for pattern detection)
    ORDER BY payment_date

    -- Output columns from matched rows
    MEASURES
        A.payment_date AS first_missed,
        LAST(B.payment_date) AS last_missed,
        C.payment_date AS partial_payment_date,
        COUNT(B.payment_date) AS consecutive_misses

    -- One summary row per match
    ONE ROW PER MATCH

    -- Resume matching after the last matched row
    AFTER MATCH SKIP PAST LAST ROW

    -- Pattern: one missed payment, then more missed, then a partial payment
    PATTERN (A B+ C)

    -- Conditions for each variable
    DEFINE
        A AS A.amount_paid = 0,
        B AS B.amount_paid = 0,
        C AS C.amount_paid > 0 AND C.amount_paid < C.emi_due
);

-- Quantifier examples:
-- A       exactly one row
-- B+      one or more rows (greedy)
-- B*      zero or more rows (greedy)
-- B?      zero or one row
-- B{3}    exactly 3 rows
-- B{2,5}  between 2 and 5 rows

-- Detect 3 or more consecutive failed payments on a loan
SELECT * FROM payment_events
MATCH_RECOGNIZE (
    PARTITION BY loan_id
    ORDER BY payment_date
    MEASURES
        FIRST(A.payment_date) AS first_failure,
        LAST(A.payment_date) AS last_failure,
        COUNT(A.payment_date) AS failure_count
    ONE ROW PER MATCH
    AFTER MATCH SKIP PAST LAST ROW
    PATTERN (A{3,})
    DEFINE
        A AS A.status = 'FAILED'
);

package com.bank.flink;

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

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

        // Register stock_prices table with event-time watermark
        // ...

        // Detect drop-then-recovery pattern
        tEnv.executeSql(
            "SELECT * FROM stock_prices " +
            "MATCH_RECOGNIZE (" +
            "  PARTITION BY symbol " +
            "  ORDER BY event_time " +
            "  MEASURES " +
            "    A.price AS start_price, " +
            "    LAST(B.price) AS low_price, " +
            "    C.price AS recovery_price " +
            "  ONE ROW PER MATCH " +
            "  AFTER MATCH SKIP PAST LAST ROW " +
            "  PATTERN (A B+ C) " +
            "  DEFINE " +
            "    B AS B.price < A.price, " +   // B references A's price
            "    C AS C.price >= A.price" +     // C references A's price
            ")"
        );
    }
}

-- AFTER MATCH SKIP PAST LAST ROW:
--   Resume matching after the last row of the current match.
--   No overlapping matches. Lower state cost.

-- AFTER MATCH SKIP TO NEXT ROW:
--   Resume from the row after the first row of the match.
--   Allows overlapping matches. Higher state cost.

-- AFTER MATCH SKIP TO LAST B:
--   Resume from the last row bound to variable B.
--   Useful for chained patterns.

-- Loan default early warning pattern
CREATE TABLE emi_payments (
    loan_id      STRING,
    emi_amount   DECIMAL(15, 2),
    paid_amount  DECIMAL(15, 2),
    due_date     TIMESTAMP(3),
    WATERMARK FOR due_date AS due_date - INTERVAL '1' DAY
) WITH ('connector' = 'kafka', 'topic' = 'nbi-emi-payments',
        'format' = 'json',
        'properties.bootstrap.servers' = 'kafka.bank.internal:9092');

SELECT * FROM emi_payments
MATCH_RECOGNIZE (
    PARTITION BY loan_id
    ORDER BY due_date
    MEASURES
        FIRST(A.due_date) AS first_missed_date,
        LAST(A.due_date) AS last_missed_date,
        COUNT(A.due_date) AS missed_count,
        B.paid_amount AS partial_amount
    ONE ROW PER MATCH
    AFTER MATCH SKIP PAST LAST ROW
    PATTERN (A{2,} B)
    DEFINE
        A AS A.paid_amount = 0,
        B AS B.paid_amount > 0 AND B.paid_amount < B.emi_amount
);

-- WRONG: unbounded pattern with no skip control
-- PATTERN (A B*) with SKIP TO NEXT ROW
-- → keeps many partial matches alive → state explosion

-- CORRECT: bounded pattern with skip past last row
-- PATTERN (A B{1,10} C) with SKIP PAST LAST ROW
-- → at most 10 rows per partial match → bounded state

// Senior tip: EXPLAIN the MATCH_RECOGNIZE query to see the physical plan
tEnv.executeSql("EXPLAIN SELECT * FROM stock_prices MATCH_RECOGNIZE (...)");
// Shows the NFA operator and state requirements