Build an Incremental ETL Load with a Watermark Column

1. Verify the candidate watermark is actually monotonic.

psql -d source -c "SELECT count(*) FROM (SELECT updated_at, lag(updated_at) OVER (ORDER BY id) prev FROM events) t WHERE prev IS NOT NULL AND updated_at < prev;"

Expected output: A count of zero proves updated_at never moves backward, so it is safe as a watermark; any nonzero count means late updates will be missed.

2. Create the watermark store keyed by pipeline name.

psql -d warehouse -c "CREATE TABLE IF NOT EXISTS etl_watermarks (pipeline text PRIMARY KEY, watermark timestamptz NOT NULL, rows_copied bigint DEFAULT 0, updated_at timestamptz DEFAULT now());"

Expected output: The one-row-per-pipeline table exists, holding the last safely-copied position for this load.

3. Copy only rows past the watermark, in bounded batches.

-- increment.sql
WITH wm AS (
  SELECT watermark FROM etl_watermarks WHERE pipeline = 'events' FOR UPDATE
), batch AS (
  SELECT * FROM source.events
  WHERE updated_at > (SELECT watermark FROM wm)
  ORDER BY updated_at
  LIMIT 50000
)
INSERT INTO warehouse.events
SELECT * FROM batch
ON CONFLICT (id) DO UPDATE SET
  payload = EXCLUDED.payload, updated_at = EXCLUDED.updated_at;

Expected output: Each run copies at most 50k new-or-changed rows; the FOR UPDATE lock serializes concurrent runs of the same pipeline.

4. Advance the watermark only as far as the batch actually copied.

UPDATE etl_watermarks w
SET watermark = b.max_ts,
    rows_copied = w.rows_copied + b.n,
    updated_at = now()
FROM (
  SELECT max(updated_at) max_ts, count(*) n
  FROM warehouse.events
  WHERE updated_at > (SELECT watermark FROM etl_watermarks WHERE pipeline = 'events')
) b
WHERE w.pipeline = 'events'
  AND b.max_ts IS NOT NULL;

Expected output: The watermark moves to the maximum timestamp actually present in the destination, never to the wall clock, so a crashed batch leaves nothing behind.

5. Verify completeness with a source-versus-destination count.

psql -d source -t -c "SELECT count(*) FROM events WHERE updated_at <= (SELECT watermark FROM dblink('dbname=warehouse', 'SELECT watermark FROM etl_watermarks WHERE pipeline = ''events''') AS t(watermark timestamptz))" 2>/dev/null || echo "run: SELECT count(*) FROM warehouse.events vs source up-to-watermark"

Expected output: Rows-up-to-watermark counts match between source and destination, proving the incremental load lost nothing.

Problem

Rows arrive with old timestamps after being copied

Solution

Late-arriving backdated rows sit below the watermark forever; add a daily full-diff on id hashes, or run a periodic backfill sweep for updated_at <= watermark with a content checksum comparison.

Problem

Hard deletes at the source are never propagated

Solution

Watermarks copy inserts and updates only; capture deletes via a soft-delete flag, a CDC log, or a nightly anti-join that removes ids missing from the source snapshot.

Problem

Two job instances run at once

Solution

The FOR UPDATE lock makes the second wait; if your engine lacks row locks, use an advisory lock or a lease column with a heartbeat, and keep the batch small so waits stay short.