13Operations

RUNBOOK  silver_sales_order          owner: grp_dwh_engineers
alert    "expectation valid_amount: failed_records > 0"
page     yes, 24/7          (wrong numbers, not late numbers)

1  WHAT BROKE
   The Lakeflow pipeline dropped order lines violating business
   rule 5: amount_eur > 0 for non-cancelled lines. Usual cause:
   silver.fx_rate has no CHF row for the order date, so
   amount_eur comes out NULL.

2  FIRST QUERY  (SQL warehouse "ops", catalog dwh_prod)
   SELECT l.order_ts::DATE AS d, l.currency, COUNT(*) AS lines
   FROM   dwh_prod.silver.sales_order_line l
   LEFT   JOIN dwh_prod.silver.fx_rate r
          ON  r.currency  = l.currency
          AND r.rate_date = l.order_ts::DATE
   WHERE  r.currency IS NULL
     AND  l.order_ts >= current_date() - INTERVAL 3 DAYS
   GROUP  BY 1, 2 ORDER BY 1;

3  DECIDE
   rows returned -> missing FX. Load the rate, then re-run.
   zero rows     -> not FX. Escalate, do NOT re-run.

4  RE-RUN?   safe blind. Full refresh: NO — see re-run table.
5  ESCALATE  after 45 min to grp_dwh_engineers rota #2.
6  TELL      #dwh-status, before you start, not after you finish.

def run(spark, catalog: str, as_of: str) -> int:
    """Rebuild silver.customer as SCD2. Returns rows in the table after.

    as_of is a JOB PARAMETER, never current_timestamp(). Re-running the
    same logical day must produce identical intervals; a wall clock read
    inside the job turns every re-run into a new version.
    """
    target = f"{catalog}.silver.customer"
    incoming = clean_customer(spark.read.table(f"{catalog}.bronze.sap_kna1"))

    new_state = apply_scd2(                       # -> DataFrame, full new state
        current=spark.read.table(target),
        incoming=incoming,
        keys=["customer_id"],
        as_of=as_of,
    )

    # Delta refuses to overwrite a table that is also being read from, and
    # new_state reads target. Stage, then swap — INSERT OVERWRITE keeps the
    # table's comment, clustering and grants; CREATE OR REPLACE loses them.
    stage = f"{catalog}.silver.stg_customer_scd2"
    new_state.write.mode("overwrite").saveAsTable(stage)
    spark.sql(f"INSERT OVERWRITE TABLE {target} SELECT * FROM {stage}")
    spark.sql(f"DROP TABLE {stage}")
    return spark.table(target).count()

SELECT _tech_batch_id,
       COUNT(*)             AS rows_landed,
       MAX(_tech_loaded_at) AS loaded_at
FROM   dwh_prod.bronze.sap_kna1
WHERE  _tech_loaded_at >= current_date()
GROUP  BY _tech_batch_id
ORDER  BY loaded_at DESC;

-- one batch id  -> extract landed once, re-run is safe
-- two batch ids -> already re-run. Do NOT repeat; resume at silver.

-- 1. pause the downstream refresh FIRST. Diagnosing while a scheduled
--    BI refresh republishes the wrong number turns a one-hour incident
--    into a one-day one.

-- 2. how bad, and since when
DESCRIBE HISTORY dwh_prod.gold.fct_order_line LIMIT 10;

SELECT COUNT(*)      AS bad_lines,
       MIN(order_ts) AS first_bad,
       MAX(order_ts) AS last_bad
FROM   dwh_prod.gold.fct_order_line
WHERE  status <> 'CANCELLED'
  AND (amount_eur IS NULL OR amount_eur <= 0);   -- business rule 5

-- 3. blast radius: who read it while it was wrong. Filter on event_date,
--    not event_time: the table is laid out by date, and event_time scans
--    the whole of it — a full scan you are paying for mid-incident.
SELECT DISTINCT entity_type, entity_id, target_table_full_name
FROM   system.access.table_lineage
WHERE  source_table_full_name = 'dwh_prod.gold.fct_order_line'
  AND  event_date >= current_date() - INTERVAL 7 DAYS;
--    a NULL target is a notebook, dashboard or ad-hoc read. Still a consumer.

-- 4. if a bad version was published, roll back — do not patch
RESTORE TABLE dwh_prod.gold.fct_order_line TO VERSION AS OF 412;

  2  FIRST QUERY  (SQL warehouse "ops", catalog dwh_prod)
+       -- the rota needs SELECT on silver.fx_rate; PERMISSION_DENIED
+       -- here means you are not in grp_dwh_engineers.
        SELECT l.order_ts::DATE AS d, l.currency, COUNT(*) AS lines
        ...

  3  DECIDE
        rows returned -> missing FX. Load the rate, then re-run.
        zero rows     -> not FX. Escalate, do NOT re-run.
+       ALL currencies missing for a date -> the ECB CSV was empty,
+       not late. Loading it again will not help. Escalate at once.

  seen: 03.07.2026, 04:12-05:35. Detected by the expectation, not by
  the freshness alert — freshness would have fired at 06:00.