Examples & Integration Demos

-- ── Section 1: MinIO connection properties ──────────────────────────────
-- Adjust endpoint, credentials, and bucket to match your deployment.
-- These are session-scoped: re-run after any session renewal.

SET 'otter.minio.endpoint'          = 'http://minio:9000';
SET 'otter.minio.access-key'        = 'minioadmin';
SET 'otter.minio.secret-key'        = 'minioadmin';
SET 'otter.minio.bucket'            = 'ml-models';
SET 'otter.model.cache.max-size'    = '20';
SET 'otter.model.cache.ttl-minutes' = '60';

-- Verify the properties are visible to the session
SET;

-- ── Section 2: Load JAR and register scalar UDF ─────────────────────────

-- Load the shaded JAR onto the Gateway session classpath.
-- Path must exist on the SQL Gateway container filesystem.
ADD JAR '/var/www/udf-jars/otter-stream-sql-1.0.17-flink-udf.jar';

-- Confirm the JAR is visible
SHOW JARS;

-- Register the ML inference scalar function
CREATE TEMPORARY FUNCTION IF NOT EXISTS ml_score
AS 'com.codedstreams.otterstreams.sql.udf.MLInferenceFunction'
LANGUAGE JAVA;

-- Confirm registration
SHOW USER FUNCTIONS;

-- ── Section 3: Preload the model into ModelCache ────────────────────────
-- Creating this connector table triggers MinIO download and engine init.
-- The table itself is not queried directly in this demo; it exists solely
-- to warm the ModelCache before the streaming pipeline starts.

CREATE TEMPORARY TABLE fraud_model_source (
    score      DOUBLE,
    confidence DOUBLE
) WITH (
    'connector'          = 'ml-inference',

    -- Logical name used in ml_score(features, '')
    'model.name'         = 'fraud-detector',

    -- MinIO path: s3a:////
    -- The loader downloads saved_model.pb + variables/ from this prefix.
    'model.path'         = 's3a://ml-models/fraud-detector/v1/',
    'model.format'       = 'tensorflow-savedmodel',
    'model.version'      = 'v1',

    -- MinIO / S3-compatible storage settings
    'model.s3.endpoint'  = 'http://minio:9000',
    'model.s3.access-key'= 'minioadmin',
    'model.s3.secret-key'= 'minioadmin',
    'model.s3.path-style'= 'true',

    -- Cache behaviour
    'cache.enabled'      = 'true',
    'cache.ttl-minutes'  = '60',

    -- Inference settings
    'batch.size'         = '1',
    'async.enabled'      = 'false'
);

-- A lightweight describe confirms the connector initialised without error.
-- If the model failed to download you will see an error here rather than
-- silently at inference time.
DESCRIBE fraud_model_source;

-- ── Section 4: Kafka source table (transactions) ─────────────────────────
-- Adjust bootstrap.servers and topic to match your Kafka deployment.

CREATE TEMPORARY TABLE transactions (
    transaction_id   STRING,
    user_id          STRING,
    merchant_id      STRING,
    amount           DOUBLE,
    currency         STRING,
    card_type        STRING,
    hour_of_day      INT,
    day_of_week      INT,
    is_international BOOLEAN,
    -- Feature map pre-assembled upstream (or computed in the query below)
    features         MAP<STRING, STRING>,
    event_time       TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector'                     = 'kafka',
    'topic'                         = 'transactions',
    'properties.bootstrap.servers'  = 'kafka:9092',
    'properties.group.id'           = 'otter-fraud-demo',
    'scan.startup.mode'             = 'latest-offset',
    'format'                        = 'json',
    'json.timestamp-format.standard'= 'ISO-8601',
    'json.fail-on-missing-field'    = 'false',
    'json.ignore-parse-errors'      = 'true'
);

-- ── Kafka sink table (fraud alerts) ───────────────────────────────────────
CREATE TEMPORARY TABLE fraud_alerts (
    alert_id         STRING,
    transaction_id   STRING,
    user_id          STRING,
    amount           DOUBLE,
    fraud_score      DOUBLE,
    risk_tier        STRING,
    detected_at      TIMESTAMP(3)
) WITH (
    'connector'                    = 'kafka',
    'topic'                        = 'fraud-alerts',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format'                       = 'json',
    'json.timestamp-format.standard' = 'ISO-8601'
);

-- ── Section 5: Real-time fraud inference pipeline ───────────────────────
-- Writes scored events with risk_score > 0.40 to the fraud-alerts topic.
-- Adjust the threshold to suit your operating point.

INSERT INTO fraud_alerts
SELECT
    -- Deterministic alert ID from transaction ID + score bucket
    CONCAT(transaction_id, '-', CAST(FLOOR(fraud_score * 10) AS STRING))
        AS alert_id,
    transaction_id,
    user_id,
    amount,
    fraud_score,

    -- Risk tier classification downstream of the model score
    CASE
        WHEN fraud_score >= 0.85 THEN 'CRITICAL'
        WHEN fraud_score >= 0.65 THEN 'HIGH'
        WHEN fraud_score >= 0.40 THEN 'MEDIUM'
        ELSE                          'LOW'
    END AS risk_tier,

    CURRENT_TIMESTAMP AS detected_at

FROM (
    SELECT
        transaction_id,
        user_id,
        amount,

        -- Build feature map from individual columns.
        -- All values are cast to STRING because the UDF accepts MAP<STRING,STRING>.
        -- The inference engine casts them back to the correct numeric types internally.
        MAP[
            'amount',           CAST(amount          AS STRING),
            'hour_of_day',      CAST(hour_of_day     AS STRING),
            'day_of_week',      CAST(day_of_week     AS STRING),
            'is_international', CAST(is_international AS STRING),
            'card_type',        card_type,
            'currency',         currency
        ] AS feature_map,

        event_time

    FROM transactions
) enriched

-- Call the registered UDF; COALESCE guards against NULL when the
-- model name is not yet in ModelCache (should not happen after preload).
CROSS JOIN LATERAL (
    SELECT COALESCE(
        ml_score(feature_map, 'fraud-detector'),
        0.0
    ) AS fraud_score
) scores

WHERE fraud_score >= 0.40;

-- ── Section 6: Verification ──────────────────────────────────────────────

-- 1. Confirm the shaded JAR is on the classpath
SHOW JARS;
-- Expected: .../otter-stream-sql-1.0.17-flink-udf.jar

-- 2. Confirm the UDF is registered
SHOW USER FUNCTIONS;
-- Expected: ml_score appears in the list

-- 3. Smoke-test the UDF with a static feature map.
--    If the model is loaded this returns a DOUBLE between 0.0 and 1.0.
--    If it returns NULL the model name was not found in ModelCache.
SELECT ml_score(
    MAP['amount', '250.00',
        'hour_of_day', '23',
        'day_of_week', '6',
        'is_international', 'true',
        'card_type', 'CREDIT',
        'currency', 'USD'],
    'fraud-detector'
) AS smoke_test_score;

-- 4. Confirm the connector table is accessible (model loaded from MinIO)
DESCRIBE fraud_model_source;

-- 5. Count messages received on the transactions topic in the last 60 s
--    (requires the streaming job from Section 5 to be running)
SELECT COUNT(*) AS events_in_last_60s
FROM transactions
WHERE event_time >= CURRENT_TIMESTAMP - INTERVAL '60' SECOND;

-- ── Extended fraud detection with velocity features ──────────────────────

-- Step A: Create a view that computes velocity features using a 5-minute
-- tumbling window before the inference step.
CREATE TEMPORARY VIEW transaction_velocity AS
SELECT
    user_id,
    COUNT(*)                                     AS txn_count_5m,
    SUM(amount)                                  AS total_amount_5m,
    MAX(amount)                                  AS max_amount_5m,
    COUNT(DISTINCT merchant_id)                  AS distinct_merchants_5m,
    TUMBLE_START(event_time, INTERVAL '5' MINUTE) AS window_start
FROM transactions
GROUP BY
    user_id,
    TUMBLE(event_time, INTERVAL '5' MINUTE);

-- Step B: Join raw transactions with velocity window and run inference.
-- The join key is user_id; window_start aligns to the tumble boundary.
CREATE TEMPORARY VIEW scored_transactions AS
SELECT
    t.transaction_id,
    t.user_id,
    t.amount,
    t.event_time,
    COALESCE(
        ml_score(
            MAP[
                'amount',                CAST(t.amount               AS STRING),
                'hour_of_day',           CAST(t.hour_of_day          AS STRING),
                'day_of_week',           CAST(t.day_of_week          AS STRING),
                'is_international',      CAST(t.is_international      AS STRING),
                'card_type',             t.card_type,
                'currency',              t.currency,
                -- Velocity features from the window join
                'txn_count_5m',          CAST(v.txn_count_5m          AS STRING),
                'total_amount_5m',       CAST(v.total_amount_5m       AS STRING),
                'max_amount_5m',         CAST(v.max_amount_5m         AS STRING),
                'distinct_merchants_5m', CAST(v.distinct_merchants_5m AS STRING)
            ],
            'fraud-detector'
        ),
        -1.0   -- sentinel: model returned NULL (engine not loaded)
    ) AS fraud_score
FROM transactions t
LEFT JOIN transaction_velocity v
  ON t.user_id      = v.user_id
 AND t.event_time  >= v.window_start
 AND t.event_time   < v.window_start + INTERVAL '5' MINUTE;

-- Step C: Route high-confidence fraud to the alerts sink,
-- and model errors (sentinel -1.0) to a dead-letter topic.
INSERT INTO fraud_alerts
SELECT
    CONCAT(transaction_id, '-alert') AS alert_id,
    transaction_id,
    user_id,
    amount,
    fraud_score,
    CASE
        WHEN fraud_score >= 0.85 THEN 'CRITICAL'
        WHEN fraud_score >= 0.65 THEN 'HIGH'
        ELSE                          'MEDIUM'
    END AS risk_tier,
    event_time AS detected_at
FROM scored_transactions
WHERE fraud_score >= 0.40;

-- ── IoT anomaly detection with XGBoost model from MinIO ─────────────────

-- 1. Preload the XGBoost model from MinIO
CREATE TEMPORARY TABLE anomaly_model_source (
    score DOUBLE
) WITH (
    'connector'           = 'ml-inference',
    'model.name'          = 'anomaly-detector',
    'model.path'          = 's3a://ml-models/anomaly-detector/v2/model.json',
    'model.format'        = 'xgboost-json',
    'model.s3.endpoint'   = 'http://minio:9000',
    'model.s3.access-key' = 'minioadmin',
    'model.s3.secret-key' = 'minioadmin',
    'model.s3.path-style' = 'true',
    'cache.enabled'       = 'true'
);

-- 2. Sensor readings source
CREATE TEMPORARY TABLE sensor_readings (
    device_id    STRING,
    temperature  DOUBLE,
    pressure     DOUBLE,
    humidity     DOUBLE,
    vibration    DOUBLE,
    rpm          DOUBLE,
    reading_time TIMESTAMP(3),
    WATERMARK FOR reading_time AS reading_time - INTERVAL '10' SECOND
) WITH (
    'connector'                    = 'kafka',
    'topic'                        = 'sensor-readings',
    'properties.bootstrap.servers' = 'kafka:9092',
    'properties.group.id'          = 'otter-anomaly-demo',
    'scan.startup.mode'            = 'latest-offset',
    'format'                       = 'json'
);

-- 3. Anomaly alerts sink
CREATE TEMPORARY TABLE anomaly_alerts (
    device_id      STRING,
    anomaly_score  DOUBLE,
    severity       STRING,
    reading_time   TIMESTAMP(3),
    detected_at    TIMESTAMP(3)
) WITH (
    'connector'                    = 'kafka',
    'topic'                        = 'anomaly-alerts',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format'                       = 'json'
);

-- 4. Sliding window aggregation then inference
INSERT INTO anomaly_alerts
SELECT
    device_id,
    anomaly_score,
    CASE
        WHEN anomaly_score >= 0.90 THEN 'CRITICAL'
        WHEN anomaly_score >= 0.70 THEN 'HIGH'
        ELSE                           'MEDIUM'
    END AS severity,
    window_end AS reading_time,
    CURRENT_TIMESTAMP AS detected_at
FROM (
    SELECT
        device_id,
        HOP_END(reading_time, INTERVAL '30' SECOND, INTERVAL '5' MINUTE) AS window_end,
        COALESCE(
            ml_score(
                MAP[
                    'temperature', CAST(AVG(temperature) AS STRING),
                    'pressure',    CAST(AVG(pressure)    AS STRING),
                    'humidity',    CAST(AVG(humidity)    AS STRING),
                    'vibration',   CAST(AVG(vibration)   AS STRING),
                    'rpm',         CAST(AVG(rpm)         AS STRING)
                ],
                'anomaly-detector'
            ),
            0.0
        ) AS anomaly_score
    FROM sensor_readings
    GROUP BY
        device_id,
        HOP(reading_time, INTERVAL '30' SECOND, INTERVAL '5' MINUTE)
) windowed
WHERE anomaly_score >= 0.60;