Data Extraction & Hashing Workflows › Incremental Extraction Strategies

# Incremental Extraction Strategies

Incremental extraction is the discipline of reading only the rows that changed since the last reconciliation ran, instead of re-scanning an entire table every cycle. This reference sits inside the data extraction & hashing workflows stage and specialises it for the cost-and-freshness problem that dominates any pipeline past a certain size: a full extract that was fine at a million rows is ruinous at ten billion, and the only way to reconcile large tables frequently is to extract the delta. It is written for the data engineers, migration specialists, Python pipeline builders, and platform operations teams who need reconciliation to keep up with a growing table without a linearly growing bill.

The hard part is not reading changed rows — it is advancing the cursor correctly so that no changed row is ever missed and none is processed twice. A high-water mark advanced too eagerly skips rows committed just after it was read; advanced too conservatively re-processes rows already seen. Both are silent: the first loses data from the reconciliation, the second inflates it. This reference defines how to track the mark, choose an extraction strategy, and advance the cursor exactly once, so incremental extraction is a correctness win, not a source of the very drift reconciliation exists to catch.

# Architectural Boundaries

This workload consumes a source table and the persisted high-water mark from the previous run, and produces the delta — the changed rows — plus an advanced mark for the next run. It feeds the column-level checksum generation and comparison stages exactly as a full extract would; only the set of rows differs. It owns cursor correctness and nothing else: it does not compare, hash-decide, or route. Keeping it that narrow is what lets it be tested for the one property that matters — no row missed, none double-read — in isolation from everything downstream.

# Prerequisites

• A reliable change column exists. A monotonic updated_at, sequence, or version column is present and indexed; without one, incremental extraction by change column is impossible and CDC is the fallback.
• The mark is persisted transactionally. The high-water mark advances only after the batch is durably processed, so a crash cannot lose or skip the delta.
• Out-of-order commit window known. The maximum lateness with which a row can commit below the current mark is measured, so the safety margin is sized correctly.
• Idempotent downstream apply. Reprocessing an overlapping range on restart must be a no-op, per exactly-once reconciliation with idempotent sinks.
• Soft-delete visibility. Deletes are captured — via a tombstone column or CDC — because a change-column scan alone cannot see a removed row.
• Dependency libraries pinned. The database driver and cursor logic are pinned so pagination semantics are reproducible.

# Step-by-Step Implementation

# 1. Read only rows above the mark, with a safety margin

Query rows whose change value exceeds the last mark minus a safety margin, so a row committed slightly out of order is not skipped. The margin is the price of correctness under concurrency.

import logging
from dataclasses import dataclass
from typing import Iterator, List, Dict

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


@dataclass(frozen=True)
class Watermark:
    change_value: int          # last extracted max change value
    safety_margin: int         # re-scan window for out-of-order commits


def select_delta(rows: List[Dict], wm: Watermark, change_col: str) -> List[Dict]:
    """Return rows above (mark - margin); idempotent apply dedupes the overlap."""
    floor = wm.change_value - wm.safety_margin
    delta = [r for r in rows if r[change_col] > floor]
    logger.info("delta of %d row(s) above floor %d", len(delta), floor)
    return delta

# 2. Advance the mark only after a durable commit

The new mark is the maximum change value in the successfully processed batch, written transactionally with the batch’s completion so a crash never advances past unprocessed rows.

def next_watermark(delta: List[Dict], wm: Watermark, change_col: str) -> Watermark:
    """Advance to the max change value seen; never past what was committed."""
    if not delta:
        return wm
    new_value = max(r[change_col] for r in delta)
    logger.info("advancing watermark %d -> %d", wm.change_value, new_value)
    return Watermark(change_value=max(new_value, wm.change_value), safety_margin=wm.safety_margin)

# 3. Capture deletes explicitly

A change-column scan sees inserts and updates but not deletes; reconcile the current key set against the previous extraction’s keys, or consume tombstones, so a removed row is detected as missing rather than silently retained.

def detect_deletes(prev_keys: set, current_keys: set) -> List[str]:
    """Keys present last cycle and absent now are deletes the change scan cannot see."""
    deleted = sorted(prev_keys - current_keys)
    if deleted:
        logger.warning("%d delete(s) detected via key-set difference", len(deleted))
    return deleted

# Strategy Trade-Off

The extraction strategy is chosen from what the source exposes and how deletes must be handled.

# Scaling and Performance

Incremental extraction turns reconciliation cost from a function of table size into a function of change rate, which is the entire point — but only if the change column is indexed, or every run degrades to a full scan. Page the delta with keyset pagination rather than offset so deep ranges stay cheap, and shard by key range for parallel extraction, combining per-shard marks by taking the minimum so no shard’s lag advances the global mark past unprocessed rows. Size the safety margin to the measured out-of-order commit window: too small skips late commits, too large re-scans needlessly. For sources with a change log, prefer log-based CDC — it captures deletes natively and reads the log rather than the table — and fall back to a periodic snapshot diff as a backstop that catches anything the incremental path missed, exactly as batch reconciliation backstops streaming.

# Failure Modes and Diagnostic Runbook

• Skipped rows on eager advance. Cause: the mark advanced past rows that committed out of order below it. Detection signal: the periodic full backstop finds rows the incremental path never extracted. Remediation: widen the safety margin to the measured commit-lateness window.
• Duplicate processing. Cause: the overlap re-scan reprocesses rows into a non-idempotent sink. Detection signal: counts inflate across runs. Remediation: make the downstream apply idempotent on a stable key.
• Missed deletes. Cause: a change-column scan cannot see removed rows. Detection signal: deleted rows persist in the target. Remediation: add key-set-diff delete detection or consume CDC tombstones.
• Full-scan regression. Cause: the change column lost its index. Detection signal: extraction time jumps to full-table scale. Remediation: restore the index; alert on extraction latency, not just row count.

# Deep Dives

• Watermark-based incremental extraction — durable cursor advancement, out-of-order safety margins, and exactly-once extraction across restarts.

# Related

• Data Extraction & Hashing Workflows — the parent stage this delta extraction feeds.
• Async batching for large datasets — the batching pattern that consumes the extracted delta.
• Schema validation pre-checks — the contract gate that runs before extraction.
• Batch reconciliation vs streaming reconciliation — where incremental batch and CDC streaming are chosen between.