Data Engineering Design Patterns: Incremental Loads and Watermarks

This is part 2 of the data engineering design patterns series. Part 1 made a load safe to run twice, but every load in it was handed its batch: one file, one date. This part is about the step before that: deciding which rows count as the next batch when the source is a table that keeps growing.

There are two common answers. A watermark is a stored value, usually the newest updated_at already loaded, and each run reads the rows after it. A data interval is the time window the scheduler assigns to a run, such as 10:00 to 11:00, and each run reads the rows inside it. Both look simple, and both lose rows without raising an error.

#The failure

An hourly job copies new transactions from an application database into the warehouse. It reads the rows after the watermark, saves the new watermark, and then writes the rows. One run is killed after the second step. The scheduler retries it, the retry succeeds, and every later run succeeds too.

Five transactions from that hour are never loaded. The retry read the watermark the killed run had already moved, found nothing newer, and wrote nothing. No run failed after the crash itself, and no later run will ever look at those rows again.

#The companion code

Everything below comes from data-engineering-design-patterns, part 2. It uses the same Docker setup as part 1: Postgres 16 in one container, and Python 3.11 with Spark 3.5 and Delta Lake 3.2 in another. From the part's folder:

docker compose build
docker compose up -d postgres
docker compose run --rm lab python postgres_loads.py
docker compose run --rm lab python delta_loads.py
docker compose run --rm lab python -m unittest -v test_incremental

Each scenario starts with an empty source and target and runs three hourly runs. Before each run, the script commits the source rows that arrived in that hour. After every step it writes the whole target to out/<engine>/<scenario>/ as a CSV, and its report lists the source rows missing from the target and any duplicated in it. The reports used in this post are in example-results.

#The data

The source is one Postgres table, src_transactions, with an updated_at column. Eleven rows arrive over three hours:

Two rows matter more than the others. T-08 sits exactly on the boundary between two hourly intervals. T-09 is stamped 10:58, but in some scenarios its transaction is still open while run 2 reads the source, and it commits only afterwards. Both are ordinary: a timestamp rounded to the minute, and a transaction that took a few seconds longer than the one after it.

#The variant that breaks: the watermark moves first

The load in the opening story keeps its watermark in a small table and does three things per run:

select * from src_transactions where updated_at > :watermark;
 
begin;
update etl_watermark set watermark = :newest_updated_at_read;
commit;
 
begin;
-- write the rows to the target
commit;

Run 2 is killed between the two commits:

The retry reports success. It read the rows after 11:00, found none, and wrote none. Run 3 then loads T-10 and T-11 normally, and the five rows from run 2 stay missing for good. Nothing in the pipeline knows they exist, because the only record of what has been loaded is the watermark, and the watermark says they were.

The cause is that the watermark is a claim about the target: everything up to this value has been written. Committing it before the write makes the claim before it is true. Every crash between the two commits turns into permanent loss.

#Writing first

Swapping the order makes the opposite claim safe. If the rows are written first and the run dies before the watermark moves, the retry reads the same rows again. Whether that is harmless depends on the write:

This is part 1 again. A crash before the watermark commit turns into a rerun, and a rerun is only safe when the write is idempotent. With a plain append, the order trades lost rows for duplicated ones. With a keyed write, the retry replaces exactly the rows it wrote before.

#Both in one transaction

When the target and the watermark live in the same database, the ordering question goes away. The keyed write and the watermark update go into one transaction:

begin;
delete from fact_transactions where txn_id = any(:txn_ids_read);
insert into fact_transactions (...) values (...);
update etl_watermark set watermark = greatest(watermark, :newest_updated_at_read);
commit;

Killed between the write and the update, the transaction rolls back. The target still holds run 1's four rows, the watermark still says 09:55, and the retry loads all five rows. A reader never sees the rows without the watermark, or the watermark without the rows.

greatest stops the watermark from ever moving backwards, which matters in the next section, where a run deliberately re-reads rows older than the watermark.

#The variant that breaks: a late commit

The one-transaction load survives every crash, and it still loses a row. In this scenario nothing crashes. T-09 is inserted by another session that stays open while run 2 reads the source:

Postgres did nothing wrong. A row is invisible to other sessions until it commits, and updated_at records when the row was written, not when its transaction committed. By the time T-09 is visible, the watermark has already moved past 10:58, because T-07 and T-08 committed first.

A watermark built from the newest updated_at read assumes that rows become visible in updated_at order. No OLTP database promises that. Any two transactions that overlap can commit in either order, and the watermark skips whichever one is slower.

#The fix: a lookback window

The direct fix is to read a little behind the watermark on every run:

select * from src_transactions where updated_at > :watermark - interval '10 minutes';

Run 3 reads T-07 and T-08 again, and T-09 for the first time, along with its own rows. That is only safe because the write is keyed on txn_id. A lookback on top of an append would duplicate the overlap on every run. The lookback and the idempotent write come as a pair.

The window has to be longer than the longest transaction that writes to the source table. Ten minutes covers this example. It does not cover a batch job in the application that holds a transaction open for an hour, and nothing in the load can detect that it was too short. The lookback reduces the risk to a window you choose, and the choice needs to be written down next to the load.

#Data intervals instead of a watermark

A scheduler such as Airflow gives every run a data interval: run 2 owns 10:00 to 11:00, whenever it actually starts. In Airflow these are data_interval_start and data_interval_end. In the companion code, a small Run object carries the same two values, so the scenarios run without Airflow. The load reads exactly that window and replaces exactly that window, which is part 1's delete-then-insert with the interval as the key:

select * from src_transactions
where updated_at >= :interval_start and updated_at < :interval_end;
 
begin;
delete from fact_transactions where interval_start = :interval_start;
insert into fact_transactions (...) values (...);
commit;

There is no watermark table, so there is nothing to move too early. A run depends only on its own interval, so any run can be retried, or rerun a week later, and produce the same rows for that window.

The late commit still affects it:

T-08 is missing after run 2 only because it belongs to run 3's interval, and run 3 loads it. T-09 belongs to run 2's interval, which had already closed when it committed, so no scheduled run will ever read it. The difference from the watermark is what fixing it takes. Rerunning run 2 picks up T-09, and rerunning it again changes nothing. With a watermark, a missed row has no run that owns it. With intervals, it has exactly one, and that run is safe to repeat.

In practice that means the interval load needs a scheduled rerun of recent intervals, which is the interval form of the lookback. Part 10 on backfills and reprocessing covers how to do that without readers seeing a half-rebuilt day.

#The variant that breaks: the closed interval

The interval bounds need the same care as the delete key in part 1. With between, both ends are inclusive:

where updated_at between :interval_start and :interval_end

T-08 is stamped exactly 11:00, so it falls inside both run 2's interval and run 3's. Each run deletes only its own interval's rows, so neither removes the other's copy, and a rerun of either replaces only its own. The interval has to be half-open, >= start and < end, so that every instant belongs to exactly one run. Timestamps rounded to the second or the minute land on the boundary far more often than real-time clocks suggest.

#Delta Lake: the watermark and the rows in two systems

On Delta Lake, the source and the watermark table stay in Postgres and the target is a Delta table. No transaction spans the two, so the one-commit version of the load is not available. The keyed write is a replaceWhere on the txn_ids being written, the one-commit delete-then-insert from part 1:

(
    batch.write.format("delta")
    .mode("overwrite")
    .option("replaceWhere", f"txn_id IN ({txn_ids})")
    .save(path)
)

Moving the watermark first fails exactly as it did on Postgres: the same five rows go missing for good. The commit history shows why the retry looked successful. It made no commit at all:

Writing first and committing the watermark afterwards survives the same crash, with the target matching the source after every step. When the watermark and the rows cannot share a transaction, write first and make the write idempotent is the only order that works.

A replaceWhere with a list of keys is fine at this size. For a large batch, the keyed write is a MERGE, which is part 4.

#Reading the watermark from the target

A third option removes the watermark table. Each run reads the newest updated_at already in the Delta table and subtracts the lookback:

newest = spark.read.format("delta").load(path).agg(F.max("updated_at")).first()[0]
rows = extract_after(conn, newest - LOOKBACK)

That watermark cannot run ahead of the rows, because it is computed from them. In this scenario T-09 is in flight during run 2, and run 3 is killed before it writes:

The crash leaves the watermark at 11:00 without any special handling, because nothing was written. The retry reads after 10:50, which picks up T-09 along with run 3's own rows.

The cost is the query. Reading max(updated_at) scans the target unless the table's statistics or layout make it cheap, and on a large table that needs checking. It also ties the watermark to one target: two targets loaded from the same source each need their own.

#Side by side

The last two rows differ from the others in one more way. A watermark load has no notion of "the rows for 10:00 to 11:00", so the only rerun it offers is "start again from this watermark". An interval load can rebuild any past hour on its own, and repeat it.

#Checks worth keeping

The companion code's tests state the claims above as assertions:

• Moving the watermark before the write loses T-05 to T-09 for good, on both engines.
• Writing first with an append duplicates the same five rows.
• Writing first with a keyed write, and the one-transaction load, both end with the target matching the source.
• A watermark taken from the newest row read misses T-09, and a 10-minute lookback recovers it.
• The interval load misses T-09 until its interval is rerun, and rerunning it twice gives the same table.
• An inclusive interval end loads T-08 twice.
• On Delta, the watermark read from the target survives both a crash and the late commit.

The check I would add to a real pipeline is a reconciliation between source and target for a closed window: count, and ideally sum, the source rows for yesterday and compare them with the target. Every failure in this post passes silently on each individual run, and a comparison between the two sides is the only thing that sees them. That is part 14.

#What is next

Every run here read a few rows from one hour. The first run of a new pipeline has to read years. Part 3 is the initial load versus the daily load: loading the history through the same pipeline that runs every day, and what drifts when the history load is a separate one-off script.

#Summary to save

• An incremental load needs a rule for which rows are the next batch: a stored watermark, or the data interval the scheduler assigns the run.
• A watermark is a claim that everything up to it has been written. Never commit it before the write it describes. A crash between the two loses the rows for good, and every later run succeeds.
• When the watermark and the target share a database, commit the write and the watermark in one transaction. When they do not, write first, make the write idempotent, and move the watermark afterwards.
• A watermark taken from the newest updated_at read assumes rows commit in timestamp order, and they do not. Re-read a lookback window behind the watermark, and key the write so the overlap is replaced rather than duplicated.
• The lookback has to be longer than the longest transaction that writes to the source. Write the chosen value down next to the load.
• A data interval removes the watermark state: each run reads and replaces its own window, so any past run can be repeated. Late commits still need a scheduled rerun of recent intervals.
• Make intervals half-open, >= start and < end. A closed interval loads every boundary row twice.
• On Delta Lake, the watermark can be read from the target itself, which cannot run ahead of the rows. Check what the max query costs on a large table.