Incremental Extraction and Watermarks

Scenario

Example company: to stop copying the whole orders table every night, the team switches to incremental extraction every 10 minutes: select rows whose updated_at is newer than the highest updated_at seen in the previous run (the watermark).

What you see in production

• Finance finds a handful of orders each week that exist in production but never appear in the warehouse.
• The missing orders all have updated_at values slightly older than the watermark at the time they should have been copied.
• Nothing failed; every run reports success.

The naive code

SELECT * FROM orders WHERE updated_at > :last_watermark;   -- then: last_watermark = max(updated_at) of this batch

Reproduce it

updated_at is set when a row is written, but the row becomes visible only when its transaction commits. A slow transaction can commit after an extract ran, carrying a timestamp that is already behind the watermark.

# (updated_at, committed_at) in seconds after 10:00:00
rows = {
    "A": (2, 3),
    "B": (5, 25),   # written at :05 by a slow transaction that commits at :25
    "C": (8, 9),
    "D": (30, 31),
}

def run_extract(now, watermark, overlap=0):
    visible = {k: v for k, v in rows.items() if v[1] <= now}       # only committed rows are visible
    since = watermark - overlap
    batch = sorted(k for k, (upd, _) in visible.items() if upd > since)
    new_watermark = max([watermark] + [rows[k][0] for k in batch])
    return batch, new_watermark

for overlap in [0, 60]:
    loaded, watermark = set(), -1
    for now in [10, 20, 40]:  # extract runs at :10, :20 and :40
        batch, watermark = run_extract(now, watermark, overlap)
        loaded |= set(batch)
    missing = sorted(set(rows) - loaded)
    print(f"overlap {overlap:>2}s: loaded {sorted(loaded)}, missing {missing}")

overlap  0s: loaded ['A', 'C', 'D'], missing ['B']
overlap 60s: loaded ['A', 'B', 'C', 'D'], missing []

Root cause

The watermark assumes rows appear in updated_at order. They don't: timestamps are taken when a transaction writes, visibility happens at commit, and long transactions, clock skew between application servers, and identical timestamps all break the assumption. Hard deletes are also invisible to this query.

The fix

Pick one or combine them:

1. Overlap (lookback) window: extract updated_at > watermark - overlap, where the overlap is longer than your longest transaction (often 5 to 60 minutes). This re-reads some rows, so the load must be idempotent: upsert by primary key (next module).
2. Commit-ordered positions: use something that increases in commit order, such as the database log position, instead of a wall-clock timestamp. That is what change data capture does (next lesson).
3. Reconciliation: periodically compare counts or checksums per day between source and warehouse, and reload days that differ.

Set updated_at in the database (DEFAULT now() plus a trigger, or the ORM's equivalent) so every write updates it.

Trade-offs

• Overlap windows re-read data, costing some source load and requiring upserts.
• Log-based CDC is accurate but needs database permissions and extra infrastructure.
• Full reloads are simplest and always correct, which is fine for small tables.

Detect it in production

• Daily row-count and sum reconciliation between source and warehouse.
• Monitor the age of the oldest open transaction in the source database; long ones are your overlap's lower bound.
• Alert when the watermark stops moving.

Key points

• updated_at > watermark misses rows from slow transactions and ignores deletes.
• Use an overlap window plus an idempotent upsert, or log-based CDC.
• Reconcile counts with the source regularly; silent gaps report success.