DE Mastery GitHub

Lakehouse & open table formats · LKH-04

Row-level changes — SCD2, CoW & MoR

From your world. Your SCD2 dimension snapshots and Redshift upserts

← All exercises

SCD2 with MERGE INTO, checked against a DuckDB oracle

Apply a daily batch to a Slowly-Changing-Dimension (Type 2) stored in Iceberg. Version the history exactly as you do in your Redshift/dbt dimensions. Implement this in solution.py:

apply_scd2(spark, dim_ident, src_values_sql, batch_date)

  • dim_ident — the dimension table, for example local.db.dim. It has the columns nk INT, attr STRING, valid_from DATE, valid_to DATE, is_current BOOLEAN.
  • src_values_sql — a SQL query that yields the natural-key rows of the day with the columns nk, attr (for example SELECT * FROM VALUES (1,'z'),(3,'w') AS t(nk, attr)).
  • batch_date — the effective date of this batch as a 'YYYY-MM-DD' string.

Do it in two steps, in this order:

  1. Close every current row whose tracked attribute changed. Set its valid_to = batch_date and is_current = false. Use MERGE INTO ... WHEN MATCHED AND t.attr <> u.attr THEN UPDATE ... and join on nk and t.is_current = true.
  2. Insert a new open version (valid_from = batch_date, valid_to = NULL, is_current = true) for every source key that is not currently present as a current row. This covers both the keys you just closed and brand-new keys.

Rules and traps:

  • Do step 1 before step 2. If you insert first, the new row looks current and the close step fires wrongly.
  • Never overwrite the attribute in place. That is SCD Type 1 and it destroys history.
  • The valid_to of the old row must equal the valid_from of the new row. The dates are contiguous: no gap and no overlap.
  • Leave unchanged keys exactly as they were. Each brand-new key gets one current row.

The test seeds a day-1 dimension and applies a day-2 batch through your function. Then it compares the full table against an independent DuckDB oracle. The oracle computes the same close and insert over the same batches (assert_frames_equal, order-independent). The test also checks the SCD2 invariants directly: one current row per key, and contiguous dates.

Starter code

solution.py
"""LKH-04 / 01-scd2-merge — starter. See README.md."""


def apply_scd2(spark, dim_ident, src_values_sql, batch_date):
    # TODO: step 1 close changed current rows (valid_to=batch_date, is_current=false)
    # via MERGE; step 2 insert a new open version for changed + new keys. Step 1 first.
    raise NotImplementedError

Hints

Hint 1

Two steps. Step 1 CLOSES changed current rows: MERGE INTO {dim} t USING ({src}) u ON t.nk=u.nk AND t.is_current=true WHEN MATCHED AND t.attr<>u.attr THEN UPDATE SET t.valid_to=DATE'{batch_date}', t.is_current=false.

Hint 2

Step 2 INSERTS a new open version for changed AND brand-new keys: INSERT the source rows whose nk is not among the still-current rows, with valid_from=DATE'{batch_date}', valid_to=NULL, is_current=true. A LEFT ANTI JOIN (or NOT IN) against the current rows selects them.

Hint 3

Do step 1 before step 2 (otherwise the just-inserted new version looks 'current' and blocks the close). Never UPDATE the attr in place (that is SCD1). Close date == new valid_from (contiguous, no gap/overlap).

Run it locally
uv run de start LKH-04/01-scd2-merge
# edit workspace/LKH-04/01-scd2-merge/
uv run de test