Handling Late-Arriving and Updated Records in a Cleaning Pipeline

Most cleaning pipelines quietly assume that a day's data is complete when the day's job runs. It rarely is. A mobile app uploads events when the phone gets signal, a store's till syncs overnight, a supplier corrects an invoice a week later, an order moves from placed to shipped to refunded. Records arrive late, and records that already arrived change.

A pipeline that ignores this produces numbers that drift from the truth without any error. This post covers how to recognise the two problems, keep the timestamps that let you handle them, and choose a pattern for each: lookback windows for late rows, latest-version rules and change feeds for updated ones. The SQL was run on DuckDB 1.5.1 and the Python on pandas 3.0.2. It builds on idempotent data cleaning, because every reprocessing pattern here has to be safe to run twice.

Key Takeaways

• Store two timestamps on every record: when it happened (event_at) and when your pipeline received it (loaded_at). Report by the first, load incrementally by the second.
• An incremental filter on event time silently skips late rows. Filter on load time, or reprocess a lookback window of recent days.
• For updated records, keep the latest version per key with a deterministic order, or apply the source's change feed including deletes.
• Decide up front whether closed periods are restated when late data arrives, and write the decision down.

What are late-arriving and updated records?

A late-arriving record is one whose event happened in a period your pipeline has already processed. An updated record is a new version of something already loaded: for example, the same key with a different status, amount or address. Both mean the data for a past period is not final when the job first runs.

The underlying issue is the gap between event time and processing time. Google's Dataflow Model paper, which shaped Apache Beam, argues that pipelines should stop expecting unbounded data to become complete at some point and should be built on the assumption that they may never know when all the data for a period has arrived (Akidau et al., "The Dataflow Model", VLDB 2015). Batch cleaning pipelines face the same thing on a slower clock: yesterday's file is almost never everything that happened yesterday.

There is a third, related case: a late-arriving reference row. An order arrives for customer C-77 before the customer record does. That is an orphan today and a valid row tomorrow.

Why do you need both event time and load time?

Event time tells you which period a record belongs to. Load time tells you which records your pipeline has already seen. You need the first for correct reports and the second for correct incremental loads, so store both, as instants with a time zone offset.

Here are four events. E3 happened late on 31 October but reached the pipeline on 2 November:

SET TimeZone = 'UTC';

CREATE TABLE events AS
SELECT * FROM (VALUES
  ('E1', TIMESTAMPTZ '2026-10-31 22:10:00+00', TIMESTAMPTZ '2026-10-31 22:15:00+00', 40.00),
  ('E2', TIMESTAMPTZ '2026-11-01 08:00:00+00', TIMESTAMPTZ '2026-11-01 08:01:00+00', 15.00),
  ('E3', TIMESTAMPTZ '2026-10-31 23:50:00+00', TIMESTAMPTZ '2026-11-02 06:30:00+00', 25.00),
  ('E4', TIMESTAMPTZ '2026-11-02 09:00:00+00', TIMESTAMPTZ '2026-11-02 09:02:00+00', 10.00)
) t(event_id, event_at, loaded_at, amount);

Totals by the month the event happened, and by the month it was loaded:

SELECT strftime(event_at, '%Y-%m') AS event_month, sum(amount) AS by_event_time
FROM events GROUP BY ALL ORDER BY ALL;

SELECT strftime(loaded_at, '%Y-%m') AS load_month, sum(amount) AS by_load_time
FROM events GROUP BY ALL ORDER BY ALL;

┌─────────────┬───────────────┐     ┌────────────┬──────────────┐
│ event_month │ by_event_time │     │ load_month │ by_load_time │
├─────────────┼───────────────┤     ├────────────┼──────────────┤
│ 2026-10     │         65.00 │     │ 2026-10    │        40.00 │
│ 2026-11     │         25.00 │     │ 2026-11    │        50.00 │
└─────────────┴───────────────┘     └────────────┴──────────────┘

October's real revenue was 65.00. A report grouped by load time says 40.00 and pushes the difference into November. Late events belong to the month they happened in. When a source sends local timestamps without an offset, fix that first; the lesson on instants, offsets and local time explains why.

Why does an event-time high-water mark lose rows?

An incremental load usually keeps a high-water mark: the largest timestamp it has processed, so the next run only reads newer rows. If that mark is on event time, a late row whose event time is older than the mark is never read.

Say the last run finished with a mark of 2026-11-01 08:00, the event time of E2. The next run reads rows after it:

SELECT event_id FROM events
WHERE event_at > TIMESTAMPTZ '2026-11-01 08:00:00+00'
ORDER BY event_id;

┌──────────┐
│ event_id │
├──────────┤
│ E4       │
└──────────┘

E3 arrived after the last run, but its event time is earlier than the mark, so it is skipped now and on every future run. Nothing errors. Put the mark on load time instead (here the last run's mark is E2's loaded_at, 08:01):

SELECT event_id, event_at FROM events
WHERE loaded_at > TIMESTAMPTZ '2026-11-01 08:01:00+00'
ORDER BY event_id;

┌──────────┬──────────────────────────┐
│ event_id │         event_at         │
├──────────┼──────────────────────────┤
│ E3       │ 2026-10-31 23:50:00+00   │
│ E4       │ 2026-11-02 09:00:00+00   │
└──────────┴──────────────────────────┘

Both new rows come through, and E3 still carries its October event time for reporting. A load-time mark can still miss rows if ingestion transactions commit out of order or several rows share the mark's exact loaded_at, so re-read a small overlap before the mark and deduplicate by key, or track batch ids instead (the exercise load only rows after the high-water mark practises the mark itself). The load-time column has to be stamped by something you control (the ingestion step), not by the source, or the same problem comes back. The exercises apply a change feed and late events belong to their own month practise the patterns below; load a new batch only once practises the companion rule: track which batches were processed so a re-delivered file is not loaded twice.

How do lookback windows handle late data?

When you cannot get a reliable load timestamp, or when downstream tables are aggregates by event date, reprocess a fixed window of recent days on every run. Each run deletes and rebuilds the last N days, so any late row that arrives within N days is picked up:

CREATE TABLE daily (day DATE, total DECIMAL(38,2));

BEGIN TRANSACTION;
DELETE FROM daily WHERE day >= DATE '2026-11-02' - INTERVAL 3 DAY;
INSERT INTO daily
SELECT event_at::DATE AS day, sum(amount) AS total
FROM events
WHERE event_at >= DATE '2026-11-02' - INTERVAL 3 DAY
GROUP BY ALL;
COMMIT;

FROM daily ORDER BY day;

┌────────────┬───────┐
│    day     │ total │
├────────────┼───────┤
│ 2026-10-31 │ 65.00 │
│ 2026-11-01 │ 15.00 │
│ 2026-11-02 │ 10.00 │
└────────────┴───────┘

31 October now includes E3. Three things make this safe:

1. The delete and the insert use the same window, inside one transaction, so the step is idempotent: running it twice gives the same table. See the glossary entry on idempotency.
2. The window comes from the run date parameter, not current_date, so a backfill rebuilds the right days.
3. The window length comes from data, not habit. Measure how late rows actually arrive and pick N to cover nearly all of them. Rows later than the window need a separate catch-up, or an agreed rule that they are counted in the period they arrived.

Measuring lateness is one query. On the four sample events:

SELECT list_transform(
         quantile_cont(epoch(loaded_at - event_at) / 3600, [0.5, 0.95, 0.99]),
         x -> round(x, 1)) AS hours_late_p50_p95_p99
FROM events;

┌────────────────────────┐
│ hours_late_p50_p95_p99 │
├────────────────────────┤
│ [0.1, 26.1, 29.7]      │
└────────────────────────┘

On real data, run it over a few weeks of loads; the 99th percentile in hours, rounded up to whole days, is a reasonable first window.

How do you keep the latest version of an updated record?

When a source sends a new row every time a record changes, keep the most recent version per key. The rule needs three parts: a key, an ordering column, and a tie-breaker so the same row wins on every run.

CREATE TABLE order_versions AS
SELECT * FROM (VALUES
  ('O-1', 'placed',  49.90, TIMESTAMPTZ '2026-11-01 10:00:00+00'),
  ('O-1', 'shipped', 49.90, TIMESTAMPTZ '2026-11-02 14:00:00+00'),
  ('O-2', 'placed',  20.00, TIMESTAMPTZ '2026-11-01 11:00:00+00'),
  ('O-2', 'placed',  18.00, TIMESTAMPTZ '2026-11-01 11:30:00+00'),
  ('O-3', 'placed',  12.00, NULL)
) t(order_id, status, amount, updated_at);

SELECT order_id, status, amount, updated_at
FROM order_versions
QUALIFY row_number() OVER (
  PARTITION BY order_id
  ORDER BY updated_at DESC NULLS LAST, status, amount
) = 1
ORDER BY order_id;

┌──────────┬─────────┬────────┬──────────────────────────┐
│ order_id │ status  │ amount │        updated_at        │
├──────────┼─────────┼────────┼──────────────────────────┤
│ O-1      │ shipped │  49.90 │ 2026-11-02 14:00:00+00   │
│ O-2      │ placed  │  18.00 │ 2026-11-01 11:30:00+00   │
│ O-3      │ placed  │  12.00 │ NULL                     │
└──────────┴─────────┴────────┴──────────────────────────┘

NULLS LAST matters: DuckDB 1.5 already sorts nulls last in both directions, but other databases put them first in a descending sort, and a version with an unknown update time should never beat one with a known time. Stating it keeps the rule portable. The trailing status, amount only breaks exact ties; if the source has a version number or sequence, use that instead. The exercises unknown times never win and keep the current price per product practise this rule, and pick the current row, then filter covers the order-of-operations trap: filter after choosing the latest version, or a record whose latest status is "cancelled" will fall back to an older "placed" row.

The pandas version sorts by the same columns, including the tie-breakers, and keeps the first row per key:

import pandas as pd

versions = pd.DataFrame({
    "order_id": ["O-1", "O-1", "O-2", "O-2", "O-3"],
    "status": ["placed", "shipped", "placed", "placed", "placed"],
    "amount": [49.90, 49.90, 20.00, 18.00, 12.00],
    "updated_at": pd.to_datetime(
        ["2026-11-01 10:00", "2026-11-02 14:00", "2026-11-01 11:00",
         "2026-11-01 11:30", None], utc=True),
})

latest = (
    versions
    .sort_values(["order_id", "updated_at", "status", "amount"],
                 ascending=[True, False, True, True],
                 na_position="last")
    .drop_duplicates(subset="order_id", keep="first")
    .reset_index(drop=True)
)
print(latest)

  order_id   status  amount                updated_at
0      O-1  shipped    49.9 2026-11-02 14:00:00+00:00
1      O-2   placed    18.0 2026-11-01 11:30:00+00:00
2      O-3   placed    12.0                       NaT

If you need to know what a record looked like on a past date, keep every version with valid-from and valid-to timestamps rather than only the latest; the exercise pick the version in force on a date works with exactly that shape.

How do you apply a change feed with deletes?

Some sources send changes rather than full rows: an operation code (insert, update, delete), the key, the new values and a change timestamp. Apply them with MERGE INTO, after reducing the feed to the last change per key so one target row never matches two source rows:

CREATE TABLE orders AS
SELECT * FROM (VALUES ('O-1', 'placed', 49.90), ('O-2', 'placed', 20.00), ('O-9', 'placed', 5.00))
t(order_id, status, amount);

CREATE TABLE changes AS
SELECT * FROM (VALUES
  ('O-1', 'U', 'shipped', 49.90, TIMESTAMPTZ '2026-11-02 14:00:00+00'),
  ('O-9', 'D', NULL, NULL,       TIMESTAMPTZ '2026-11-02 15:00:00+00'),
  ('O-4', 'I', 'placed', 31.00,  TIMESTAMPTZ '2026-11-02 16:00:00+00')
) t(order_id, op, status, amount, changed_at);

MERGE INTO orders AS t
USING (
  FROM changes
  QUALIFY row_number() OVER (PARTITION BY order_id ORDER BY changed_at DESC) = 1
) AS c
ON t.order_id = c.order_id
WHEN MATCHED AND c.op = 'D' THEN DELETE
WHEN MATCHED THEN UPDATE SET status = c.status, amount = c.amount
WHEN NOT MATCHED AND c.op <> 'D' THEN INSERT VALUES (c.order_id, c.status, c.amount);

The feed updates O-1, deletes O-9 and inserts O-4, leaving:

┌──────────┬─────────┬────────┐
│ order_id │ status  │ amount │
├──────────┼─────────┼────────┤
│ O-1      │ shipped │  49.90 │
│ O-2      │ placed  │  20.00 │
│ O-4      │ placed  │  31.00 │
└──────────┴─────────┴────────┘

Write the conditional delete before the general update. PostgreSQL's documentation states the standard behaviour: WHEN clauses are tested in the order written and the first one that is true is executed (PostgreSQL, "MERGE"). In a test on DuckDB 1.5.1 the result here was the same with the clauses swapped, but the delete-first order is the one that behaves the same everywhere. The same PostgreSQL page explains why the feed is reduced first: a target row should join to at most one source row (PostgreSQL, "MERGE"). The DuckDB syntax is in "MERGE INTO statement". Deletes are the part most often forgotten. A pipeline that only applies inserts and updates keeps cancelled or erased records forever, which is a correctness problem and, for personal data, sometimes a legal one.

What about late-arriving reference data?

When a fact row arrives before the row it refers to, do not drop it and do not invent the parent. Load the fact, report the missing key, and resolve it on a later run when the reference arrives. In practice that means one of two patterns:

• Hold and retry. Put the orphan in an exceptions table with a reason, and re-check it on each run until the parent appears or an agreed time limit passes.
• Load with an explicit unknown. Load the fact with the key intact and let reports show it under "unknown customer" until the customer row lands. Never replace the real key with a placeholder, or you cannot join it later.

Either way, a referential check should report how many orphans exist and how old they are. The lesson report orphans, don't drop them and the free exercise referential integrity with explicit exceptions cover both halves.

Should past reports change when late data arrives?

That is a business decision, and it must be made explicitly. There are two defensible policies:

Many teams do both: restate in the analytical tables, freeze in the published report, and show the difference as an adjustment line. Whatever you choose, store loaded_at so you can always reproduce what a report said as of a given date, and state the policy in the report itself.

Frequently asked questions

What is a late-arriving fact?

A fact row (an event, transaction or measurement) that reaches the pipeline after the period it belongs to has already been processed. It should be reported in the period of its event time, which means that period's totals have to be recomputed or adjusted.

How long should a lookback window be?

Long enough to cover the delays you actually observe. Compute loaded_at - event_at for recent data and choose a window that covers nearly all rows, then decide what happens to the few that arrive even later.

Is keeping the latest version the same as deduplication?

They use the same window-function pattern, but the intent differs. Deduplication removes copies of the same fact; latest-version logic chooses between different versions of the same record. Mixing them up can delete legitimate updates or keep stale ones.

Related reading

• QUALIFY in DuckDB: The Cleanest Way to Keep the Latest Row
• Cleaning Timesheet and Payroll Data: Hours, Shifts and Overtime

Sources

• Tyler Akidau et al., "The Dataflow Model: A Practical Approach to Balancing Correctness, Latency, and Cost in Massive-Scale, Unbounded, Out-of-Order Data Processing", Proceedings of the VLDB Endowment 8(12), 2015.
• DuckDB, "MERGE INTO statement".
• PostgreSQL, "MERGE": order of WHEN clause evaluation and the one-source-row-per-target-row rule.
• DuckDB, "QUALIFY clause".

All SQL was run on DuckDB 1.5.1 and the Python on pandas 3.0.2; outputs are from those runs, with type rows removed and two result tables shown side by side in the event-time example.