SQL for
Analytics Engineering

-- ❌ WRONG — can't use COUNT() in WHERE
SELECT machine_id, COUNT(*) FROM downtime_events
WHERE COUNT(*) > 5 GROUP BY machine_id;

-- ✅ CORRECT
SELECT machine_id, COUNT(*) AS event_count
FROM downtime_events
WHERE shift_date >= '2024-01-01'   -- row-level filter ✓
GROUP BY machine_id
HAVING COUNT(*) > 5;              -- group-level filter ✓

-- INNER: only machines that have had downtime events
SELECT m.machine_name, COUNT(d.event_id)
FROM machines m INNER JOIN downtime_events d ON m.id = d.machine_id
GROUP BY m.machine_name;

-- LEFT: ALL machines, NULL count if no downtime events
SELECT m.machine_name, COUNT(d.event_id) AS event_count
FROM machines m LEFT JOIN downtime_events d ON m.id = d.machine_id
GROUP BY m.machine_name;
-- Zero-downtime machines show count = 0, not missing

-- Combine downtime events from two shifts
-- UNION ALL: keep all (may include duplicates if data is duped)
SELECT machine_id, event_time, reason FROM shift_1_events
UNION ALL
SELECT machine_id, event_time, reason FROM shift_2_events;

-- UNION: deduplicate (slower — only use when needed)
SELECT machine_id, event_time, reason FROM shift_1_events
UNION
SELECT machine_id, event_time, reason FROM shift_2_events;

-- Sensor dropout: fall back to previous reading, then to 0
SELECT
    sensor_id,
    recorded_at,
    COALESCE(current_reading, prev_reading, 0) AS display_value,
    CASE WHEN current_reading IS NULL THEN 'IMPUTED' ELSE 'ACTUAL' END AS data_source
FROM sensor_readings;

-- NULL-safe division (avoid division-by-zero)
COALESCE(units_good * 1.0 / NULLIF(units_total, 0), 0) AS quality_rate

SELECT shift_id, machine_id, total_downtime, rnk
FROM (
    SELECT
        shift_id,
        machine_id,
        SUM(down_mins) AS total_downtime,
        RANK() OVER (
            PARTITION BY shift_id
            ORDER BY SUM(down_mins) DESC
        ) AS rnk
    FROM production_events
    GROUP BY shift_id, machine_id
) ranked
WHERE rnk <= 3
ORDER BY shift_id, rnk;

-- Snowflake shorthand: QUALIFY instead of subquery
SELECT shift_id, machine_id, SUM(down_mins) AS total_downtime,
       RANK() OVER (PARTITION BY shift_id ORDER BY SUM(down_mins) DESC) AS rnk
FROM production_events
GROUP BY shift_id, machine_id
QUALIFY rnk <= 3;

-- ROW_NUMBER: deduplicate — get exactly one row per machine
SELECT * FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY machine_id ORDER BY ts DESC) AS rn
    FROM sensor_readings
) WHERE rn = 1;   -- latest reading per machine

-- DENSE_RANK: rank machines by defect count
DENSE_RANK() OVER (ORDER BY defect_count DESC) AS defect_rank

SELECT
    event_date,
    daily_defects,
    AVG(daily_defects) OVER (
        ORDER BY event_date
        ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
    ) AS rolling_7day_avg,
    -- Calendar-based (handles missing dates differently):
    AVG(daily_defects) OVER (
        ORDER BY event_date
        RANGE BETWEEN INTERVAL '6 days' PRECEDING AND CURRENT ROW
    ) AS rolling_7day_calendar_avg
FROM daily_quality_summary
ORDER BY event_date;

-- Step 1: Find duplicates
SELECT machine_id, event_timestamp, COUNT(*) AS occurrences
FROM sensor_readings
GROUP BY machine_id, event_timestamp
HAVING COUNT(*) > 1;

-- Step 2: Keep only the latest row per (machine, timestamp)
WITH deduped AS (
    SELECT *,
           ROW_NUMBER() OVER (
               PARTITION BY machine_id, event_timestamp
               ORDER BY ingested_at DESC
           ) AS rn
    FROM sensor_readings
)
SELECT * FROM deduped WHERE rn = 1;

-- ROW_NUMBER approach (preferred — deterministic)
SELECT machine_id, sensor_type, reading_value, recorded_at
FROM (
    SELECT *,
           ROW_NUMBER() OVER (
               PARTITION BY machine_id, sensor_type
               ORDER BY recorded_at DESC
           ) AS rn
    FROM sensor_readings
    WHERE dq_flag = 'GOOD'
) t
WHERE rn = 1;

-- Snowflake shorthand with QUALIFY
SELECT machine_id, sensor_type, reading_value, recorded_at
FROM sensor_readings
WHERE dq_flag = 'GOOD'
QUALIFY ROW_NUMBER() OVER (PARTITION BY machine_id, sensor_type ORDER BY recorded_at DESC) = 1;

SELECT
    machine_id,
    event_start,
    LAG(event_start) OVER (
        PARTITION BY machine_id
        ORDER BY event_start
    ) AS prev_event_start,
    DATEDIFF('minute',
        LAG(event_start) OVER (PARTITION BY machine_id ORDER BY event_start),
        event_start
    ) AS mins_since_last_downtime,
    LEAD(event_start) OVER (
        PARTITION BY machine_id
        ORDER BY event_start
    ) AS next_event_start
FROM downtime_events
ORDER BY machine_id, event_start;

-- ROWS: exactly 3 preceding rows regardless of date gaps
AVG(reading) OVER (ORDER BY ts ROWS BETWEEN 3 PRECEDING AND CURRENT ROW)

-- RANGE: all rows within a 3-hour interval (handles gaps correctly)
AVG(reading) OVER (
    ORDER BY ts
    RANGE BETWEEN INTERVAL '3 hours' PRECEDING AND CURRENT ROW
)

WITH shift_oee AS (
    SELECT
        pe.shift_id,
        pe.machine_id,
        SUM(pe.run_mins) * 1.0 / MAX(s.planned_mins)    AS availability,
        SUM(pe.units_good) * 1.0
            / NULLIF(SUM(pe.units_total), 0)              AS quality,
        SUM(pe.units_total * dm.ideal_cycle_secs)
            / NULLIF(SUM(pe.run_mins) * 60, 0)             AS performance
    FROM production_events pe
    JOIN dim_shift s  ON pe.shift_id  = s.shift_id
    JOIN dim_machine dm ON pe.machine_id = dm.machine_id
    GROUP BY pe.shift_id, pe.machine_id
)
SELECT
    shift_id,
    machine_id,
    ROUND(availability * 100, 1)               AS avail_pct,
    ROUND(performance * 100, 1)               AS perf_pct,
    ROUND(quality * 100, 1)                  AS qual_pct,
    ROUND(availability * performance * quality * 100, 1) AS oee_pct
FROM shift_oee
WHERE availability < 0.75   -- flag underperforming shifts
ORDER BY oee_pct;

WITH RECURSIVE org AS (
    -- Anchor: root nodes (no manager)
    SELECT
        employee_id,
        employee_name,
        manager_id,
        0                                    AS depth,
        employee_name                         AS path
    FROM employees
    WHERE manager_id IS NULL

    UNION ALL

    -- Recursive: each employee joins to their manager's row
    SELECT
        e.employee_id,
        e.employee_name,
        e.manager_id,
        o.depth + 1,
        o.path || ' > ' || e.employee_name
    FROM employees e
    JOIN org o ON e.manager_id = o.employee_id
)
SELECT employee_id, employee_name, depth, path
FROM org
ORDER BY path;

-- Sample output:
-- Alice        | depth 0 | Alice
-- Bob          | depth 1 | Alice > Bob
-- Dana         | depth 2 | Alice > Bob > Dana
-- Eve          | depth 2 | Alice > Bob > Eve
-- Carol        | depth 1 | Alice > Carol

WITH grouped AS (
    SELECT
        sensor_id, reading_value, recorded_at,
        ROW_NUMBER() OVER (PARTITION BY sensor_id ORDER BY recorded_at)
        - ROW_NUMBER() OVER (PARTITION BY sensor_id, reading_value ORDER BY recorded_at)
        AS grp_id
    FROM sensor_readings
),
consecutive_counts AS (
    SELECT
        sensor_id, reading_value, grp_id,
        COUNT(*) AS consecutive_count,
        MIN(recorded_at) AS streak_start,
        MAX(recorded_at) AS streak_end
    FROM grouped
    GROUP BY sensor_id, reading_value, grp_id
)
SELECT sensor_id, reading_value, consecutive_count, streak_start, streak_end
FROM consecutive_counts
WHERE consecutive_count >= 10
ORDER BY consecutive_count DESC;

WITH sensor_stats AS (
    -- Compute baseline stats on GOOD readings only
    SELECT
        sensor_id,
        AVG(reading_value)    AS mean_val,
        STDDEV(reading_value) AS stddev_val
    FROM sensor_readings
    WHERE dq_flag = 'GOOD'
    GROUP BY sensor_id
)
SELECT
    r.sensor_id,
    r.reading_value,
    r.recorded_at,
    ROUND(s.mean_val, 3)                                          AS mean_val,
    ROUND(ABS(r.reading_value - s.mean_val)
        / NULLIF(s.stddev_val, 0), 2)                            AS z_score,
    CASE
        WHEN ABS(r.reading_value - s.mean_val)
            / NULLIF(s.stddev_val, 0) > 3  THEN 'SUSPECT'
        ELSE 'GOOD'
    END                                                               AS dq_flag
FROM sensor_readings r
JOIN sensor_stats s ON r.sensor_id = s.sensor_id;

SELECT
    a.machine_id,
    a.event_id      AS event_a,
    b.event_id      AS event_b,
    a.start_time    AS a_start,
    a.end_time      AS a_end,
    b.start_time    AS b_start,
    b.end_time      AS b_end
FROM downtime_events a
JOIN downtime_events b
    ON  a.machine_id = b.machine_id
    AND a.event_id   < b.event_id     -- avoid self-join & duplicate pairs
    AND a.start_time < b.end_time     -- overlap condition part 1
    AND a.end_time   > b.start_time;  -- overlap condition part 2

-- Define cluster key on most common filter columns
ALTER TABLE production_events
CLUSTER BY (shift_date, machine_id);

-- Check clustering information
SELECT SYSTEM$CLUSTERING_INFORMATION('production_events');