Pick Up New Changes Without Losing Late Readings

Botie’s pipeline remembers the latest event time and asks for anything newer on the next run. A delayed sensor message arrives with yesterday’s event time, so it never enters the training source. A later correction has the same problem.

We need to track when the source exposes a change, as well as when its measured event happened. The lesson builds a versioned merge and a checkpoint that can be replayed after a failure.

Select changes after the committed cursor, merge each identity by version, then keep resulting state and cursor together.

{
  "state": [
    {
      "id": "R1",
      "version": 2,
      "deleted": false,
      "value": 22
    },
    {
      "id": "R2",
      "version": 2,
      "deleted": true,
      "value": null
    }
  ],
  "cursor": 12
}

Event time belongs to the measured phenomenon. Arrival order describes when a change becomes available to the pipeline. Version orders successive states of one identity. A late event can have an old event time and a new arrival sequence.

Our source guarantees unique increasing sequence values and monotonic versions per identity. A production source must document its ordering, transaction and retention behaviour. A poll over updated_at needs careful tie handling and cannot detect hard deletions unless another mechanism records them.

Do not turn MAX(event_time) into a reliable ingestion cursor by naming it a watermark.

The checkpoint is the last source change position whose effects are committed in the sink. Read only later positions for the next batch. If the sink commits data but loses the checkpoint update, a replay can occur. If the checkpoint commits first, a crash can skip data that was never written.

Commit both in one transaction when they share a database. Across systems, use a delivery protocol and idempotent merge that tolerate replay, and reconcile partial outcomes. Exactly-once claims require specifying the entire boundary and its assumptions.

The exercise’s checkpoint table and current table share one SQLite transaction.

A stable primary key identifies the record. A newer version replaces its stored payload; an older version does not. Replaying an equal version with the same payload leaves the result unchanged.

The exercise assumes equal versions always carry equal payloads because conflicting records have already been quarantined. A real pipeline must check that contract. Choosing a random last arrival for equal conflicting versions destroys reproducibility. The workshop rejects that conflict explicitly.

Source position and record version solve different ordering problems.

CREATE TABLE current_readings(id TEXT PRIMARY KEY NOT NULL,version INTEGER NOT NULL,deleted INTEGER NOT NULL,value REAL);
INSERT INTO current_readings VALUES('R1',1,0,20),('R2',1,0,30);
CREATE TABLE checkpoint(seq INTEGER NOT NULL); INSERT INTO checkpoint VALUES(10);
CREATE TABLE changes(seq INTEGER PRIMARY KEY,id TEXT NOT NULL,version INTEGER NOT NULL,deleted INTEGER NOT NULL,value REAL);
INSERT INTO changes VALUES(11,'R1',2,0,22),(12,'R2',2,1,NULL),(13,'R3',1,0,18);

INSERT INTO current_readings VALUES('R1',0,0,99)
ON CONFLICT(id) DO UPDATE SET version=excluded.version,value=excluded.value
WHERE excluded.version>current_readings.version;
SELECT id,version,value FROM current_readings WHERE id='R1';

If a deleted identity simply disappears from a snapshot, an incremental consumer may never learn what to remove. A deletion event or tombstone tells the sink that this identity was deliberately removed at a particular version.

Retaining that tombstone prevents an older late update from recreating the deleted record. Our current_readings table stores deleted as 0 or 1 and keeps a NULL value for a tombstone. Readers of active data must explicitly filter deleted=0. Tombstone retention needs to cover the source’s replay and late-arrival assumptions.

A deletion is data about an identity. It is not equivalent to a missing sensor value.

A batch may contain an arrival whose version is older than the current record. Its payload is ignored, but its source position was still processed. Advance the cursor past it once the batch commits. Otherwise the consumer may read the same obsolete change indefinitely.

An empty batch leaves the cursor unchanged. Never move a checkpoint backwards because an old source extract was supplied. The SQL exercise uses the greater of the previous checkpoint and the maximum available change sequence.

The resulting cursor describes processed delivery positions, not only rows whose values changed.

A busy source can keep producing changes while the consumer works. Capture a source-supported upper position for the batch, then read positions after the committed cursor up to that bound. Commit the result before starting the next interval. The source must retain the interval and expose complete changes within it; reading an arbitrary MAX from a changing business table does not establish those properties.

Timestamp polling adds another problem. If two rows share the final timestamp and a row limit splits them, resuming with only updated_at greater than that timestamp skips the remaining row. An ordered timestamp-and-identity cursor can paginate a stable set, but it still cannot detect a later change that reuses an earlier timestamp. Reliable polling needs a source contract, an overlap-and-deduplication policy where suitable, or a change log. Our teaching input uses unique source sequence numbers so that these assumptions are visible.

The exercise receives a complete finite extract. A production reader must additionally define how it obtains and retains a complete bounded interval.

A historical backfill may revisit a time range already represented in features. Give it a declared source snapshot and transformation version, and merge by stable identity or rebuild a partition deliberately. Do not mix a backfill cursor into the live stream unless its ordering contract supports that.

Corrections can invalidate previously built aggregates or training examples. Decide whether to recompute affected partitions, keep immutable historical snapshots, or publish a new dataset version. Retain enough lineage to explain which source revisions produced a model’s features.

Idempotent row merging does not automatically refresh every derived aggregate downstream.

Suppose the sink stopped at change position 100, but the source has already discarded everything before 150. Resuming at 150 would silently omit an interval. The consumer needs to detect that its checkpoint is older than the retained history and take a documented recovery path.

A new full snapshot can rebuild the sink, but the snapshot and following change stream need a consistent handoff position. Otherwise changes that occur during the copy can be missed or duplicated. Preserve the old dataset version while rebuilding when it is needed for model reproducibility. Reconcile the rebuilt state before directing readers to it.

A successful connection does not prove that the source can still replay the interval the sink needs.

Periodically compare record counts, keys and selected aggregates against a trusted source snapshot. Missing change retention, a broken cursor or a source schema change can produce a plausible but incomplete sink. Monitor lag and checkpoint progress separately from job success messages.

The final challenge replays a batch, introduces an older version after a deletion, and tests an empty source extract. These are ordinary recovery cases to design before the first production outage.

The pipeline is ready for the next lesson when its checkpoint and data tell the same story after a restart.