SQL-Based Data Validation and Anomaly Detection Questions

WITH latest AS (
  SELECT * FROM snapshots
  WHERE snapshot_date = (SELECT MAX(snapshot_date) FROM snapshots)
),
prev AS (
  SELECT * FROM snapshots
  WHERE snapshot_date = (SELECT MAX(snapshot_date) FROM snapshots WHERE snapshot_date < (SELECT MAX(snapshot_date) FROM snapshots))
),
all_cols AS (
  SELECT COALESCE(l.schema_name, p.schema_name) AS schema_name,
         COALESCE(l.table_name, p.table_name) AS table_name,
         COALESCE(l.column_name, p.column_name) AS column_name,
         l.data_type AS new_data_type, p.data_type AS old_data_type,
         l.is_nullable AS new_nullable, p.is_nullable AS old_nullable
  FROM latest l
  FULL OUTER JOIN prev p
    ON l.schema_name = p.schema_name AND l.table_name = p.table_name AND l.column_name = p.column_name
)
SELECT schema_name, table_name, column_name,
  CASE
    WHEN old_data_type IS NULL THEN 'ADDED'
    WHEN new_data_type IS NULL THEN 'REMOVED'
    WHEN new_data_type <> old_data_type THEN CONCAT('TYPE_CHANGED: ', old_data_type, ' -> ', new_data_type)
    WHEN new_nullable <> old_nullable THEN CONCAT('NULLABILITY_CHANGED: ', old_nullable, ' -> ', new_nullable)
    ELSE 'UNCHANGED'
  END AS change_type
FROM all_cols
WHERE (old_data_type IS NULL OR new_data_type IS NULL OR new_data_type <> old_data_type OR new_nullable <> old_nullable)
ORDER BY schema_name, table_name, column_name;

SELECT
  normalized_name,
  email,
  COUNT(*) AS duplicate_count,
  ARRAY_AGG(user_id ORDER BY created_at) AS example_user_ids
FROM users
GROUP BY normalized_name, email
HAVING COUNT(*) > 1
ORDER BY duplicate_count DESC;

ARRAY_AGG(user_id ORDER BY created_at)[:5]  -- PostgreSQL supports slicing on arrays

import pandas as pd
import numpy as np
from scipy import stats
from sklearn.metrics import pairwise_distances

def drift_checks(raw, clean, col):
    a = raw[col].dropna()
    b = clean[col].dropna()
    stats_summary = {
        'raw_n': len(a),'clean_n': len(b),
        'raw_mean': a.mean(),'clean_mean': b.mean(),
        'raw_std': a.std(),'clean_std': b.std(),
        'raw_median': a.median(),'clean_median': b.median()
    }
    # Normality
    p_raw = stats.shapiro(a.sample(min(5000,len(a)))).pvalue
    p_clean = stats.shapiro(b.sample(min(5000,len(b)))).pvalue
    # Choose test
    if p_raw>0.05 and p_clean>0.05 and stats.levene(a,b).pvalue>0.05:
        test = stats.ttest_ind(a,b, equal_var=True)
        test_name = 't-test'
    else:
        test = stats.ks_2samp(a,b)
        test_name = 'ks-test'
    # PSI (binned)
    bins = np.histogram_bin_edges(np.concatenate([a,b]), bins=10)
    ra, _ = np.histogram(a, bins=bins, density=True)
    rb, _ = np.histogram(b, bins=bins, density=True)
    eps=1e-6
    psi = np.sum((ra+eps - rb-eps) * np.log((ra+eps)/(rb+eps)))
    return stats_summary, {'test': test_name, 'stat': test.statistic, 'p': test.pvalue}, {'psi': psi}

# use thresholds: p<0.01 or PSI>0.1 -> flag

-- compute counts and moments per dataset
SELECT
  dataset,
  COUNT(val) AS n,
  AVG(val) AS mean,
  STDDEV_SAMP(val) AS std,
  PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY val) AS median
FROM my_table
WHERE dataset IN ('raw','clean')
GROUP BY dataset;
-- for PSI: create same bins and compute proportions per bin per dataset, then compute PSI in SQL or downstream.