08Testing · SQL

-- SCD2 dimension: the business key plus the interval start must be unique.
-- Zero rows = pass.
SELECT
  customer_id,
  _tech_valid_from,
  COUNT(*) AS versions
FROM gold.dim_customer
GROUP BY customer_id, _tech_valid_from
HAVING COUNT(*) > 1

SELECT CASE WHEN COUNT(*) = 0 THEN 'PASS' ELSE 'FAIL' END AS result
FROM (
  SELECT customer_id FROM gold.dim_customer
  GROUP BY customer_id, _tech_valid_from HAVING COUNT(*) > 1
)

import pathlib
import pytest

SQL_DIR = pathlib.Path(__file__).parent
CASES = sorted(SQL_DIR.glob("*.sql"))
assert CASES, f"no .sql assertions found under {SQL_DIR}"

@pytest.mark.parametrize("case", CASES, ids=[c.stem for c in CASES])
def test_assertion(spark, catalog, case):
    spark.sql(f"USE CATALOG {catalog}")
    bad = spark.sql(case.read_text()).take(20)
    assert not bad, "{}: {} violating row(s) shown, capped at 20\n{}".format(
        case.stem, len(bad), "\n".join(str(r.asDict()) for r in bad)
    )

resources:
  jobs:
    dwh_assertions:
      name: "dwh · SQL assertions · ${bundle.target}"
      tasks:
        - task_key: assertions
          # NO environment_key here. Serverless notebook tasks are reported to
          # reject it — register key serverless-environment-key, status hot and
          # explicitly UNVERIFIED. Reproduce on the client workspace before you
          # either rely on it or design around it.
          notebook_task:
            # thin wrapper: runs pytest over tests/sql and writes the JUnit XML
            # to the ci Volume. All the logic is in test_assertions.py above.
            notebook_path: ../tests/sql/run_assertions.py
            base_parameters:
              catalog: ${var.catalog}
      schedule:
        quartz_cron_expression: "0 30 5 * * ?"
        timezone_id: "Europe/Zurich"
      email_notifications:
        on_failure: ["${var.oncall_rota}"]

SELECT
  order_id,
  line_number,
  COUNT(*)                        AS rows_at_grain,
  COUNT(DISTINCT order_line_sk)   AS distinct_sks
FROM gold.fct_order_line
GROUP BY order_id, line_number
HAVING COUNT(*) > 1

SELECT
  f.order_line_sk,
  f.customer_sk,
  f.order_ts
FROM gold.fct_order_line f
LEFT JOIN gold.dim_customer d
       ON d.customer_sk = f.customer_sk
      AND f.order_ts >= d._tech_valid_from
      AND (d._tech_valid_to IS NULL OR f.order_ts < d._tech_valid_to)
-- IS DISTINCT FROM, not <>: a NULL customer_sk must be REPORTED, and
-- f.customer_sk <> 'UNKNOWN' is NULL on exactly those rows.
WHERE f.customer_sk IS DISTINCT FROM 'UNKNOWN'
  AND d.customer_sk IS NULL

SELECT
  country,
  COUNT(*) AS affected_rows,
  MIN(customer_id) AS example_customer
FROM gold.dim_customer
WHERE country IS NULL
   OR country NOT IN ('DE', 'AT', 'CH', 'NL')
GROUP BY country

-- Intervals are [_tech_valid_from, _tech_valid_to) — end is EXCLUSIVE.
-- The asymmetric join reports each offending pair once, not twice.
SELECT
  a.customer_id,
  a._tech_valid_from AS earlier_from,
  a._tech_valid_to   AS earlier_to,
  b._tech_valid_from AS later_from
FROM gold.dim_customer a
JOIN gold.dim_customer b
  ON  b.customer_id       = a.customer_id
  AND b._tech_valid_from  > a._tech_valid_from
WHERE a._tech_valid_to IS NULL
   OR b._tech_valid_from < a._tech_valid_to

SELECT
  customer_id,
  COUNT_IF(_tech_is_current)                            AS flagged_current,
  COUNT_IF(_tech_valid_to IS NULL)                      AS open_intervals,
  COUNT_IF(_tech_is_current AND _tech_valid_to IS NULL) AS both_on_same_row
FROM gold.dim_customer
GROUP BY customer_id
HAVING COUNT_IF(_tech_is_current)                            <> 1
    OR COUNT_IF(_tech_valid_to IS NULL)                      <> 1
    OR COUNT_IF(_tech_is_current AND _tech_valid_to IS NULL) <> 1

SELECT
  order_line_sk,
  currency,
  fx_rate,
  quantity,
  unit_price,
  amount_eur,
  status,
  ROUND(quantity * unit_price * fx_rate, 2) AS expected_eur
FROM gold.fct_order_line
WHERE amount_eur IS DISTINCT FROM ROUND(quantity * unit_price * fx_rate, 2)
   OR (currency = 'EUR' AND fx_rate IS DISTINCT FROM 1)
   OR (status IS DISTINCT FROM 'CANCELLED' AND COALESCE(amount_eur, 0) <= 0)
   OR currency IS NULL
   OR currency NOT IN ('EUR', 'CHF')

WITH silver_month AS (
  SELECT
    DATE_TRUNC('MONTH', o.order_ts)                       AS month,
    -- EUR is the base currency and is absent from the ECB file, so it is 1 by
    -- definition. Rule 2 rounds PER LINE; rounding the sum drifts by cents.
    SUM(ROUND(l.quantity * l.unit_price *
              CASE WHEN l.currency = 'EUR' THEN CAST(1 AS DECIMAL(18,6))
                   ELSE r.fx_rate END, 2))                AS silver_eur,
    -- a missing rate makes BOTH sides smaller by the same amount, so the delta
    -- stays zero. This column is the only place that defect is visible.
    COUNT_IF(l.currency IS DISTINCT FROM 'EUR'
             AND r.fx_rate IS NULL)                       AS lines_without_rate
  FROM silver.sales_order_line l
  JOIN silver.sales_order      o ON o.order_id  = l.order_id
  LEFT JOIN silver.fx_rate     r ON r.currency  = l.currency
                                AND r.rate_date = DATE(o.order_ts)
  WHERE l.status IS DISTINCT FROM 'CANCELLED'
  GROUP BY 1
),
gold_month AS (
  SELECT
    DATE_TRUNC('MONTH', order_ts) AS month,
    SUM(amount_eur)               AS gold_eur
  FROM gold.fct_order_line
  WHERE status IS DISTINCT FROM 'CANCELLED'
  GROUP BY 1
)
SELECT
  COALESCE(s.month, g.month)                            AS month,
  s.silver_eur,
  g.gold_eur,
  COALESCE(g.gold_eur, 0) - COALESCE(s.silver_eur, 0)   AS delta_eur,
  COALESCE(s.lines_without_rate, 0)                     AS lines_without_rate
FROM silver_month s
FULL OUTER JOIN gold_month g ON g.month = s.month
WHERE ABS(COALESCE(g.gold_eur, 0) - COALESCE(s.silver_eur, 0)) > 0.01
   OR COALESCE(s.lines_without_rate, 0) > 0

WITH rebuilt AS (
  SELECT
    -- make_sk() verbatim: trim, upper, coalesce to the sentinel, concat_ws,
    -- sha2. Any deviation and EVERY row reports as both missing and extra.
    sha2(concat_ws('||', COALESCE(UPPER(TRIM(customer_id)), '~')), 256)
                                        AS customer_sk,
    customer_id,
    customer_name,
    country,
    customer_segment,
    CAST(_tech_valid_from AS TIMESTAMP) AS _tech_valid_from,
    CAST(_tech_valid_to   AS TIMESTAMP) AS _tech_valid_to
  FROM silver.customer
),
deployed AS (
  SELECT
    customer_sk, customer_id, customer_name, country, customer_segment,
    _tech_valid_from, _tech_valid_to
  FROM gold.dim_customer
)
SELECT 'missing_from_gold' AS side, *
FROM (SELECT * FROM rebuilt  EXCEPT ALL SELECT * FROM deployed)
UNION ALL
SELECT 'extra_in_gold' AS side, *
FROM (SELECT * FROM deployed EXCEPT ALL SELECT * FROM rebuilt)

-- same WITH rebuilt AS (...), deployed AS (...) header as above
SELECT d.customer_sk, d._tech_valid_from
FROM deployed d
LEFT JOIN rebuilt r
       ON  r.customer_sk       IS NOT DISTINCT FROM d.customer_sk
       AND r._tech_valid_from  IS NOT DISTINCT FROM d._tech_valid_from
       AND r.customer_segment  IS NOT DISTINCT FROM d.customer_segment
-- customer_sk is a hash and is never NULL, which is the ONLY reason it can
-- serve as the did-it-match sentinel. Pick a nullable column here and the
-- anti-join reports matched rows as missing.
WHERE r.customer_sk IS NULL

WITH from_view AS (
  SELECT
    country,
    CAST(month AS DATE)                      AS month,
    CAST(MEASURE(revenue) AS DECIMAL(18,2))  AS revenue,
    CAST(MEASURE(order_count) AS BIGINT)     AS order_count
  FROM gold.mv_sales
  GROUP BY country, month
),
oracle AS (
  SELECT
    c.country,
    CAST(DATE_TRUNC('MONTH', f.order_ts) AS DATE)  AS month,
    CAST(SUM(f.amount_eur) AS DECIMAL(18,2))       AS revenue,
    CAST(COUNT(DISTINCT f.order_id) AS BIGINT)     AS order_count
  FROM gold.fct_order_line f
  JOIN gold.dim_customer c
    ON  c.customer_sk = f.customer_sk
    -- the view joins the CURRENT version (see the mv_sales YAML), so the
    -- oracle does too. Join on the interval here and the oracle is red for
    -- every customer who ever moved country — a real question, but a
    -- different assertion, not this one.
    AND c._tech_is_current
  WHERE f.status IS DISTINCT FROM 'CANCELLED'
  GROUP BY 1, 2
)
SELECT 'metric_view_only' AS side, *
FROM (SELECT * FROM from_view EXCEPT ALL SELECT * FROM oracle)
UNION ALL
SELECT 'oracle_only' AS side, *
FROM (SELECT * FROM oracle EXCEPT ALL SELECT * FROM from_view)