Snowflake Streams and Tasks: CDC, MERGE & Recovery

1. Events are immutable inserts. A correction arrives as a new event, not an UPDATE to the raw event table.
2. Each event identifies one customer and has a source version number.
3. A larger source version wins even if an older version arrives later.
4. For equal versions, the example uses the largest event ID as a deterministic tie-breaker.
5. A deletion is an event with operation D. The target retains the tombstone and its version so an older event cannot resurrect the customer.

The tie-breaker is an exercise assumption, not a universal CDC rule. Before production use, require a source contract that explains whether two different payloads may share a version and how conflicts are resolved. If a source supplies only arrival timestamps, determine whether they really represent business ordering before using them to choose a winner.

Architecture

Workflow: Customer events → append-only stream → winner selection → customer state
Customer state → active-customer view
Task history + key/version checks → run evidence

-- Expected: no duplicate target keys.
SELECT customer_id, COUNT(*) AS copies
FROM CUSTOMER_STATE GROUP BY customer_id HAVING COUNT(*)>1;

-- Save the state, rerun the same MERGE with an empty stream,
-- and compare against this snapshot.
CREATE TEMP TABLE STATE_BEFORE_EMPTY_RERUN AS
SELECT * FROM CUSTOMER_STATE;

MERGE INTO CUSTOMER_STATE t
USING (
  SELECT event_id, customer_id, source_version, operation, customer_name
  FROM CUSTOMER_EVENT_STREAM
  WHERE METADATA$ACTION='INSERT'
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY customer_id
    ORDER BY source_version DESC, event_id DESC
  )=1
) s
ON t.customer_id=s.customer_id
WHEN MATCHED AND (
  s.source_version > t.source_version OR
  (s.source_version=t.source_version AND s.event_id > t.last_event_id)
) THEN UPDATE SET
  customer_name=s.customer_name,
  source_version=s.source_version,
  last_event_id=s.event_id,
  is_deleted=(s.operation='D')
WHEN NOT MATCHED THEN INSERT
  (customer_id,customer_name,source_version,last_event_id,is_deleted)
VALUES
  (s.customer_id,s.customer_name,s.source_version,s.event_id,s.operation='D');
SELECT * FROM CUSTOMER_STATE
MINUS SELECT * FROM STATE_BEFORE_EMPTY_RERUN;
SELECT * FROM STATE_BEFORE_EMPTY_RERUN
MINUS SELECT * FROM CUSTOMER_STATE;

Both comparisons should return zero rows. An empty rerun tests the consumer’s no-change path. It does not prove that every duplicated or out-of-order source event is harmless; those require separate mutation tests.

INSERT INTO CUSTOMER_EVENTS VALUES
 (104,'C01',1,'U','Old Asha'),
 (105,'C02',2,'D',NULL),
 (106,'C03',1,'U','Mina');
MERGE INTO CUSTOMER_STATE t
USING (
  SELECT event_id, customer_id, source_version, operation, customer_name
  FROM CUSTOMER_EVENT_STREAM
  WHERE METADATA$ACTION='INSERT'
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY customer_id
    ORDER BY source_version DESC, event_id DESC
  )=1
) s
ON t.customer_id=s.customer_id
WHEN MATCHED AND (
  s.source_version > t.source_version OR
  (s.source_version=t.source_version AND s.event_id > t.last_event_id)
) THEN UPDATE SET
  customer_name=s.customer_name,
  source_version=s.source_version,
  last_event_id=s.event_id,
  is_deleted=(s.operation='D')
WHEN NOT MATCHED THEN INSERT
  (customer_id,customer_name,source_version,last_event_id,is_deleted)
VALUES
  (s.customer_id,s.customer_name,s.source_version,s.event_id,s.operation='D');
SELECT * FROM CUSTOMER_STATE ORDER BY customer_id;
SELECT * FROM ACTIVE_CUSTOMERS ORDER BY customer_id;

C01 should remain Asha Rao at version 2 because event 104 is older. C02 should remain in CUSTOMER_STATE at version 2 with is_deleted TRUE. C03 should be inserted at version 1. ACTIVE_CUSTOMERS should contain only C01 and C03. A later C02 version-1 update must not reactivate the deleted customer; a deliberate version-3 U event can reactivate it under this lab’s contract.

Keeping a tombstone avoids a common failure: physically deleting the target row and then allowing a delayed old update to insert it again. Retention and cleanup of those tombstones are separate production decisions. Choose a cleanup rule only after defining the maximum source replay horizon.

5.1 Turn the late-event example into an exact state assertion

After inserting events 104–106 and running the same MERGE again, compare all state columns. Run this before the recovery exercise adds event 107.

WITH expected(customer_id,customer_name,source_version,last_event_id,is_deleted) AS (
 SELECT * FROM VALUES ('C01','Asha Rao',2,103,FALSE),
 ('C02',NULL,2,105,TRUE),('C03','Mina',1,106,FALSE)
)
SELECT 'EXPECTED_NOT_FOUND' AS issue,* FROM (
 SELECT * FROM expected MINUS SELECT * FROM CUSTOMER_STATE
)
UNION ALL
SELECT 'UNEXPECTED_STATE' AS issue,* FROM (
 SELECT * FROM CUSTOMER_STATE MINUS SELECT * FROM expected
);
SELECT COUNT(*) AS state_rows,COUNT_IF(is_deleted) AS tombstones FROM CUSTOMER_STATE;
SELECT customer_id,customer_name,source_version FROM ACTIVE_CUSTOMERS ORDER BY customer_id;

Expect zero comparison rows, three state rows and one tombstone. ACTIVE_CUSTOMERS contains C01/Asha Rao/version 2 and C03/Mina/version 1. The late version-1 event for C01 must not overwrite version 2. The deletion removes C02 from the active view while preserving C02’s version in state.

5.2 Test stale resurrection deliberately

INSERT INTO CUSTOMER_EVENTS VALUES (108,'C02',1,'U','Old Ravi');
MERGE INTO CUSTOMER_STATE t
USING (
  SELECT event_id, customer_id, source_version, operation, customer_name
  FROM CUSTOMER_EVENT_STREAM
  WHERE METADATA$ACTION='INSERT'
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY customer_id
    ORDER BY source_version DESC, event_id DESC
  )=1
) s
ON t.customer_id=s.customer_id
WHEN MATCHED AND (
  s.source_version > t.source_version OR
  (s.source_version=t.source_version AND s.event_id > t.last_event_id)
) THEN UPDATE SET
  customer_name=s.customer_name,
  source_version=s.source_version,
  last_event_id=s.event_id,
  is_deleted=(s.operation='D')
WHEN NOT MATCHED THEN INSERT
  (customer_id,customer_name,source_version,last_event_id,is_deleted)
VALUES
  (s.customer_id,s.customer_name,s.source_version,s.event_id,s.operation='D');
SELECT customer_id,source_version,last_event_id,is_deleted
FROM CUSTOMER_STATE WHERE customer_id='C02';

C02 must remain version 2, last event 105, deleted TRUE. This drill can be completed before or after event 107 because it affects a different customer; keep the task suspended so only your intended consumer advances the stream. A newer legitimate restoration event would require an explicitly agreed higher version and business permission, not deletion of the tombstone.