Idempotent Incremental Loads and Watermarks

Published by Programming Line in Data Engineering in Depth. Last updated 7th September 2026.

Purpose and outcome

An incremental load moves only data that is new or newly changed since a prior pipeline run. A watermark is the remembered position that tells the next run where to resume. Idempotence means the same run can be attempted more than once without creating duplicate rows, skipping valid rows, or corrupting the target. Together, these ideas let a data engineering batch pipeline recover from common interruptions: a worker restarts, a warehouse command times out after committing, or an upstream extract is delivered twice.

In this lesson, the outcome is practical: you should be able to design a batch ingestion job that reads source changes by a high-water mark, writes them through a deterministic merge or partition replacement, advances the watermark only after a verified commit, and handles late-arriving records without turning every rerun into a cleanup project.

How the mechanism works

The core loop has four durable facts. First, the source has an ordering column or cursor, such as updated_at, an increasing log sequence number, a file modification timestamp, or a message offset. Second, a control table stores the last committed watermark for each pipeline, source, and target. Third, the load extracts a bounded interval, often (previous_watermark, candidate_watermark]. Fourth, the target write is deterministic: it uses a stable business key or primary key so the same input interval produces the same final table state.

A subtle but important distinction is the difference between the observed maximum and the committed watermark. A job may observe that the source currently contains changes through 10:05, but it must not store 10:05 as committed until the target has safely incorporated every selected row through that point. If the watermark is advanced before the target write, a failure can make the next run start after rows that were never loaded. If the target write completes but the watermark update fails, the next run should reprocess the same interval; idempotent target logic makes that safe.

Watermark comparison must also account for ties. If many records share the same updated_at, using only updated_at > previous_watermark can miss rows inserted after a prior extract but with the same timestamp. Common solutions are to use a composite cursor such as (updated_at, id), to overlap each batch by a lookback window, or to rely on a source change log with unique monotonically increasing positions. Composite cursors are precise when available. Lookback windows are simpler but require deduplication because they intentionally reread old records.

API and table anatomy

A minimal control table stores pipeline_name, source_name, target_name, watermark_value, watermark_type, status, and audit timestamps. A more complete design also records run_id, low_watermark, high_watermark, row counts, a checksum, and the target commit identifier. These fields let operators distinguish an interval that was selected, an interval that was written, and an interval that was fully committed.

The extraction predicate usually has this shape: choose a stable high watermark at the beginning of the run, then read rows greater than the previous committed watermark and less than or equal to that high watermark. The stable high watermark prevents the batch from chasing a moving source while new changes continue to arrive. Target writes then use one of three common patterns: MERGE into a keyed table, delete-and-insert for the affected keys, or replace complete partitions when the source maps cleanly to partition boundaries.

CREATE TABLE etl_watermarks (
  pipeline_name TEXT NOT NULL,
  source_name TEXT NOT NULL,
  target_name TEXT NOT NULL,
  watermark_value TEXT NOT NULL,
  watermark_type TEXT NOT NULL,
  updated_at TEXT NOT NULL,
  PRIMARY KEY (pipeline_name, source_name, target_name)
);

SELECT COALESCE(watermark_value, '1970-01-01T00:00:00Z') AS low_watermark
FROM etl_watermarks
WHERE pipeline_name = 'orders_incremental'
  AND source_name = 'orders_api'
  AND target_name = 'warehouse_orders';

The control lookup above returns the lower bound for the next extract. In a real warehouse the timestamp type should be native rather than plain text, but the key idea is independent of database product: every pipeline instance owns exactly one committed resume point for a specific source-target pair.

Example 1: append-only events

The simplest incremental load is an append-only event table with a strictly increasing integer event id. Suppose the committed watermark is 102. The source currently has event ids 103, 104, and 105. The batch selects event_id > 102 AND event_id <= 105, inserts those rows, and then stores 105. If the insert succeeds and the watermark update fails, the next run rereads 103 through 105. A unique constraint on event_id or an insert that ignores existing keys makes the rerun harmless.

events = [101, 102, 103, 104, 105]
committed_watermark = 102
candidate_watermark = max(events)
selected = [event_id for event_id in events if committed_watermark < event_id <= candidate_watermark]
print(selected)

The deterministic output is [103, 104, 105]. This example works because the cursor is unique and monotonic. There are no updates to previous ids, and there is no ambiguity at the boundary.

Example 2: mutable orders with merge

Most operational tables are mutable. An order can be created, paid, shipped, canceled, or corrected. The incremental predicate usually follows updated_at, while the target write keys by order_id. If order 7 was loaded yesterday with status paid and appears today with status shipped, idempotence requires an upsert, not an append. Appending would preserve both versions unless the table is intentionally modeled as history.

MERGE INTO warehouse_orders AS target
USING stage_orders AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN UPDATE SET
  status = source.status,
  amount = source.amount,
  updated_at = source.updated_at
WHEN NOT MATCHED THEN INSERT (order_id, status, amount, updated_at)
VALUES (source.order_id, source.status, source.amount, source.updated_at);

Expected behavior: rerunning the same staged rows leaves one row per order_id with the same values. The operation is idempotent because the key match determines whether each source row becomes an insert or an update. The trade-off is that a merge may be more expensive than append-only ingestion, especially when the target table is large and the storage engine must rewrite files or partitions.

Example 3: late arrivals with a lookback window

A timestamp watermark becomes less exact when source systems write records late or clocks differ. If the previous committed watermark is 12:00 and a source record arrives at 12:05 with updated_at = 11:58, a strict predicate of updated_at > 12:00 will never read it. One practical response is a lookback window: subtract a fixed interval from the committed watermark and reread recent records every run. The target merge absorbs duplicates while capturing late changes.

from datetime import datetime, timedelta, timezone

committed = datetime(2026, 9, 7, 12, 0, tzinfo=timezone.utc)
lookback = timedelta(minutes=10)
low = committed - lookback
rows = [
    (1, datetime(2026, 9, 7, 11, 58, tzinfo=timezone.utc)),
    (2, datetime(2026, 9, 7, 12, 3, tzinfo=timezone.utc)),
    (3, datetime(2026, 9, 7, 12, 12, tzinfo=timezone.utc)),
]
selected_ids = [row_id for row_id, updated_at in rows if low < updated_at <= datetime(2026, 9, 7, 12, 15, tzinfo=timezone.utc)]
print(selected_ids)

The output is [1, 2, 3]. Row 1 is intentionally reread even though its timestamp is earlier than the committed watermark. This is reliable only if the target write deduplicates or merges by key. A lookback window should be chosen from measured source delay, not guesswork; too small misses late data, while too large increases scan and merge cost.

Design choices and trade-offs

The first design choice is cursor selection. Integer offsets and database log sequence numbers are excellent because they are ordered and unique. Timestamps are common but require consistent time zones, sufficient precision, and tie handling. File-based watermarks can use a manifest of processed file names instead of a single timestamp, because object storage listing order and modification times are not always a complete business ordering.

The second choice is target write strategy. Append with a uniqueness guard is fast for immutable facts. Merge supports mutable dimensions and operational snapshots. Partition replacement is attractive when each run owns complete partitions, such as one business date at a time, because rerunning the partition replacement yields the same partition contents. It is risky when a batch contains only partial partition data.

The third choice is when to advance the watermark. Advance it only after the data write and validation have completed. In systems without transactions across the target table and control table, store run metadata so recovery can decide whether to rerun, reconcile, or mark the interval committed. A two-step status such as running then committed is useful, but the committed watermark remains the only value used by the next normal extract.

Failure modes and troubleshooting

A common symptom is duplicate rows after retries. The usual cause is an append-only target without a unique key or merge. Diagnose by grouping the target by business key and counting duplicates for the affected interval. Correct it by adding a uniqueness constraint where the platform supports one, rewriting the load as a merge, or staging and replacing the impacted partition.

Another symptom is a gap: source rows exist below or equal to the stored watermark but are absent from the target. This often means the watermark was advanced before the target write finished. Check run logs for a failed load whose high watermark matches the stored value, compare stage row counts to target row counts, and inspect the control table update time. Correct by backfilling from the last known good watermark, then changing the job so the control table update occurs after target verification.

A third symptom is a job that never finishes during busy periods. The cause may be an unbounded high watermark: the query keeps seeing newer source rows while it scans. Diagnose by logging the chosen high watermark at run start and confirming the extract predicate includes an upper bound. Correct by capturing a candidate high watermark first, then extracting only through that value.

A fourth symptom is intermittent missing records at identical timestamps. The cause is boundary ambiguity. Diagnose by looking for multiple source rows with the same cursor value as the previous high watermark. Correct with a composite watermark, such as timestamp plus id, or a lookback window with key-based merge.

Security, performance, and reliability implications

Incremental pipelines often need read access to operational change columns that are not exposed in application reports. Treat the control table as operational metadata: limit writes to the orchestrator identity and audit manual changes, because a bad watermark can silently skip or replay data. If personal data is present, retries and staging tables must follow the same retention and access rules as the final warehouse table.

Performance depends on source and target pruning. Index or cluster the source by the cursor when possible, and keep the target organized by merge key, partition date, or both. Large lookback windows make every run more expensive. Reliability improves when each run records selected row count, written row count, rejected row count, and the final high watermark. Those measurements make backfills and incident analysis concrete rather than speculative.

Hands-on lab: build a rerunnable incremental load

Prerequisites: a SQL database that supports a staging table and merge-like upsert, or a local analytical engine with equivalent behavior; permission to create three small tables; and a source table with an id, status, and updated_at column.

Create source_orders, warehouse_orders, and etl_watermarks. Seed the source with three orders whose updated_at values span one hour.
Insert an initial watermark of 1970-01-01T00:00:00Z for the orders pipeline.
At run start, query the previous watermark and compute the candidate high watermark as the maximum updated_at in source_orders.
Load rows in (low, high] into stage_orders. Keep the high watermark fixed even if new source rows arrive during the run.
Merge stage_orders into warehouse_orders by id.
Verify that the target row count and key count match the staged interval, then update etl_watermarks to the candidate high watermark.
Rerun the same interval manually by resetting the watermark to the prior value. The final target should still contain one row per order id with the latest staged values.

Verification should include three checks: no duplicate id values in the target, the stored watermark equals the candidate high watermark only after the merge, and a second run with unchanged source data writes no additional logical records. Cleanup is simple for a sandbox: drop the three tables. In a shared environment, rollback by restoring the previous watermark and deleting only rows whose ids were created for the lab.

Assessment exercises

A job reads rows where updated_at > previous_watermark and then appends them to the target. It is retried after a network timeout. Explain two different ways this can create wrong results and redesign the write path.
Your source table has many rows with the same second-level timestamp. Design a composite watermark and describe the exact predicate for the next batch.
A late-arriving update is missed because its timestamp is five minutes earlier than the stored watermark. Choose a lookback strategy and explain how you would validate the window size from observed data.
A pipeline updates the watermark in one database and writes target rows in another. What run metadata would you store to make recovery deterministic after a partial failure?
For a daily partition replacement load, list the condition that must be true before partition replacement is idempotent rather than destructive.

Summary

Idempotent incremental loading is a contract between cursor selection, bounded extraction, deterministic target writes, and delayed watermark advancement. The watermark says what has been fully committed, not merely what the job noticed. A reliable data engineering pipeline chooses a cursor that matches source behavior, handles boundary ties and late arrivals, uses merge or partition replacement when data can change, and records enough run evidence to recover from partial failure without guessing.