SQL for Data Engineers: Hard Problems Across Industries

WITH ordered AS (
    SELECT *,
           MAX(end_date) OVER (PARTITION BY member_id ORDER BY start_date, end_date
                               ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS covered_until
    FROM member_eligibility
),
grouped AS (
    SELECT *,
           SUM(CASE WHEN covered_until IS NULL OR start_date > covered_until + 1 THEN 1 ELSE 0 END)
               OVER (PARTITION BY member_id ORDER BY start_date, end_date
                     ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS period_no
    FROM ordered
)
SELECT member_id, period_no, MIN(start_date) AS coverage_start, MAX(end_date) AS coverage_end
FROM grouped
GROUP BY member_id, period_no
ORDER BY member_id, period_no;

• covered_until is the running maximum end date of all earlier spans. It has to be a maximum, not LAG(end_date): M1's March span ends before the January span does, and comparing the July span with March's end would wrongly start a new period.
• A new period starts when a span begins more than one day after covered_until. The + 1 merges touching spans (June 30 then July 1); drop it if touching spans should stay separate.
• A running sum of the start flags numbers the periods, and grouping by that number collapses each into one continuous range: the same islands trick as sessionization (post 04, Q57).

Engineer check. Grain: one row per continuous coverage period. NULL: an open-ended span with a NULL end_date breaks the running maximum; store 9999-12-31 instead.

WITH latest AS (
    SELECT *
    FROM claims
    QUALIFY ROW_NUMBER() OVER (PARTITION BY claim_id, line_no ORDER BY version DESC) = 1
)
SELECT claim_id, SUM(billed_amt) AS total_billed, SUM(paid_amt) AS total_paid
FROM latest
WHERE status <> 'VOID'
GROUP BY claim_id
ORDER BY claim_id;

• Latest version first: ROW_NUMBER() … ORDER BY version DESC = 1 keeps one row per claim line.
• Then filter VOID. Filtering before picking the latest would resurrect earlier versions: CL-1 line 2's latest version is VOID, and filtering first would fall back to its version 1 and count $50 paid that was reversed. CL-2 line 1 was voided then re-adjudicated, and its latest version is PAID.
• This "latest version, then the business filter" order applies to any versioned source: orders, policies, contracts.

Engineer check. Grain: one row per claim. NULL: a NULL status passes <> 'VOID' as unknown and is dropped; use IS DISTINCT FROM 'VOID' if NULL statuses should count.

In Snowflake:

-- Snowflake buckets timestamps with TIME_SLICE
SELECT device_id, TIME_SLICE(reading_ts, 5, 'MINUTE') AS bucket_start, AVG(value) AS avg_value
FROM sensor_reading
GROUP BY ALL;

The same logic in portable SQL, as run on DuckDB to produce the output above:

WITH buckets AS (
    SELECT b AS bucket_start
    FROM generate_series(TIMESTAMP '2025-03-01 10:00', TIMESTAMP '2025-03-01 10:25', INTERVAL 5 MINUTE) t(b)
),
expected AS (
    SELECT d.device_id, b.bucket_start
    FROM (SELECT DISTINCT device_id FROM sensor_reading) d CROSS JOIN buckets b
),
actual AS (
    SELECT device_id, time_bucket(INTERVAL 5 MINUTE, reading_ts) AS bucket_start,
           ROUND(AVG(value), 2) AS avg_value, COUNT(*) AS readings
    FROM sensor_reading
    GROUP BY ALL
)
SELECT e.device_id, e.bucket_start, a.avg_value, COALESCE(a.readings, 0) AS readings
FROM expected e
LEFT JOIN actual a USING (device_id, bucket_start)
ORDER BY e.device_id, e.bucket_start;

• Bucketing floors each timestamp to its 5-minute slot: time_bucket in DuckDB, TIME_SLICE in Snowflake (docs), date_bin in Postgres. Averaging per slot downsamples the stream to a fixed grain.
• Silence isn't in the data. A device that sent nothing has no rows to group, so a GROUP BY can never report it. The expected grid (every device × every bucket) defines what should exist, and a LEFT JOIN shows the gaps: D1 was silent from 10:10 to 10:20 and again from 10:25.
• readings = 0 marks a silent bucket explicitly, which is easier to alert on than a NULL average.

Engineer check. Grain: one row per device per expected bucket. NULL: a silent bucket's average is NULL, which is honest; don't fill it with 0, which is a real sensor value.

WITH marked AS (
    SELECT *,
           CASE WHEN status IS DISTINCT FROM LAG(status) OVER (PARTITION BY machine_id ORDER BY status_ts)
                THEN 1 ELSE 0 END AS is_change
    FROM machine_status
),
runs AS (
    SELECT *,
           SUM(is_change) OVER (PARTITION BY machine_id ORDER BY status_ts
                                ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS run_no
    FROM marked
),
intervals AS (
    SELECT machine_id, run_no, status, MIN(status_ts) AS state_start
    FROM runs GROUP BY ALL
)
SELECT machine_id, status, state_start,
       LEAD(state_start) OVER (PARTITION BY machine_id ORDER BY state_start) AS state_end,
       datediff('minute', state_start,
                LEAD(state_start) OVER (PARTITION BY machine_id ORDER BY state_start)) AS minutes
FROM intervals
ORDER BY state_start;

• is_change is 1 where the status differs from the previous reading. IS DISTINCT FROM makes the first row (previous is NULL) count as a change.
• A running sum of the change flags gives every row its run number, constant within a stretch of the same status: the islands pattern again, with the change flag as the boundary.
• End at the next start, not the last reading. The first RUN stretch's last reading is 08:10, but the machine was running until it went idle at 08:15. Using MAX(status_ts) per run would report 10 minutes of running instead of 15, and a one-reading state such as the RUN at 08:25 would last 0 minutes. LEAD(state_start) closes each state when the next begins.
• The last state (DOWN) is still in progress, so its end is NULL. Fill it with the reporting time if you need a duration.

Engineer check. Grain: one row per state interval. NULL: the current state's end is NULL by design; datediff returns NULL for it.

WITH hours AS (
    SELECT h FROM generate_series(TIMESTAMP '2025-03-01 08:00', TIMESTAMP '2025-03-01 10:00', INTERVAL 1 HOUR) t(h)
),
events AS (
    SELECT start_ts AS ts, +1 AS delta FROM trips
    UNION ALL SELECT end_ts, -1 FROM trips
    UNION ALL SELECT h, 0 FROM hours                 -- a marker at every hour start
),
running AS (
    SELECT ts, delta,
           SUM(delta) OVER (ORDER BY ts, delta
                            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS in_progress
    FROM events
)
SELECT date_trunc('hour', ts) AS hour, MAX(in_progress) AS peak_concurrent
FROM running
WHERE ts >= TIMESTAMP '2025-03-01 08:00' AND ts < TIMESTAMP '2025-03-01 11:00'
GROUP BY ALL
ORDER BY hour;

• The sweep line: every trip becomes two events, +1 at its start and −1 at its end. Sorted by time, the running sum is the number of trips in progress at each moment, and the peak of an hour is the largest running sum inside it. One sort instead of comparing every trip with every other.
• ORDER BY ts, delta puts −1 before +1 at the same instant, which implements half-open intervals: T1 ending at 08:40 and T3 starting at 08:40 don't overlap. The tie order is the interval rule.
• Hour markers (delta = 0) carry the running sum into every hour. Without them, an hour where nothing starts or ends has no events at all and disappears from the result, even though T4 was in progress for all of 10:00–11:00. Worse, an hour whose only event is an end under-reports: 09:00 would show 1, because the moment of peak concurrency (the hour start, with T3 and T4 both running) has no row. With a marker, 09:00 reads 2 and 10:00 reads 1.

Engineer check. Grain: one row per hour. NULL: a trip with no end time never subtracts; give in-progress trips the reporting time as their end.

WITH t AS (
    SELECT *, CAST(start_ts AS DATE) AS d,
           datediff('minute', start_ts, end_ts) AS drive_min,
           LAG(end_ts) OVER (PARTITION BY driver_id, CAST(start_ts AS DATE) ORDER BY start_ts) AS prev_end
    FROM trips
)
SELECT driver_id, d,
       SUM(drive_min)                                                     AS driving_min,
       SUM(COALESCE(datediff('minute', prev_end, start_ts), 0))           AS idle_min,
       ROUND(100.0 * SUM(drive_min)
             / NULLIF(SUM(drive_min) + SUM(COALESCE(datediff('minute', prev_end, start_ts), 0)), 0), 1) AS utilisation_pct
FROM t
GROUP BY driver_id, d
ORDER BY driver_id, d;

• LAG(end_ts) per driver and day gives the previous trip's end; idle is the gap to this trip's start. DR1 waited 15 and then 60 minutes.
• The first trip has no previous trip, so its gap is NULL, treated as 0. Time before the first trip and after the last isn't counted, which is the usual convention: utilisation within the working span.
• NULLIF guards the division; DR2 has one trip and no idle time, so 100%.
• Overlapping trips (a data error) would give negative gaps. Check prev_end <= start_ts in a quality test rather than silently netting them.

Engineer check. Grain: one row per driver per day. NULL: the first trip's NULL gap becomes 0 explicitly.

WITH daily AS (
    SELECT subscriber_id, CAST(date_trunc('month', call_ts) AS DATE) AS month, CAST(call_ts AS DATE) AS d,
           SUM(duration_sec) / 60.0 AS minutes
    FROM cdr GROUP BY ALL
),
ladder AS (
    SELECT *,
           SUM(minutes) OVER w - minutes AS from_min,      -- usage before this day
           SUM(minutes) OVER w           AS to_min         -- usage after this day
    FROM daily
    WINDOW w AS (PARTITION BY subscriber_id, month ORDER BY d ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)
),
tiers AS (
    SELECT *,
           GREATEST(0, LEAST(to_min, 200) - GREATEST(from_min, 100)) AS tier2_min,
           GREATEST(0, to_min - GREATEST(from_min, 200))             AS tier3_min
    FROM ladder
)
SELECT subscriber_id, month,
       SUM(minutes)                                          AS minutes,
       ROUND(SUM(tier2_min) * 0.05 + SUM(tier3_min) * 0.10, 2) AS charge
FROM tiers
GROUP BY ALL
ORDER BY subscriber_id;

• The ladder: a running total over the month gives each day a slice [from_min, to_min) of cumulative usage. S1's second day covers minutes 60 to 140.
• Clamping a slice to a tier: LEAST(to_min, cap) − GREATEST(from_min, floor), floored at 0, is the part of the slice inside the tier [floor, cap). S1's second day has 40 minutes in the free tier and 40 in tier 2; its third day has 60 in tier 2 and 40 in tier 3.
• S1: tier 2 is 100 minutes (5.00), tier 3 is 40 minutes (4.00), so 9.00. S2 stays in the free tier.
• Why not price each day on its own: a day doesn't know which tier it starts in. Only the running total does.

GREATEST and LEAST are safe here because no argument can be NULL; remember Snowflake's NULL behaviour (post 02, Q27) if one could.

Engineer check. Grain: one row per subscriber per month. NULL: a NULL duration_sec is skipped by SUM, so that call is free; reject it in quality checks instead.

SELECT c.conversion_id, c.user_id, c.value,
       i.campaign_id AS attributed_campaign, i.imp_ts
FROM conversions c
LEFT JOIN impressions i
       ON i.user_id = c.user_id
      AND i.imp_ts <= c.conv_ts
      AND i.imp_ts >  c.conv_ts - INTERVAL 7 DAY
QUALIFY ROW_NUMBER() OVER (PARTITION BY c.conversion_id
                           ORDER BY i.imp_ts DESC NULLS LAST, i.campaign_id) = 1
ORDER BY c.conversion_id;

• The range join pairs each conversion with that user's impressions in (conv − 7 days, conv].
• QUALIFY ROW_NUMBER() … = 1 per conversion keeps the most recent one, so each conversion is credited exactly once and revenue isn't double-counted across campaigns.
• The tiebreak: u3 saw two campaigns at the same second. Without campaign_id in the ORDER BY, which one wins can change between runs, and so would the campaign report. Any deterministic rule is better than none; agree it with the business.
• LEFT JOIN keeps CV2: u2's only impression is 14 days old, so the conversion stays, unattributed, with a NULL campaign. An inner join would drop $50 of revenue from the report.
• Engines with an as-of join (Q85) can find the latest impression directly, then check it's within 7 days.

Engineer check. Grain: one row per conversion. NULL: unattributed conversions have NULL campaigns; report them as their own bucket.

WITH imp AS (SELECT DISTINCT imp_id, campaign_id FROM impressions),
clk AS (SELECT DISTINCT click_id, imp_id FROM clicks)
SELECT i.campaign_id,
       COUNT(DISTINCT i.imp_id)                                       AS impressions,
       COUNT(DISTINCT c.click_id)                                     AS clicks,
       ROUND(100.0 * COUNT(DISTINCT c.click_id) / COUNT(DISTINCT i.imp_id), 1) AS ctr_pct,
       (SELECT COUNT(*) FROM clk WHERE imp_id NOT IN (SELECT imp_id FROM imp)) AS orphan_clicks_all_campaigns
FROM imp i
LEFT JOIN clk c ON c.imp_id = i.imp_id
GROUP BY i.campaign_id
ORDER BY i.campaign_id;

• Deduplicate both sides first. Duplicate impression I2 inflates the denominator, and duplicate click K1 inflates the numerator. Deduplicating after the join can't undo the fan-out a duplicate causes.
• LEFT JOIN from impressions keeps impressions nobody clicked; they're most of the denominator.
• Orphan clicks (K3 points at an impression that never arrived) can't be credited to a campaign. Counting them separately turns a silent loss into a data-quality signal: bots, lost impression logs or a key mismatch.
• NOT IN is safe here only because imp_id is never NULL in the deduplicated set; with a nullable key, use NOT EXISTS (post 01, Q14).

Engineer check. Grain: one row per campaign. NULL: a click with a NULL imp_id matches no impression, and NULL NOT IN (…) is unknown, so the orphan count silently skips it too. Count clicks with NULL keys as their own quality metric.

WITH milestones AS (
    SELECT shipment_id,
           MIN(CASE WHEN event_type = 'PICKED'    THEN event_ts END) AS picked_ts,
           MIN(CASE WHEN event_type = 'DELIVERED' THEN event_ts END) AS delivered_ts
    FROM shipment_event
    GROUP BY shipment_id
)
SELECT m.shipment_id, m.picked_ts, m.delivered_ts,
       datediff('hour', m.picked_ts, m.delivered_ts) AS transit_hours,
       l.sla_hours,
       CASE WHEN m.delivered_ts IS NULL                                     THEN 'OPEN'
            WHEN datediff('hour', m.picked_ts, m.delivered_ts) > l.sla_hours THEN 'BREACHED'
            ELSE 'MET' END                                                  AS sla_status
FROM milestones m
JOIN shipment s USING (shipment_id)
JOIN lane_sla l ON l.origin = s.origin AND l.dest = s.dest
ORDER BY m.shipment_id;

• Pivot by type with MIN(CASE …): each milestone's time comes from the event's own timestamp, so the order events arrived in doesn't matter, and the duplicate PICKED collapses.
• The SLA comes from the lane, so join shipment → lane before comparing.
• OPEN before the comparison: SH3 hasn't been delivered, so its transit time is NULL, and NULL > 24 is unknown. Without the first branch it would fall into ELSE and be reported as MET, a shipment that hasn't arrived counted as on time. Ordering the CASE so the NULL case is handled first avoids that.

Engineer check. Grain: one row per shipment. NULL: an undelivered shipment's NULL comparison is the trap; the first CASE branch handles it.

WITH demand AS (
    SELECT sku, SUM(units) / 7.0 AS avg_daily_demand          -- divide by calendar days, not rows
    FROM sales_daily
    WHERE sale_date BETWEEN DATE '2025-03-07' - 6 AND DATE '2025-03-07'
    GROUP BY sku
)
SELECT i.sku, i.on_hand,
       ROUND(COALESCE(d.avg_daily_demand, 0), 2)                          AS avg_daily_demand,
       ROUND(i.on_hand / NULLIF(d.avg_daily_demand, 0), 1)                AS days_of_cover,
       COALESCE(i.on_hand / NULLIF(d.avg_daily_demand, 0) < 14, FALSE)    AS reorder
FROM inventory i
LEFT JOIN demand d USING (sku)
ORDER BY i.sku;

• Divide by calendar days. B sold 14 units on two of the seven days: its true demand is 28 ÷ 7 = 4 per day. The obvious window function, AVG(units) OVER (ORDER BY sale_date ROWS 6 PRECEDING), averages the last seven rows, which for B are spread over weeks, and reports 14 per day, three and a half times too much (step 2). That B would be flagged for an urgent reorder it doesn't need.
• Infinite cover: C has sold nothing in the window, so NULLIF(…, 0) makes the division NULL rather than an error, and the reorder flag is COALESCEd to FALSE.
• For a rolling version over many days, use a calendar spine (post 04, Q62) or a RANGE frame of 6 days divided by 7, never a ROWS frame over sparse data.

Engineer check. Grain: one row per SKU. NULL: no demand gives NULL cover (infinite) and an explicit FALSE flag.

WITH days AS (
    SELECT CAST(d AS DATE) AS day FROM generate_series(DATE '2025-03-01', DATE '2025-03-06', INTERVAL 1 DAY) t(d)
),
daily_actual AS (                                      -- the LAST read of each day
    SELECT meter_id, CAST(read_ts AS DATE) AS day, cumulative_kwh
    FROM meter_read
    QUALIFY ROW_NUMBER() OVER (PARTITION BY meter_id, CAST(read_ts AS DATE) ORDER BY read_ts DESC) = 1
),
grid AS (
    SELECT m.meter_id, days.day, a.cumulative_kwh AS actual
    FROM (SELECT DISTINCT meter_id FROM meter_read) m CROSS JOIN days
    LEFT JOIN daily_actual a ON a.meter_id = m.meter_id AND a.day = days.day
),
bracketed AS (
    SELECT *,
           LAST_VALUE(actual IGNORE NULLS)  OVER prev AS prev_val,
           LAST_VALUE(CASE WHEN actual IS NOT NULL THEN day END IGNORE NULLS) OVER prev AS prev_day,
           FIRST_VALUE(actual IGNORE NULLS) OVER next AS next_val,
           FIRST_VALUE(CASE WHEN actual IS NOT NULL THEN day END IGNORE NULLS) OVER next AS next_day
    FROM grid
    WINDOW prev AS (PARTITION BY meter_id ORDER BY day ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW),
           next AS (PARTITION BY meter_id ORDER BY day ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING)
)
SELECT meter_id, day,
       ROUND(COALESCE(actual,
             prev_val + (next_val - prev_val) * (day - prev_day) / (next_day - prev_day)), 1) AS kwh,
       actual IS NULL AS is_estimated
FROM bracketed
ORDER BY day;

• One reading per day first. 1 March has two reads; joining both to the calendar would give 1 March two rows. Keeping the last read of the day matches "end-of-day reading".
• The spine gives every day a row; missing days have a NULL actual.
• LAST_VALUE(… IGNORE NULLS) over the rows up to the current one finds the previous actual read; FIRST_VALUE(… IGNORE NULLS) over the rows from the current one finds the next. The same trick on the date gives how far apart they are.
• Linear interpolation: prev + (next − prev) × (days since prev ÷ days between). 3 March is one day into a three-day gap from 1,030 to 1,090, so 1,050.
• is_estimated flags the synthetic values. Billing and analytics must be able to tell measured from estimated.
• Days before the first read or after the last have no bracket, so they stay NULL rather than being extrapolated.

Engineer check. Grain: one row per meter per day. NULL: unbracketed days stay NULL on purpose; IGNORE NULLS is what lets the window skip the gaps.

In Snowflake:

SELECT t.trade_id, t.symbol, t.trade_ts, q.quote_ts, q.bid, q.ask
FROM trades t
ASOF JOIN quotes q
  MATCH_CONDITION (t.trade_ts >= q.quote_ts)
  ON t.symbol = q.symbol;

The same logic in portable SQL, as run on DuckDB to produce the output above:

SELECT t.trade_id, t.symbol, t.trade_ts, q.quote_ts, q.bid, q.ask
FROM trades t
ASOF LEFT JOIN quotes q
  ON q.symbol = t.symbol
 AND t.trade_ts >= q.quote_ts
ORDER BY t.trade_id;

• An as-of join matches each trade to the quote with the greatest quote_ts that is <= the trade time, per symbol. DuckDB writes the inequality inside ON (docs); Snowflake puts it in MATCH_CONDITION (docs).
• >= includes a quote at the same instant: TR4 at 09:30:10 matches the quote from 09:30:10 exactly. Use > if a simultaneous quote shouldn't count.
• The left join keeps TR3: BOLT's first quote comes after the trade, so there's no prevailing quote and bid/ask are NULL.
• Why not an equality join: quotes and trades tick at different instants, so quote_ts = trade_ts matches almost nothing. Without an as-of join, the portable alternative is a range join plus QUALIFY ROW_NUMBER() … ORDER BY quote_ts DESC = 1 (Q80), which is far more expensive on large tables.

Engineer check. Grain: one row per trade. NULL: trades before any quote have NULL bid and ask.

In Snowflake:

SELECT match_key,
       ARRAY_AGG(full_name) WITHIN GROUP (ORDER BY completeness_score DESC, updated_at DESC)[0]::STRING AS full_name,
       ARRAY_AGG(email)     WITHIN GROUP (ORDER BY completeness_score DESC, updated_at DESC)[0]::STRING AS email,
       ARRAY_AGG(phone)     WITHIN GROUP (ORDER BY completeness_score DESC, updated_at DESC)[0]::STRING AS phone
FROM customer_source
GROUP BY match_key;
-- Snowflake's ARRAY_AGG leaves NULL values out, so [0] is the best non-NULL value.

The same logic in portable SQL, as run on DuckDB to produce the output above:

SELECT match_key,
       (array_agg(full_name ORDER BY completeness_score DESC, updated_at DESC) FILTER (WHERE full_name IS NOT NULL))[1] AS full_name,
       (array_agg(email     ORDER BY completeness_score DESC, updated_at DESC) FILTER (WHERE email     IS NOT NULL))[1] AS email,
       (array_agg(phone     ORDER BY completeness_score DESC, updated_at DESC) FILTER (WHERE phone     IS NOT NULL))[1] AS phone,
       string_agg(DISTINCT source_system, ', ' ORDER BY source_system)                                                 AS sources
FROM customer_source
GROUP BY match_key
ORDER BY match_key;

• Field-level survivorship: each field is ranked independently. The CRM record is the most complete, so it wins the name and email, but it has no phone; the phone comes from the best record that has one. Between Billing and Support, both scored 70, the more recent Support record wins.
• FILTER (WHERE field IS NOT NULL) removes NULLs before ranking, so a NULL from the top record never wins. Snowflake's ARRAY_AGG already omits NULLs (docs), and its arrays are 0-indexed where DuckDB's are 1-indexed.
• Picking one whole row (QUALIFY ROW_NUMBER() … = 1) would give MK-1 a NULL phone even though two systems know it.
• Keep lineage: sources records which systems contributed. Real MDM also stores which source won each field, so a steward can explain or override it.

Engineer check. Grain: one row per match_key. NULL: a field NULL in every source stays NULL; the FILTER only stops NULLs from beating real values.