sql: Add correlated x AS OF SYSTEM TIME t SINCE l reads over pinned history

[frankmcsherry]

Draft, PR 3 of 3, stacked on #38749 (CHANGES). Until the earlier PRs merge this diff includes their commits; review the last commit. This one is a reference implementation for discussion, not a merge candidate.

What it does

SELECT o.id, p.price
FROM orders o
LEFT JOIN LATERAL (
  SELECT price FROM prices AS OF SYSTEM TIME o.ts SINCE 1788538608103
  WHERE prices.sym = o.sym
) p ON true;

x AS OF SYSTEM TIME expr SINCE lit is a table factor whose expr may refer to the enclosing query: the contents of x at virtual time expr. Each order sees the price at its own time, and that value never moves afterwards. A time before lit yields the empty relation (NULL under a LEFT JOIN), a future time delays the row until then, and replicas and restarts agree because x's history is pinned.

How, and why it is a reference implementation

It desugars to a correlated query over CHANGES and is planned as a LATERAL derived table:

SELECT cols FROM (
  SELECT cols, sum(mz_diff)::bigint AS c FROM CHANGES(x AS OF lit)
  WHERE mz_timestamp <= expr AND expr <= mz_now() GROUP BY cols
), repeat_row(c)

That is the definition of "the state at t" as the integral of the changelog up to t, so the semantics are inherited from CHANGES rather than specified twice, and the two features cannot disagree about what history means. It also gives a differential-test oracle for the operator this should become.

The cost is the problem. As a relational join plus reduce, the plan arranges the facts (twice) and keeps one accumulator per (fact, dimension value it scanned), so state grows with facts times versions. A one-sided lookup does not need any of that: the output for (key, t) depends only on the dimension's history at or before t, which is frozen once the dimension's frontier passes t, so a fact can be looked up once and forgotten (retractions recompute the same value from the pinned history). differential_dogs3::operators::half_join already stashes records until the arrangement frontier passes them and compares trace times against a per-record time; the variant needed compares against the payload time (for late facts), emits at max(capability, payload time), and accumulates per key at discharge so it can emit the NULL row. The planner would lower the correlated shape to that operator instead of decorrelating. State becomes proportional to the dimension's history since the pin plus facts still waiting for their time.

Evidence (local instance)

• Orders at three times against a price sequence 1 -> 2 -> 3 got 1, 2, 3; an order before the pin got NULL; an order an hour in the future was withheld and later appeared with the value at its time. Bumping every price by 100 six days later changed none of the existing rows; a new order saw 103.
• Against a hand-written SCD-2 lookup (ORDER BY ts DESC LIMIT 1 over an explicit history table) on 10k orders and a 21k-row history: identical results, comparable one-shot latency, one third of the arrangement memory (194k vs 590k records). The dedicated operator would sit near orders plus history, about 31k.
• Materialized, one-shot, and both across a restart: identical.

Things to discuss

• Where the AS OF sits: on the table factor (this PR) or on the subquery (SELECT dim.val AS OF facts.time ... FROM dim). Equivalent for one input; the subquery form is friendlier, the table-factor form composes per input.
• SINCE lit versus AT LEAST lit, and whether a time below the pin yields empty (this PR, per the design note) or clamps to the snapshot at the pin. One token in the desugared predicate either way.
• The SINCE literal is baked into each dependent, so advancing retention means recreating dependents with a later literal before the pin can move.
• Desugaring goes through SQL text and repeat_row, which sits behind enable_repeat_row; a real implementation would build HIR directly.

Tests

Parser round-trip cases in src/sql-parser/tests/testdata/select and create.

Release note: none.

🤖 Generated with Claude Code