The four checks

-- Two clocks, deliberately. A full refresh that rewrites yesterday's rows
-- keeps _tech_loaded_at current while the newest order is 40 hours old.
--
-- 26 h = 24 (run to run) + 1 (early finish one night, late the next)
--      + 1 (October: two 03:00 Europe/Berlin starts are 25 h apart).
SELECT
  'gold.fct_order_line' AS object_name,
  MAX(_tech_loaded_at)  AS last_write,
  MAX(order_ts)         AS newest_event,
  timestampdiff(HOUR, MAX(_tech_loaded_at), current_timestamp()) AS write_age_h,
  timestampdiff(HOUR, MAX(order_ts),        current_timestamp()) AS event_age_h
FROM dwh_prod.gold.fct_order_line
HAVING timestampdiff(HOUR, MAX(_tech_loaded_at), current_timestamp()) > 26
    OR timestampdiff(HOUR, MAX(order_ts),        current_timestamp()) > 26;

-- The calendar is the spine, not the fact table. A day on which nothing
-- landed produces no group, so a GROUP BY over the fact alone cannot emit a
-- row for the total outage this check exists to catch.
WITH calendar AS (
  SELECT dd.date AS d
  FROM dwh_prod.gold.dim_date dd
  WHERE dd.date >= current_date() - INTERVAL 40 DAYS
    AND dd.date <  current_date()
),
daily AS (
  SELECT c.d, count(f.order_line_sk) AS line_count
  FROM calendar c
  LEFT JOIN (
         SELECT date_sk, order_line_sk
         FROM dwh_prod.gold.fct_order_line
         WHERE date_sk >= current_date() - INTERVAL 40 DAYS
           AND date_sk <  current_date()
       ) f
    ON f.date_sk = c.d
  GROUP BY c.d
),
baseline AS (
  SELECT t.d, t.line_count, median(h.line_count) AS median_7d
  FROM daily t
  JOIN daily h
    ON h.d >= t.d - INTERVAL 7 DAYS
   AND h.d <  t.d
  GROUP BY t.d, t.line_count
)
SELECT d, line_count, median_7d,
       round(line_count / nullif(median_7d, 0), 2) AS ratio
FROM baseline
WHERE d = current_date() - INTERVAL 1 DAY
  AND (line_count < 0.5 * median_7d OR line_count > 2.0 * median_7d);

-- Same spine problem, one dimension worse: a country that goes to zero stops
-- producing rows at all. Build the country x day grid first, join the fact in.
WITH calendar AS (
  SELECT dd.date AS d
  FROM dwh_prod.gold.dim_date dd
  WHERE dd.date >= current_date() - INTERVAL 40 DAYS
    AND dd.date <  current_date()
),
countries AS (
  SELECT DISTINCT country
  FROM dwh_prod.gold.dim_customer
  WHERE _tech_is_current
),
grid AS (
  SELECT k.country, c.d
  FROM countries k CROSS JOIN calendar c
),
fact_lines AS (
  SELECT c.country, f.date_sk AS d, f.order_line_sk
  FROM dwh_prod.gold.fct_order_line f
  JOIN dwh_prod.gold.dim_customer  c
    ON c.customer_sk = f.customer_sk
   AND c._tech_is_current   -- customer_sk is stable across SCD2 versions;
                            -- without this predicate every version fans out
  WHERE f.date_sk >= current_date() - INTERVAL 40 DAYS
    AND f.date_sk <  current_date()
),
daily AS (
  SELECT g.country, g.d, count(l.order_line_sk) AS line_count
  FROM grid g
  LEFT JOIN fact_lines l
    ON l.country = g.country
   AND l.d       = g.d
  GROUP BY g.country, g.d
),
baseline AS (
  SELECT t.country, t.d, t.line_count, median(h.line_count) AS median_7d
  FROM daily t
  JOIN daily h
    ON h.country = t.country
   AND h.d >= t.d - INTERVAL 7 DAYS
   AND h.d <  t.d
  GROUP BY t.country, t.d, t.line_count
)
SELECT country, d, line_count, median_7d
FROM baseline
WHERE d = current_date() - INTERVAL 1 DAY
  AND (line_count < 0.5 * median_7d OR line_count > 2.0 * median_7d);

-- Eight days side by side, for a human to read during an incident.
-- pct_amount_null is the column that shows the missing fx_rate.
-- This file is deliberately NOT an alert: it has no threshold and always
-- returns eight rows, so scheduling it would fire every night forever.
SELECT
  date_sk,
  count(*)                                                        AS line_count,
  round(100.0 * count_if(amount_eur IS NULL)      / count(*), 2)  AS pct_amount_null,
  round(100.0 * count_if(customer_sk = 'UNKNOWN') / count(*), 2)  AS pct_unknown_customer,
  round(100.0 * count_if(status = 'CANCELLED')    / count(*), 2)  AS pct_cancelled,
  round(100.0 * count_if(currency = 'CHF')        / count(*), 2)  AS pct_chf
FROM dwh_prod.gold.fct_order_line
WHERE date_sk >= current_date() - INTERVAL 8 DAYS
  AND date_sk <  current_date()   -- today is a partial day and always reads
                                  -- as a step change; it is not one
GROUP BY date_sk
ORDER BY date_sk;

-- Yesterday's shape against the median of the seven days before it, one row
-- per percentage that moved. Median for the same reason volume uses one: a
-- single odd day must not redefine normal.
WITH daily AS (
  SELECT
    date_sk,
    round(100.0 * count_if(amount_eur IS NULL)      / count(*), 2) AS pct_amount_null,
    round(100.0 * count_if(customer_sk = 'UNKNOWN') / count(*), 2) AS pct_unknown_customer,
    round(100.0 * count_if(status = 'CANCELLED')    / count(*), 2) AS pct_cancelled,
    round(100.0 * count_if(currency = 'CHF')        / count(*), 2) AS pct_chf
  FROM dwh_prod.gold.fct_order_line
  WHERE date_sk >= current_date() - INTERVAL 8 DAYS
    AND date_sk <  current_date()
  GROUP BY date_sk
),
long AS (
  SELECT date_sk, metric, pct
  FROM daily
  UNPIVOT (pct FOR metric IN (
    pct_amount_null, pct_unknown_customer, pct_cancelled, pct_chf))
),
baseline AS (
  SELECT metric, median(pct) AS median_7d
  FROM long
  WHERE date_sk < current_date() - INTERVAL 1 DAY
  GROUP BY metric
)
SELECT y.metric,
       y.pct                        AS pct_yesterday,
       b.median_7d,
       round(y.pct - b.median_7d, 2) AS points_moved
FROM long y
JOIN baseline b
  ON b.metric = y.metric
WHERE y.date_sk = current_date() - INTERVAL 1 DAY
  -- Two percentage points, or a quarter of the metric's own baseline,
  -- whichever is larger. The absolute floor is what makes 0 -> 8 fire on
  -- pct_amount_null; the relative term is what stops pct_cancelled at 30%
  -- alarming on ordinary daily wobble.
  AND abs(y.pct - b.median_7d) > greatest(2.0, 0.25 * b.median_7d);

WITH prod AS (
  SELECT table_schema, table_name, column_name, ordinal_position,
         full_data_type, is_nullable
  FROM dwh_prod.information_schema.columns
  WHERE table_schema IN ('bronze', 'silver', 'gold')
),
test AS (
  SELECT table_schema, table_name, column_name, ordinal_position,
         full_data_type, is_nullable
  FROM dwh_test.information_schema.columns
  WHERE table_schema IN ('bronze', 'silver', 'gold')
)
SELECT 'only_in_prod' AS side, * FROM (SELECT * FROM prod EXCEPT SELECT * FROM test)
UNION ALL
SELECT 'only_in_test' AS side, * FROM (SELECT * FROM test EXCEPT SELECT * FROM prod);