Surviving a re-run

# Importable only because the entry point is split from the module: the
# dbutils.widgets.get() calls live in the __main__ half, so this import does
# not raise outside a notebook. See the standards route, and the gotchas.
from dwh.jobs import silver_customer

# Wall-clock only. If you have to add anything else to this list to make the
# test pass, you have found the bug, not the exception.
VOLATILE = ["_tech_loaded_at"]


def snapshot(spark, table: str) -> list:
    return (
        spark.read.table(table)
        .drop(*VOLATILE)
        .orderBy("customer_id", "_tech_valid_from")
        .collect()                       # collect NOW — see the gotchas
    )


def test_silver_customer_load_is_idempotent(spark, catalog):
    # run(spark, source, target) — the same two names the entry point assembles
    # from its widgets. The test passes them; the function never invents them.
    source = f"{catalog}.bronze.sap_kna1"
    target = f"{catalog}.silver.customer"

    silver_customer.run(spark, source, target)
    first = snapshot(spark, target)

    silver_customer.run(spark, source, target)   # same bronze input, nothing new
    second = snapshot(spark, target)

    assert second == first

import pytest
from dwh.jobs import gold_load

TABLES = ["gold.dim_customer", "gold.fct_order_line", "gold.agg_revenue_month"]


def fingerprint(spark, catalog: str) -> dict:
    """Order-independent content hash per table, ignoring load timestamps."""
    out = {}
    for t in TABLES:
        row = spark.sql(f"""
            SELECT count(*)                              AS n,
                   bit_xor(xxhash64(to_json(struct(*)))) AS h
            FROM  (SELECT * EXCEPT (_tech_loaded_at) FROM {catalog}.{t})
        """).first()
        out[t] = (row.n, row.h)
    return out


class Boom(RuntimeError):
    pass


def explode(*_args, **_kwargs):
    raise Boom("simulated executor loss between tables")


def test_retry_after_partial_failure(spark, catalog, monkeypatch):
    gold_load.run(spark, catalog)
    clean = fingerprint(spark, catalog)

    # die AFTER dim_customer is written, BEFORE fct_order_line
    monkeypatch.setattr(gold_load, "build_fct_order_line", explode)
    with pytest.raises(Boom):
        gold_load.run(spark, catalog)
    monkeypatch.undo()

    gold_load.run(spark, catalog)          # the retry, from the top
    assert fingerprint(spark, catalog) == clean

-- vbap_typed is the rename-only view from the ingestion route: bronze keeps
-- SAP's field names (VBELN, POSNR, MATNR, …), the view maps them to canonical
-- ones and passes change_seq and the _tech_ columns through untouched.
MERGE INTO silver.sales_order_line AS t
USING (
  SELECT order_id, line_number, product_id, quantity, unit_price, currency,
         status, change_seq          -- carried through: the guard needs it
  FROM (
    SELECT *,
           row_number() OVER (
             PARTITION BY order_id, line_number
             -- SAP's change counter, NOT _tech_loaded_at: every row in this
             -- batch shares the load time. See the gotchas.
             ORDER BY     change_seq DESC
           ) AS rn
    FROM vbap_typed
    WHERE _tech_batch_id = :batch_id
  )
  WHERE rn = 1
) AS s
ON  t.order_id    = s.order_id
AND t.line_number = s.line_number
-- The replay guard. Without it, re-running an older batch overwrites the newer
-- version of the line with no error and a green run. Only newer wins; an equal
-- change_seq is the same version and is a no-op, which is what keeps the
-- re-run of the SAME batch idempotent.
WHEN MATCHED AND s.change_seq > t.change_seq THEN UPDATE SET
  t.product_id      = s.product_id,
  t.quantity        = s.quantity,
  t.unit_price      = s.unit_price,
  t.currency        = s.currency,
  t.status          = s.status,
  t.change_seq      = s.change_seq,
  t._tech_loaded_at = current_timestamp()
WHEN NOT MATCHED THEN INSERT
  (order_id, line_number, product_id, quantity, unit_price, currency,
   status, change_seq, _tech_loaded_at)
  VALUES
  (s.order_id, s.line_number, s.product_id, s.quantity, s.unit_price,
   s.currency, s.status, s.change_seq, current_timestamp())