Data Quality Rule Engine (SQL)

Replace {{PLACEHOLDERS}} with your environment values, then deploy to your stack.

-- =============================================================================
-- DATA QUALITY RULE ENGINE (SQL)
-- Plug in your schema/table/column values below, then run as a scheduled check
-- or wrap in a dbt test / observability tool query.
-- =============================================================================

-- ── CONFIGURATION: replace these values ─────────────────────────────────────
-- {{DATABASE}}     e.g. analytics
-- {{SCHEMA}}       e.g. marts
-- {{TABLE}}        e.g. fct_orders
-- {{DATE_COLUMN}}  e.g. order_date
-- {{PK_COLUMN}}    e.g. order_id

WITH config AS (
    SELECT
        '{{DATABASE}}'   AS database_name,
        '{{SCHEMA}}'     AS schema_name,
        '{{TABLE}}'      AS table_name,
        '{{DATE_COLUMN}}' AS date_column,
        '{{PK_COLUMN}}'  AS pk_column,
        24               AS freshness_sla_hours,      -- max hours since last load
        0.01             AS max_null_rate,           -- 1% allowed nulls on critical col
        0.05             AS max_duplicate_rate        -- 5% duplicate PK tolerance (0 for strict)
),

source_stats AS (
    SELECT
        COUNT(*)                                                    AS row_count,
        COUNT(DISTINCT {{PK_COLUMN}})                               AS distinct_pk_count,
        MAX({{DATE_COLUMN}})                                        AS max_date,
        SUM(CASE WHEN {{PK_COLUMN}} IS NULL THEN 1 ELSE 0 END)      AS null_pk_count,
        DATEDIFF('hour', MAX(_loaded_at), CURRENT_TIMESTAMP())      AS hours_since_load
    FROM {{DATABASE}}.{{SCHEMA}}.{{TABLE}}
    WHERE {{DATE_COLUMN}} >= DATEADD('day', -7, CURRENT_DATE())     -- last 7 days window
),

rule_results AS (
    SELECT
        c.table_name,
        'freshness_sla' AS rule_name,
        CASE WHEN s.hours_since_load <= c.freshness_sla_hours THEN 'PASS' ELSE 'FAIL' END AS status,
        s.hours_since_load AS observed_value,
        c.freshness_sla_hours AS threshold
    FROM config c CROSS JOIN source_stats s

    UNION ALL

    SELECT
        c.table_name,
        'pk_uniqueness' AS rule_name,
        CASE
            WHEN s.row_count = 0 THEN 'SKIP'
            WHEN (s.row_count - s.distinct_pk_count)::FLOAT / NULLIF(s.row_count, 0) <= c.max_duplicate_rate
            THEN 'PASS' ELSE 'FAIL'
        END,
        (s.row_count - s.distinct_pk_count)::FLOAT / NULLIF(s.row_count, 0),
        c.max_duplicate_rate
    FROM config c CROSS JOIN source_stats s

    UNION ALL

    SELECT
        c.table_name,
        'pk_completeness' AS rule_name,
        CASE
            WHEN s.row_count = 0 THEN 'SKIP'
            WHEN s.null_pk_count::FLOAT / s.row_count <= c.max_null_rate THEN 'PASS' ELSE 'FAIL'
        END,
        s.null_pk_count::FLOAT / NULLIF(s.row_count, 0),
        c.max_null_rate
    FROM config c CROSS JOIN source_stats s
)

SELECT
    table_name,
    rule_name,
    status,
    observed_value,
    threshold,
    CURRENT_TIMESTAMP() AS evaluated_at

// ... download full template for remaining code