11Observability

-- Two clocks, deliberately. A full refresh that rewrites yesterday's rows
-- keeps _tech_loaded_at current while the newest order is 40 hours old.
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. A step change in any column is the alert;
-- pct_amount_null is the one that catches the missing fx_rate.
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;

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);

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.catalog import MonitorTimeSeries

w = WorkspaceClient()

w.quality_monitors.create(
    table_name="dwh_prod.gold.fct_order_line",
    output_schema_name="dwh_prod.gold",
    assets_dir="/Workspace/Shared/dwh/monitors/fct_order_line",
    time_series=MonitorTimeSeries(
        timestamp_col="order_ts",
        granularities=["1 day"],
    ),
    # slicing is evaluated on the monitored table, so only its own columns
    slicing_exprs=["status", "currency", "region_sk"],
)