Data Cleaning and Quality Validation in SQL Questions

SELECT 1.0 * SUM(CASE WHEN loaded_at <= expected_load_time THEN 1 ELSE 0 END) / COUNT(*) AS timely_rate
FROM dataset_loads
WHERE load_date = '2025-12-05';

SELECT 1.0 * SUM(CASE WHEN src_count != wh_count THEN 1 ELSE 0 END) / COUNT(*) AS reconciliation_fail_rate
FROM reconciliation_reports
WHERE report_date = '2025-12-05';

SELECT 1.0 * COUNT(DISTINCT table_name) FILTER (WHERE change_detected = TRUE) / COUNT(DISTINCT table_name) AS schema_drift_rate
FROM schema_events
WHERE event_date = '2025-12-05';

SELECT column_name,
  1.0 * SUM(CASE WHEN col IS NULL THEN 1 ELSE 0 END)/COUNT(*) AS null_rate
FROM analytics_table
GROUP BY column_name
HAVING is_critical = TRUE;

SELECT 1.0 * SUM(cnt-1) / SUM(cnt) AS duplicate_rate
FROM (
  SELECT event_id, COUNT(*) AS cnt
  FROM events
  GROUP BY event_id
) t;

SELECT
  percentile_cont(0.5) WITHIN GROUP (ORDER BY EXTRACT(EPOCH FROM (now()-ingest_time))/60) AS median_min,
  percentile_cont(0.95) WITHIN GROUP (ORDER BY EXTRACT(EPOCH FROM (now()-ingest_time))/60) AS p95_min
FROM table_ingest_times;

SELECT column_name, js_divergence
FROM feature_drift_metrics
WHERE event_date = '2025-12-05' AND js_divergence > 0.2;

SELECT event_type, 1.0 * COUNT(*) FILTER (WHERE required_fields_present) / COUNT(*) AS completeness
FROM events
GROUP BY event_type;

CREATE MATERIALIZED VIEW mv_user_events_daily AS
SELECT user_id, event_date, COUNT(*) AS events, SUM(value) AS total_value
FROM raw.events
GROUP BY user_id, event_date;
-- nightly: REFRESH MATERIALIZED VIEW CONCURRENTLY mv_user_events_daily;

INSERT INTO validation_failures (run_ts, id, reason)
SELECT now(), e.id, 'missing_fk'
FROM raw.events e
LEFT JOIN dim.users u ON e.user_id = u.id
WHERE e.event_ts > (SELECT last_validated_ts FROM validate_meta)
  AND e.event_dt >= current_date - INTERVAL '7 days';
-- update meta afterwards

SELECT COUNT(*) FROM raw.events
WHERE event_date BETWEEN '2025-12-01' AND '2025-12-02'  -- prunes partitions
  AND event_type = 'purchase';

-- 1% sample
SELECT user_id, SUM(value) FROM raw.events TABLESAMPLE BERNOULLI (1)
WHERE event_date >= '2025-12-01'
GROUP BY user_id;

SELECT approx_count_distinct(user_id) FROM raw.events WHERE event_date = '2025-12-01';
-- or use t-digest/approx_percentile for latency-sensitive quantiles

WITH t1_hash AS (
  SELECT id, MD5(CONCAT_WS('#', a, b, c)) AS row_hash, a,b,c FROM t1
),
t2_hash AS (
  SELECT id, MD5(CONCAT_WS('#', a, b, c)) AS row_hash, a,b,c FROM t2
),
joined AS (
  SELECT
    COALESCE(t1_hash.id, t2_hash.id) AS id,
    t1_hash.row_hash AS old_hash,
    t2_hash.row_hash AS new_hash,
    t1_hash.a AS old_a, t2_hash.a AS new_a,
    t1_hash.b AS old_b, t2_hash.b AS new_b,
    t1_hash.c AS old_c, t2_hash.c AS new_c
  FROM t1_hash
  FULL OUTER JOIN t2_hash USING (id)
)
SELECT
  id,
  CASE
    WHEN old_hash IS NULL THEN 'added'
    WHEN new_hash IS NULL THEN 'removed'
    WHEN old_hash <> new_hash THEN 'changed'
    ELSE 'unchanged'
  END AS change_type,
  -- per-column diffs as JSON (Postgres syntax)
  CASE WHEN old_hash <> new_hash THEN jsonb_strip_nulls(jsonb_build_object(
    'a', CASE WHEN old_a IS DISTINCT FROM new_a THEN concat(old_a, ' -> ', new_a) END,
    'b', CASE WHEN old_b IS DISTINCT FROM new_b THEN concat(old_b, ' -> ', new_b) END,
    'c', CASE WHEN old_c IS DISTINCT FROM new_c THEN concat(old_c, ' -> ', new_c) END
  )) END AS column_diffs
FROM joined
WHERE NOT (old_hash IS NOT NULL AND new_hash IS NOT NULL AND old_hash = new_hash);

-- Compare today vs 7-day median
WITH today AS (SELECT COUNT(*) cnt FROM raw.my_table WHERE dt = CURRENT_DATE),
hist AS (SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY cnt) median_cnt
         FROM (SELECT dt, COUNT(*) cnt FROM raw.my_table WHERE dt BETWEEN CURRENT_DATE - INTERVAL '14 days' AND CURRENT_DATE - INTERVAL '1 day' GROUP BY dt) t)
SELECT CASE WHEN today.cnt < hist.median_cnt * 0.7 THEN 'ALERT' ELSE 'OK' END status FROM today, hist;

SELECT SUM(CASE WHEN user_id IS NULL THEN 1 ELSE 0 END)/COUNT(*) AS null_rate
FROM raw.events WHERE dt = CURRENT_DATE;

SELECT column_name FROM information_schema.columns
WHERE table_name='my_table' AND column_name NOT IN (SELECT name FROM expected_schema);

# example: restart job via orchestration API
import requests
resp = requests.post(f"{ORCH_URL}/jobs/{job_id}/run", headers={'Authorization':TOKEN})