Temporal joins and future leakage

How do you join two things that both change over time, and why is the wrong join the cheap one?

When both sides of a join change over time, an equality join on the key alone is underspecified. You must say as of when. There are two mechanisms.

Interval join (stream with stream)

An interval join pairs records where |tA − tB| ≤ w. The window width is at the same time the meaning of the join and its memory bill: both sides are held for w, and nothing can be emitted as complete until the watermark passes tA + w (what a stream join keeps).

Versioned or as-of join (stream with table)

For each probe row, take the version of the dimension table whose validity interval contains the probe's event time. This requires the dimension to be versioned, and the probe's own timestamp to be used.

-- For each delivery, the courier rating that was valid when it happened
SELECT d.parcel_id, d.event_time, r.rating
FROM deliveries d
JOIN courier_rating_history r
  ON  r.courier_id = d.courier_id
  AND r.valid_from <= d.event_time
  AND d.event_time  < r.valid_to;

Future leakage

The failure has one name worth remembering: future leakage. Joining with the current version instead of the as-of version puts into the row information that did not exist when the event happened.

• In reporting, results cannot be reproduced.
• In machine learning, the model scores beautifully offline and fails in production, because in production the future is not available. A large share of what is called training-serving skew is exactly this bug.

Leakage does not degrade the offline metric. It inflates it. That is why it survives code review: the numbers look better, so nobody looks for a problem.

This was the data scientist's problem. The model had seen a courier's rating after the delivery.

Availability time, not event time

valid_from ≤ event time is still not enough. A value computed by a nightly batch job is not known until the batch runs: an event at 09:00 sees data as of the previous midnight, not as of 08:59. Joining on event time when the true availability was nine hours later is still leakage, just smaller. The correct predicate uses availability time, which is a property of the pipeline, not of the data, so the pipeline's schedule becomes part of the meaning of the data it produces. And because the join needs to know when each value became known, a table that overwrites values in place has destroyed what the join needs: the history must be stored bitemporally, append-only.

The incentive to be wrong

The correct meaning is "the latest version with valid_from ≤ event time". That is an inequality join with an argmax (pick the row with the largest value). No hash join does this directly. Engines run it as a sort-merge over both sides. So the correct answer is the expensive plan and the wrong answer is the cheap and simple one. Systems drift towards the bug without anyone choosing it.