Data Engineering / Intermediate: change, time and coordination / Lesson 8

Change data capture and versioned state

Apply updates and deletes according to source order rather than arrival order. Preserve enough version history to prevent stale events from resurrecting deleted records.

About 4 min readContent reviewed 2026-10-26

On this page

Lesson goals

Distinguish snapshots, change events and materialized current state.
Implement explicit version, duplicate and tombstone policies.

A change event describes a source transition

A snapshot describes state at a chosen boundary, while change data capture represents transitions such as insert, update and delete. A connector’s event envelope, source position and transaction metadata have tool-specific meanings. Arrival order across workers or partitions is not automatically the source’s global commit order. Define the identity and ordering domain for each business key and how an initial snapshot joins the subsequent change stream. A gap or overlap at that handoff can cause missing or repeated state even when both pieces individually look valid.

Reject stale per-key changes explicitly

The offline model applies only events with a higher source version for the same key. It stores a delete marker alongside the version rather than removing all evidence of the key. A later-arriving version-two update cannot resurrect a version-three delete. This numeric version is a synthetic contract, not a universal interpretation of timestamps or connector offsets. Duplicate same-version events with conflicting content still need detection. The model demonstrates current-state application only and does not persist an audit log or reproduce a real connector’s transaction behavior.

python Offline versioned delete/replay model

state = {}
def apply(key, version, value, deleted=False):
    current = state.get(key)
    if current is not None and version <= current['version']:
        return False
    state[key] = {'version': version, 'value': value, 'deleted': deleted}
    return True

assert apply('s1', 3, None, True)
assert not apply('s1', 2, 'old')
assert state['s1']['deleted'] is True

The delete remains authoritative over the stale update. Retaining its version is the modeled protection; no source log is consumed.

Keep delete semantics and retention consistent

A hard delete removes source rows, so periodic queries of currently existing records cannot always reconstruct deletion history. A tombstone or equivalent state transition can preserve the information consumers need. Retain version/deduplication state for at least the replay and lateness assumptions that require it, while respecting privacy and retention policy. Deleting deduplication state too early can turn an old replay into a new event. Conversely, retaining sensitive old payloads indefinitely is not justified merely because metadata is useful. Separate necessary identity/version evidence from unnecessary original content.

Apply newer state in one isolated database statement

The SQLite memory example accepts a newer source version and ignores a stale update through an explicit conflict target and WHERE condition. It verifies the final version and value rather than only counting rows. This demonstrates one sink implementation under SQLite UPSERT rules; other warehouses have different MERGE behavior, constraints and concurrency semantics. A duplicate-key policy is not a substitute for a source contract. If equal versions can conflict, compare their content and fail or reconcile under the agreed rule rather than silently letting arrival order decide.

python Version-aware UPSERT in SQLite memory

import sqlite3

db = sqlite3.connect(':memory:')
db.execute('CREATE TABLE current_state(id TEXT PRIMARY KEY,version INTEGER,value TEXT)')
sql = '''INSERT INTO current_state VALUES(?,?,?) ON CONFLICT(id)
DO UPDATE SET version=excluded.version,value=excluded.value
WHERE excluded.version > current_state.version'''
db.execute(sql, ('s1', 3, 'new'))
db.execute(sql, ('s1', 2, 'old'))
assert db.execute('SELECT version,value FROM current_state').fetchone() == (3, 'new')
db.close()

The stale version-two value does not replace version three. No disk database, connector or warehouse is used.

Validate snapshot-to-log continuity and repair paths

Record source positions and accepted target state across snapshot completion, connector restart and schema changes. A connector being healthy does not prove all required events remain in the source log or that a downstream sink accepted them. Monitor lag relative to retention and define a re-snapshot/reconciliation procedure before history expires. Test duplicate delivery, out-of-order changes, deletes, key changes and transactional groups with the chosen tool. Never skip an event solely to make monitoring green without assessing which business state that omission creates.

Common errors and fixes

An old update resurrects a deleted key

Why it happens: Delete/version evidence was removed before the replay horizon ended.
How to fix it: Preserve appropriate tombstone/version state and reconcile the affected key.
Prevent it next time: Test delete followed by stale replay.

A timestamp is treated as universal commit order

Why it happens: Clock and ordering domains were not defined.
How to fix it: Use the source connector’s documented positions and per-key/transaction semantics.
Prevent it next time: Record ordering assumptions explicitly.

A green connector hides missing sink state

Why it happens: Source retrieval and target acceptance were conflated.
How to fix it: Reconcile accepted sink boundaries and event coverage.
Prevent it next time: Monitor end-to-end freshness and log-retention headroom.

Check your understanding

1. Why can physically deleting a target row be insufficient for replay-safe deletion?

Show worked solution

It can erase the version evidence needed to reject a late older update. Preserve an appropriate delete/version marker or another equivalent ordering record for the required replay horizon. Decide payload retention separately from that minimal metadata, and test restart and re-snapshot behavior rather than assuming absence is permanent.

2. Does an UPSERT on a business key automatically solve conflicting same-version changes?