SQL for Data Engineers: Medallion Layers and CDC

In Snowflake:

CREATE FILE FORMAT ff_csv TYPE = CSV SKIP_HEADER = 1 FIELD_OPTIONALLY_ENCLOSED_BY = '"';

CREATE TABLE bronze_txn (
    txn_id STRING, account_id STRING, txn_ts STRING, amount STRING, direction STRING, merchant STRING,
    _file_name STRING, _ingested_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(), _batch_id STRING
);

COPY INTO bronze_txn (txn_id, account_id, txn_ts, amount, direction, merchant, _file_name, _batch_id)
FROM (SELECT $1, $2, $3, $4, $5, $6, METADATA$FILENAME, 'B-2025-03-02' FROM @raw_stage/txn/)
FILE_FORMAT = (FORMAT_NAME = ff_csv);

The same logic in portable SQL, as run on DuckDB to produce the output above:

CREATE TABLE bronze_txn (
    txn_id       VARCHAR,
    account_id   VARCHAR,
    txn_ts       VARCHAR,     -- everything lands as text: typing happens in Silver
    amount       VARCHAR,
    direction    VARCHAR,
    merchant     VARCHAR,
    _file_name   VARCHAR,     -- lineage: which file
    _ingested_at TIMESTAMP,   -- lineage: when it landed
    _batch_id    VARCHAR      -- lineage: which run
);

INSERT INTO bronze_txn
SELECT txn_id, account_id, txn_ts, amount, direction, merchant,
       filename, current_timestamp, 'B-2025-03-02'
FROM read_csv('txn_*.csv', header = true, all_varchar = true, filename = true);

• Everything lands as text. If amount were DECIMAL, the load would fail on 'n/a' (or, with error-skipping turned on, drop the row). Bronze's job is to keep exactly what arrived, so typing and validation happen in Silver, where a bad row can be quarantined with a reason (Q50).
• Lineage columns: _file_name (DuckDB's filename = true, Snowflake's METADATA$FILENAME (docs)), _ingested_at and _batch_id let you trace any row to its file and run, and replay or delete one bad file's rows.
• Duplicates stay. T-2 is in both files, and Bronze keeps both copies. Deduplication is a Silver decision (Q39, Q44); dropping rows at the door would make the archive incomplete.
• Bronze is append-only. Never update or delete it. Because it is complete and immutable, Silver and Gold can be rebuilt whenever their logic changes.

Snowflake's COPY INTO also remembers which files it has already loaded and skips them on a re-run, which makes the load itself idempotent for unchanged files (docs). DuckDB's read_csv has no such memory, so a re-run there loads the files again.

Engineer check. Grain: one row per row delivered, duplicates included. Run twice: on Snowflake, COPY INTO's load history skips files it has seen; elsewhere, track loaded files yourself. NULL: an empty CSV field arrives as NULL or empty string depending on the file format, so decide which before Silver relies on it.

SELECT op, account_id, balance, status, source_ts, seq_no, _ingested_at
FROM bronze_account_cdc
QUALIFY ROW_NUMBER() OVER (PARTITION BY account_id, seq_no
                           ORDER BY _ingested_at DESC) = 1
ORDER BY account_id, seq_no;

• PARTITION BY account_id, seq_no: one change is one log position for one key. The two deliveries of change 101 collapse to one.
• Not PARTITION BY account_id: that would also collapse ACC-1's two different changes (101 and 102) into one. Removing redeliveries and reducing many changes to the latest are two separate steps, and confusing them loses history (Q40, Q42).
• ORDER BY _ingested_at DESC: the copies are identical, so either is fine; taking the latest delivery is a deterministic rule.

Engineer check. Grain: one row per change (account_id, seq_no). Run twice: read-only and deterministic. NULL: a change with a NULL seq_no can't be ordered or deduplicated reliably; reject it at Silver.

SELECT *
FROM cdc_batch
QUALIFY ROW_NUMBER() OVER (PARTITION BY account_id
                           ORDER BY seq_no DESC) = 1
ORDER BY account_id;

• seq_no is the commit order. It's the source database's log position, assigned in exactly the order changes were committed. ACC-1 went 100 → 50 → 80, so the final balance is 80.
• source_ts can tie or skew. Two commits in the same second share a timestamp, and timestamps from different source nodes can disagree. Ordered by source_ts, changes 102 and 103 tie, and the engine may pick 50: a balance the account no longer has (step 2).
• Why collapse at all: the MERGE that applies the batch (Q41) must see at most one source row per key, or it's nondeterministic and Snowflake rejects it (docs).
• Collapse only for the current-state table. A history table needs every change (Q42).

Engineer check. Grain: one row per account. Run twice: deterministic, because seq_no is unique per account. NULL: never order changes by a nullable or non-unique column.

MERGE INTO silver_account t
USING changes s
   ON t.account_id = s.account_id
WHEN MATCHED AND s.op = 'D' THEN UPDATE SET
    is_deleted = TRUE, _updated_at = s.source_ts
WHEN MATCHED AND s.op IN ('I', 'U') THEN UPDATE SET
    balance = s.balance, status = s.status, is_deleted = FALSE, _updated_at = s.source_ts
WHEN NOT MATCHED AND s.op IN ('I', 'U') THEN INSERT
    VALUES (s.account_id, s.balance, s.status, FALSE, s.source_ts);

• Conditional branches (WHEN MATCHED AND s.op = 'D') let one statement handle all three operations. Branches are checked in order and the first that applies wins.
• Soft delete: ACC-3 stays, with is_deleted = TRUE. Readers filter WHERE NOT is_deleted. Physically deleting in Silver would lose the fact that the account existed and was closed, which history (Q42), downstream incremental loads and audits all need, and a replay from Bronze couldn't tell "deleted" from "never loaded".
• An I for an existing key is treated as an update (ACC-5). After a replay or an at-least-once redelivery, an insert can arrive for a row that's already there; treating it as an upsert keeps the statement idempotent instead of failing on the primary key.
• A D for an unknown key does nothing (ACC-4): no branch handles NOT MATCHED with op = 'D', so it falls through. Deleting what isn't there is a no-op.
• is_deleted = FALSE on updates revives an account that's re-inserted after a delete.

Engineer check. Grain: one row per account, current state. Run twice: verified unchanged. NULL: the delete events carry NULL balances, which is fine because the delete branch doesn't copy them.

CREATE OR REPLACE TABLE silver_account_history AS
WITH ordered AS (
    SELECT *,
           LEAD(source_ts) OVER (PARTITION BY account_id ORDER BY seq_no) AS next_ts
    FROM changes
)
SELECT
    account_id, balance, status,
    source_ts                                       AS effective_from,
    COALESCE(next_ts, TIMESTAMP '9999-12-31 00:00')  AS effective_to,
    next_ts IS NULL                                 AS is_current,
    ROW_NUMBER() OVER (PARTITION BY account_id ORDER BY seq_no) AS version
FROM ordered
WHERE op IN ('I', 'U')
ORDER BY account_id, version;

• Use every change, not the collapsed batch. Collapsing to the last change per account (Q40) is right for the current-state table and wrong for history: ACC-1's FROZEN version would vanish, because a later change in the same batch replaced it (step 2). The original version of this exercise made exactly that mistake.
• LEAD(source_ts) OVER (PARTITION BY account_id ORDER BY seq_no) is the start of the next change, which is where this version's window ends. Ordering by seq_no (Q40) keeps it correct even when timestamps tie.
• The D row is used, then dropped. It's still in ordered when LEAD runs, so ACC-2's last version ends at the delete time. Then WHERE op IN ('I', 'U') removes the D row itself, so no version is opened for it. ACC-2 has no current version: correctly, it no longer exists.
• is_current = next_ts IS NULL: only the last version of a live account is open.

To maintain this incrementally, process each new batch the same way, and close the account's currently open version at the timestamp of its first new change.

Engineer check. Grain: one row per account version. Run twice: CREATE OR REPLACE rebuilds the same table. NULL: the open version's end is a sentinel, not NULL, so as-of queries (post 06) can use a plain range predicate.

SELECT account_id,
       MIN(CAST(txn_ts AS DATE))                                       AS recompute_from,
       string_agg(DISTINCT CAST(CAST(txn_ts AS DATE) AS VARCHAR), ', ' ORDER BY CAST(CAST(txn_ts AS DATE) AS VARCHAR)) AS days_touched
FROM bronze_txn
WHERE _ingested_at > (SELECT watermark_ts FROM etl_control WHERE table_name = 'silver_txn')
GROUP BY account_id
ORDER BY account_id;

• Select the batch by arrival (_ingested_at > watermark), then describe it by event time (txn_ts). Mixing the two up is the root of most late-data bugs (Q27).
• T-7 is late: it happened on 2 March and arrived on 5 March, after 2 March's Gold aggregate had already run. Without this step, 2 March stays wrong forever.
• recompute_from matters for cumulative measures like a closing balance: a late row on 2 March changes 2 March and every day after it, so Q45 recomputes from this day forward, not just the late day itself.
• string_agg(DISTINCT …) lists the days for a log message; the pipeline uses recompute_from.

Engineer check. Grain: one row per account with new data. Run twice: read-only. NULL: a NULL txn_ts can't be placed on a day; quarantine it before this step.

SELECT
    txn_id, account_id, amount, merchant, _ingested_at,
    MIN(md5(concat_ws('|', account_id, amount, merchant))) OVER w
      <> MAX(md5(concat_ws('|', account_id, amount, merchant))) OVER w   AS has_conflict
FROM bronze_txn
WINDOW w AS (PARTITION BY txn_id)
QUALIFY ROW_NUMBER() OVER (PARTITION BY txn_id ORDER BY _ingested_at) = 1
ORDER BY txn_id;

• Keep first for events, keep latest for records. A transaction happened once, so its first record is the truth and later copies are redeliveries. A customer record genuinely changes, so there the latest version wins (Q11). Choosing the wrong rule either resurrects stale values or silently overwrites real ones.
• T-2's copies are identical, so dropping the second is safe.
• T-3's copies disagree (60.00 then 66.00). Keep-first would silently keep 60, keep-latest would silently keep 66, and either way nobody learns that the source sent two different stories. The window compares the smallest and largest payload fingerprint per txn_id; if they differ, the deliveries conflict. Route those rows to review (Q50) instead of letting a rule decide.
• In Postgres, which has no QUALIFY, wrap the query and filter on the row number.

Engineer check. Grain: one row per txn_id. NULL: concat_ws skips NULLs on DuckDB and Postgres, so for a real fingerprint use Q19's NULL-safe recipe.

DELETE FROM gold_daily_balance g
USING affected a
WHERE g.account_id = a.account_id
  AND g.balance_date >= a.recompute_from;

INSERT INTO gold_daily_balance
WITH days AS (
    SELECT CAST(d AS DATE) AS d
    FROM generate_series(DATE '2025-03-01', DATE '2025-03-04', INTERVAL 1 DAY) t(d)
),
daily AS (                                     -- every account x every day, empty days included
    SELECT o.account_id, days.d,
           COALESCE(SUM(t.signed_amount), 0) AS net_change,
           COUNT(t.txn_id)                   AS txn_count
    FROM account_opening o
    CROSS JOIN days
    LEFT JOIN silver_txn t
           ON t.account_id = o.account_id AND CAST(t.txn_ts AS DATE) = days.d
    GROUP BY ALL
),
running AS (                                   -- cumulative over the FULL history
    SELECT daily.*,
           o.opening_balance + SUM(net_change) OVER (PARTITION BY daily.account_id ORDER BY d) AS closing_balance
    FROM daily JOIN account_opening o USING (account_id)
)
SELECT r.d, r.account_id, r.closing_balance, r.txn_count
FROM running r
JOIN affected a ON a.account_id = r.account_id AND r.d >= a.recompute_from;

• Recompute from the earliest affected day forward. The closing balance is cumulative, so T-7 lowers 2, 3 and 4 March by $15. Recomputing only 2 March would leave 3 and 4 March contradicting it.
• The running sum runs over the full history, then the final JOIN affected keeps only the rows to replace. Computing the running sum over only the affected days would restart it from zero.
• The opening balance is the starting point of every account's running sum. Without it, the "closing balance" is just the sum of transactions, which is wrong for any account that existed before the warehouse.
• The day spine gives 3 March a row (no transactions, balance carried forward). Grouping the transactions alone would skip it, and a balance chart would have a hole.
• Delete-then-insert over the affected range makes the step idempotent: run it twice and the same rows are replaced with the same values. The build verified it. A MERGE on (balance_date, account_id) would also work here, but delete-then-insert is simpler when whole ranges are recomputed.

Engineer check. Grain: one row per account per day, every day. Run twice: verified unchanged. NULL: COALESCE(SUM(...), 0) turns an empty day's NULL into a zero change, so the running sum carries the balance forward.

SELECT txn_id, txn_ts, signed_amount,
       1000.00 + SUM(signed_amount) OVER (
           PARTITION BY account_id
           ORDER BY txn_ts, txn_id
           ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
       ) AS running_balance
FROM silver_txn
ORDER BY txn_ts, txn_id;

• ORDER BY txn_ts, txn_id adds a tiebreaker, so the order of T-2 and T-3 is the same on every run.
• ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW accumulates row by row.
• The trap: when a window has ORDER BY and no explicit frame, the SQL default is RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW, and RANGE includes every peer (row with the same ORDER BY value) in the current row's frame (Postgres docs). T-2 and T-3 then both show 1,054.50, the balance after both of them, and the intermediate balance of 1,074.50 never appears (step 1). The end-of-day total is right, which is why this survives testing, but a statement or an overdraft check that reads it is wrong.

Engineer check. Grain: one row per transaction. Run twice: repeatable only because of the txn_id tiebreak. NULL: a NULL amount is skipped by SUM, so the balance carries over that row unchanged; decide if that's acceptable or reject such rows in Silver.

In Snowflake:

SELECT *
FROM (SELECT account_id, DATE_TRUNC('month', txn_ts) AS month, direction, amount FROM silver_txn)
PIVOT (SUM(amount) FOR direction IN ('DEBIT', 'CREDIT'))
    AS p (account_id, month, total_debit, total_credit);
-- PIVOT leaves a NULL where a month has no rows of one kind: wrap with COALESCE if 0 is wanted.

The same logic in portable SQL, as run on DuckDB to produce the output above:

SELECT account_id,
       CAST(date_trunc('month', txn_ts) AS DATE)                    AS month,
       SUM(CASE WHEN direction = 'DEBIT'  THEN amount ELSE 0 END)   AS total_debit,
       SUM(CASE WHEN direction = 'CREDIT' THEN amount ELSE 0 END)   AS total_credit
FROM silver_txn
GROUP BY ALL
ORDER BY account_id, month;

• Conditional aggregation turns values of direction into columns: each SUM(CASE …) adds only its kind of row. ELSE 0 makes a missing kind 0 rather than NULL.
• date_trunc('month', …) buckets timestamps by month.
• Snowflake's PIVOT is shorter but needs the value list written out, and it returns NULL for a missing kind (docs). The CASE form works on every engine and handles derived buckets, which is why it's the usual choice in pipelines.

Engineer check. Grain: one row per account and month. NULL: a NULL direction falls into neither column, so those rows disappear from both totals. Quarantine them first (Q50).

WITH marks AS (                          -- active days, plus a marker before the start and after the end
    SELECT DISTINCT account_id, CAST(txn_ts AS DATE) AS d FROM silver_txn
    UNION
    SELECT account_id, opened_date - 1 FROM accounts
    UNION
    SELECT account_id, DATE '2025-03-31' + 1 FROM accounts
),
gaps AS (
    SELECT account_id, d,
           LAG(d) OVER (PARTITION BY account_id ORDER BY d) AS prev_d
    FROM marks
)
SELECT account_id,
       prev_d + 1           AS gap_start,
       d - 1                AS gap_end,
       d - prev_d - 1       AS gap_days
FROM gaps
WHERE d - prev_d - 1 >= 5
ORDER BY account_id, gap_start;

• LAG(d) pairs each active day with the previous one; the days strictly between them (d − prev_d − 1) are dormant.
• Edge markers: the obvious query only measures gaps between transactions, so it misses ACC-2's nine dormant days after opening and its 21 quiet days at the end of the period. Adding a marker the day before opening and the day after the reporting date turns those edges into ordinary gaps.
• UNION, not UNION ALL, so a transaction on the marker day doesn't create a zero-length duplicate.
• Dates subtract to whole days in DuckDB, Postgres and Snowflake; use DATEDIFF('day', …) if your engine needs it.

Engineer check. Grain: one row per dormant stretch. NULL: LAG is NULL on each account's first row, the comparison is unknown, and the row is filtered out, which is right because the start marker is always first.

SELECT a.account_id, a.txn_id, a.txn_ts, COUNT(*) AS txns_in_window
FROM silver_txn a
JOIN silver_txn b
  ON b.account_id = a.account_id
 AND b.txn_ts >  a.txn_ts - INTERVAL 10 MINUTE
 AND b.txn_ts <= a.txn_ts
GROUP BY a.account_id, a.txn_id, a.txn_ts
HAVING COUNT(*) >= 3
ORDER BY a.account_id, a.txn_ts;

• Each transaction is the end of a window (t − 10 min, t]. The self-join collects every transaction of the same account in that window, including itself, and HAVING COUNT(*) >= 3 keeps the bursts. T-3 at 10:09 sees T-1, T-2 and itself.
• The half-open window (> on the left) means a transaction exactly 10 minutes earlier doesn't count. Choose the boundary deliberately; it's a business rule.
• The trade-off: a self-join compares every pair of an account's transactions, which is slow for a busy account. A window function does it in one sorted pass: COUNT(*) OVER (PARTITION BY account_id ORDER BY txn_ts RANGE BETWEEN INTERVAL 10 MINUTE PRECEDING AND CURRENT ROW). DuckDB and Postgres support interval offsets in RANGE frames; check your engine's window-frame documentation before relying on it, because support varies.
• ACC-2's transactions are 11 and 9 minutes apart, so no 10-minute window ever holds three.

Engineer check. Grain: one row per transaction that completes a burst. NULL: a NULL txn_ts matches no window; it can't be placed in time, so reject it in Silver.

CREATE OR REPLACE TEMP TABLE checked AS
SELECT *,
       COALESCE(amount <> 0 AND direction IN ('DEBIT', 'CREDIT'), FALSE) AS is_valid
FROM bronze_txn;

INSERT INTO silver_txn
SELECT txn_id, amount, direction FROM checked WHERE is_valid;

INSERT INTO reject_txn
SELECT txn_id, amount, direction,
       CASE WHEN amount IS NULL                       THEN 'MISSING_AMOUNT'
            WHEN amount = 0                           THEN 'ZERO_AMOUNT'
            WHEN direction IS NULL                    THEN 'MISSING_DIRECTION'
            ELSE                                           'BAD_DIRECTION' END
FROM checked
WHERE NOT is_valid;

• The contract becomes one boolean column, is_valid, and both inserts route on it, so the two sides can't disagree.
• COALESCE(…, FALSE) is the critical part. For T-4 (NULL amount) and T-5 (NULL direction) the rule evaluates to UNKNOWN, not FALSE. With the obvious pair of filters, WHERE rule for Silver and WHERE NOT (rule) for rejects, both evaluate to UNKNOWN for those rows, so they land in neither table and silently disappear (step 2). Forcing UNKNOWN to FALSE sends them to quarantine.
• The reasons are checked in order, NULLs first, so each rejected row gets the most specific reason.
• Quarantine, don't drop: rejected rows can be investigated, fixed upstream and replayed. Snowflake now enforces CHECK constraints too (post 01, Q1), but a failed check aborts the whole insert; routing rows keeps the good ones flowing.
• Prove it: count Bronze rows, Silver rows and rejected rows, and fail the run unless they add up.

Engineer check. Grain: every Bronze row in exactly one target. Run twice: these inserts append again, so in a pipeline they sit behind a watermark (Q27) or replace the batch. NULL: the whole point. A rule that can be UNKNOWN routes nowhere unless you collapse it to TRUE or FALSE.

-- 1. One pass per side, no join
SELECT 'silver' AS side, COUNT(*) AS row_count, SUM(hash(account_id, balance, status)) AS checksum
FROM silver_account WHERE NOT is_deleted
UNION ALL
SELECT 'source', COUNT(*), SUM(hash(account_id, balance, status))
FROM src_account_snapshot;

-- 2. Only when they differ: the row-level diff
SELECT COALESCE(s.account_id, x.account_id) AS account_id,
       CASE WHEN s.account_id IS NULL THEN 'missing in silver'
            WHEN x.account_id IS NULL THEN 'missing in source'
            ELSE 'values differ' END   AS problem
FROM (SELECT * FROM silver_account WHERE NOT is_deleted) s
FULL OUTER JOIN src_account_snapshot x ON x.account_id = s.account_id
WHERE s.account_id IS NULL OR x.account_id IS NULL
   OR hash(s.balance, s.status) <> hash(x.balance, x.status)
ORDER BY account_id;

• SUM(hash(…)) is an order-independent fingerprint of a table. Together with the row count, matching sums mean the tables almost certainly match, and it costs one scan per side with no join. On huge tables that's the check you can afford to run every time.
• Step 2 runs only when step 1 disagrees. The FULL OUTER JOIN finds keys missing on either side (ACC-6 is only in the source, ACC-7 only in Silver), and comparing row hashes finds keys whose values differ (ACC-2's balance).
• Why the count isn't enough: both sides have four live rows, so a row-count check passes even though three accounts are wrong.
• Compare like with like: soft-deleted ACC-3 is excluded from Silver because the source snapshot only has live accounts.
• hash() is engine-specific (DuckDB and Snowflake both have one, with different outputs), so compute both sides on the same engine. To compare across engines, use md5 of a NULL-safe string (Q19).

Engineer check. Grain: one row per mismatched account. Run twice: read-only. NULL: hash(NULL) is a fixed value on DuckDB, but a NULL inside a hashed string can behave differently elsewhere; use the NULL-safe recipe for cross-engine checks.

WITH acct AS (                                -- one row per customer, from accounts alone
    SELECT customer_id,
           COUNT(*)     AS live_accounts,
           SUM(balance) AS total_balance
    FROM silver_account
    WHERE NOT is_deleted
    GROUP BY customer_id
),
txn AS (                                      -- one row per customer, from transactions alone
    SELECT a.customer_id, MAX(t.txn_ts) AS last_txn_ts, COUNT(*) AS txn_count
    FROM silver_txn t
    JOIN silver_account a ON a.account_id = t.account_id
    WHERE NOT a.is_deleted
    GROUP BY a.customer_id
)
SELECT acct.customer_id, live_accounts, total_balance, last_txn_ts,
       COALESCE(txn_count, 0) AS txn_count
FROM acct
LEFT JOIN txn ON txn.customer_id = acct.customer_id
ORDER BY acct.customer_id;

• The fan-out: joining accounts to transactions repeats ACC-1's row three times (it has three transactions). SUM(balance) over the joined rows counts its $1,000 three times, so CUST-1's balance comes out as $3,500 instead of $1,500 (step 1). COUNT(DISTINCT account_id) hides the problem for the account count, which makes the wrong balance easy to miss.
• Aggregate each side to the output grain first (one row per customer), then join one-to-one. The same rule as drill-across in post 02 (Q34).
• LEFT JOIN keeps CUST-2, whose live account has no transactions; an inner join would drop the customer entirely. COALESCE(txn_count, 0) reports zero rather than NULL.
• Deleted accounts are excluded on both sides, so ACC-4's $900 doesn't count.

Engineer check. Grain: one row per customer with a live account. NULL: customers with no transactions get NULL last_txn_ts, which is honest, and a txn_count of 0.

SELECT s.account_id,
       s._batch_id                 AS last_written_by,
       s._source_system,
       s._record_hash <> md5(concat_ws('|', CAST(x.balance AS VARCHAR), x.status)) AS is_stale
FROM silver_account s
JOIN src_account_snapshot x USING (account_id)
ORDER BY s.account_id;

• _batch_id answers "which run wrote this?", which is the first question in any incident: you can find every row a bad run touched, and replay or roll back just those.
• _record_hash is a fingerprint of the business columns stored at write time. Recomputing it from the source with the same recipe and comparing gives a one-column staleness check: ACC-2's source balance moved to 450, so its Silver row is stale and was last written by batch B-0301.
• The recipe must match byte for byte: the same column order, separator, casts and NULL handling (Q19). A different cast (450.0 vs 450.00) makes every row look stale.

Engineer check. Grain: one row per account. Run twice: read-only. NULL: a NULL in a hashed column needs the NULL-safe recipe, or two different rows can share a fingerprint.

-- 1. Freeze this run's window: (watermark, upper]
CREATE OR REPLACE TEMP TABLE run_window AS
SELECT (SELECT watermark_ts FROM etl_control WHERE table_name = 'silver_account') AS lower_ts,
       MAX(_ingested_at) AS upper_ts
FROM bronze_account_cdc
WHERE _ingested_at > (SELECT watermark_ts FROM etl_control WHERE table_name = 'silver_account');

-- 2. The change set: drop redeliveries, then keep the last change per account
CREATE OR REPLACE TEMP TABLE chg AS
WITH new_rows AS (
    SELECT c.* FROM bronze_account_cdc c, run_window w
    WHERE c._ingested_at > w.lower_ts AND c._ingested_at <= w.upper_ts
),
deduped AS (
    SELECT * FROM new_rows
    QUALIFY ROW_NUMBER() OVER (PARTITION BY account_id, seq_no ORDER BY _ingested_at DESC) = 1
)
SELECT * FROM deduped
QUALIFY ROW_NUMBER() OVER (PARTITION BY account_id ORDER BY seq_no DESC) = 1;

-- 3. Apply I/U/D (Q41)
MERGE INTO silver_account t
USING chg s ON t.account_id = s.account_id
WHEN MATCHED AND s.op = 'D' THEN UPDATE SET is_deleted = TRUE, _updated_at = s.source_ts
WHEN MATCHED AND s.op IN ('I', 'U') THEN UPDATE SET
    balance = s.balance, status = s.status, is_deleted = FALSE, _updated_at = s.source_ts
WHEN NOT MATCHED AND s.op IN ('I', 'U') THEN INSERT
    VALUES (s.account_id, s.balance, s.status, FALSE, s.source_ts);

-- 4. Advance the watermark to the bound this run used; an empty run leaves it alone
UPDATE etl_control
SET watermark_ts = COALESCE((SELECT upper_ts FROM run_window), watermark_ts)
WHERE table_name = 'silver_account';

Each step carries a guard, and together they make the job safe to retry:

The second run finds no rows above the new watermark, so chg is empty, the MERGE does nothing and the watermark stays put. The build ran it twice and checked.

ACC-2 never appears. It was opened and closed inside this batch, so the collapsed change is a D for an account Silver has never seen, and nothing is inserted. That is right for a current-state table: the account doesn't exist now. The history table (Q42), built from every change, still records that it existed.

Where the guarantee ends: steps 3 and 4 are separate statements. If the job dies between them, the next run re-applies the same window, which is harmless because the MERGE is idempotent. If your engine supports it, wrap steps 3 and 4 in one transaction so the watermark can never move without the data.

Engineer check. Grain: one current row per account. Run twice: verified unchanged. NULL: delete events carry NULL attributes that the delete branch ignores; an empty window produces a NULL upper_ts that step 4 absorbs.