Incremental Extraction Strategies › Watermark-Based Incremental Extraction

Watermark-Based Incremental Extraction

This page answers the single hardest question in incremental extraction: how do you advance the high-water mark so that, across normal runs and crashes alike, every changed row is extracted exactly once? It sits under the incremental extraction strategies reference and focuses on cursor correctness — the property that, when it fails, silently drops rows from the reconciliation or silently re-processes them. Getting the watermark right is what makes incremental extraction a correctness improvement rather than a new source of the drift reconciliation exists to detect.

Problem Framing

You extract changed rows by an updated_at column and store the maximum value seen as the watermark. Two subtle bugs lurk. First, concurrency: a transaction that started before your extract but committed after it can write a row with an updated_at below your new watermark, so the next run — which reads strictly above the watermark — skips it forever. Second, crash timing: if the watermark is advanced before the batch is durably processed, a crash loses the delta between the old and new marks. The fix is a watermark that trails the newest row by a safety margin sized to the commit-lateness window, and that advances only after the batch is committed — with an idempotent apply so the re-scanned overlap is harmless.

Implementation

The extractor reads above watermark − margin, processes the batch, and only then persists the new watermark transactionally with a processed-batch marker, so a crash resumes from the last durable mark and the overlap re-scan is deduplicated downstream.

python
import logging
from dataclasses import dataclass
from typing import Callable, Dict, List, Optional

logging.basicConfig(
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", level=logging.INFO
)
logger = logging.getLogger("recon.extract.watermark")


@dataclass(frozen=True)
class Watermark:
    value: int
    margin: int                 # out-of-order commit window


@dataclass
class WatermarkExtractor:
    load_since: Callable[[int], List[Dict]]     # returns rows with change_value > floor
    persist: Callable[[int], None]              # durably store the new watermark
    change_col: str = "updated_at"

    def run_once(self, wm: Watermark, process: Callable[[List[Dict]], bool]) -> Watermark:
        """Extract the delta, process it, and advance only on a durable success."""
        floor = wm.value - wm.margin            # re-scan the margin for out-of-order commits
        delta = self.load_since(floor)
        if not delta:
            return wm
        if not process(delta):                  # process = hash + hand off downstream
            logger.error("batch processing failed; watermark NOT advanced")
            return wm                            # crash-safe: stay put, reprocess next run
        new_value = max(r[self.change_col] for r in delta)
        advanced = Watermark(max(new_value, wm.value), wm.margin)
        self.persist(advanced.value)            # durable, after the batch succeeded
        logger.info("watermark advanced %d -> %d over %d rows", wm.value, advanced.value, len(delta))
        return advanced

Key Implementation Notes

  • Advance after the commit, never before. The watermark must move only once the batch is durably processed. Advancing first means a crash loses every row between the old and new marks — the classic silent incremental-extraction data loss.
  • Trail the newest row by the commit-lateness window. Reading strictly above the last max skips rows that committed out of order below it. Subtract a safety margin sized to the measured maximum commit lateness so those rows are re-read; the idempotent apply makes the re-read free.
  • Reprocessing the overlap must be a no-op. The margin guarantees some rows are read twice; correctness depends on the downstream apply being idempotent on a stable key, exactly the exactly-once property the streaming stage relies on.
  • Use a monotonic, indexed change column. The watermark is only as reliable as the column it tracks; a non-monotonic or unindexed change column either misses rows or degrades every run to a full scan, defeating the purpose of extracting incrementally before column-level checksum generation.

Verification

Simulate a crash between processing and persistence and assert no row is lost, and that an out-of-order commit within the margin is still caught.

python
store = {"rows": [{"id": i, "updated_at": i} for i in range(1, 6)]}
persisted: List[int] = []

def load_since(floor): return [r for r in store["rows"] if r["updated_at"] > floor]
def persist(v): persisted.append(v)

ext = WatermarkExtractor(load_since, persist)
wm = Watermark(value=0, margin=1)

# Normal run advances to 5.
wm = ext.run_once(wm, process=lambda rows: True)
assert wm.value == 5 and persisted[-1] == 5

# A row commits out of order at updated_at=5 (== newest); margin re-scans it.
store["rows"].append({"id": 99, "updated_at": 5})
seen = ext.run_once(wm, process=lambda rows: bool(rows))
assert any(r["id"] == 99 for r in load_since(seen.value - seen.margin))

# A crash: process fails, watermark must NOT advance.
before = wm.value
wm = ext.run_once(wm, process=lambda rows: False)
assert wm.value == before
logger.info("watermark extraction crash-safety verified")

Operational Considerations

Persist the watermark in the same transaction as the batch’s completion marker where the source allows it, so advancement and processing are atomic; where they cannot share a transaction, persist the watermark strictly after a durable downstream acknowledgement. Measure the real commit-lateness distribution and set the margin to its P99.9, then alert if observed lateness ever exceeds the margin, because that is the precise condition under which rows are silently skipped. Run a periodic full-table backstop reconciliation to catch anything the incremental path missed — the margin reduces that risk but does not eliminate it — and expose watermark_value, delta_rows_per_run, and overlap_rows_reprocessed so an operator can see the cursor advancing and the overlap staying bounded. Record the watermark and margin with each run so an extraction is reproducible during an audit.