Implement CHANGES table function for changelog reads

[antiguru]

Motivation

Closes #4527

Users want to consume the change stream of a collection as a relation they can transform with SQL. SUBSCRIBE already exposes this stream, but only as a top-level streaming statement whose output flows directly to the client — it cannot be wrapped in a SELECT, aggregated, joined, or materialized.

This PR implements the CHANGES table function, which reads a collection's changelog as a relation with the per-update timestamp and diff available as ordinary mz_timestamp / mz_diff columns.

Syntax

CHANGES(<relation> AS OF [AT LEAST] <bound>)

• <relation> is a collection named directly or a parenthesized subquery (mirroring SUBSCRIBE). A subquery must reduce to a bare read of a single persist-backed object (table, source, materialized view); anything that filters/transforms is the deferred arbitrary-expression case and is rejected. A view is rejected in both forms — it is not persist-backed, and a non-materialized view is not inlined at this planning stage (the subquery reduces to a bare Get of the view itself).
• AS OF vs AS OF AT LEAST — strict (error if the requested start is below the input's since) vs advisory (age in: clamp the bound up to since). Reuses the existing AsOf grammar; the parens around a subquery delimit it from this trailing AS OF.
• fixed vs sliding bound — a constant timestamp is a static changelog start (append-only); an mz_now()-relative bound is a sliding window that trails the query time by its lag.

The bound's shape transparently signals the semantics; the design deliberately reuses AS OF (Materialize's logical-time read-hold meaning, distinct from SQL:2011), resolving the design's "parameter keyword" question.

Where it is allowed

context	fixed bound	sliding (mz_now()-relative) bound
one-off SELECT	✅ implemented	✅ implemented (as_of = query_time − lag; window [query_time − lag, query_time])
materialized view	❌ rejected (would pin since indefinitely)	✅ implemented (advisory ages in; strict requires a fully retained window at creation)
index	❌ rejected	🚧 follow-up (same machinery, gate not yet relaxed)

Implementation

• Parser / AST (src/sql-parser): TableFactor::Changes { relation: ChangesRelation, as_of: AsOf, .. }, where ChangesRelation is Name | Query (mirrors SubscribeRelation). AS OF [AT LEAST] clause required.
• Planner (src/sql/src/plan/): plan_changes resolves the relation (a name, or a subquery reduced to a bare global Get) to a single persist-backed collection, classifies the bound (strict/advisory, fixed/sliding), gates by lifetime (a fixed bound is rejected in a maintained context), and builds the changelog schema (input columns + mz_timestamp + mz_diff). Feature-flagged by enable_changes_table_function.
• IR & lowering (src/expr, src/sql/src/plan/): MirRelationExpr::Changes { id, typ, bound, strict } — an opaque leaf carrying the lower-bound scalar; lowers to a Get of the source import marked as a changelog read.
• Coordinator, one-off (src/adapter/src/optimize/peek.rs): evaluates each changelog bound at the query time (resolving mz_now()), clamps an advisory (AS OF AT LEAST) bound up to the input's since, and marks each source import ChangelogMode::OneShot { start } with its own resolved start. The dataflow as_of stays the query time, so mz_now() elsewhere in the query resolves to the query time; the controller installs each import's read hold at its start.
• Compute (src/compute/src/render.rs): the changelog import binds the raw stream, read from the import-level start, and each consumer reinterprets it at its Get site (GetPlan::Changelog): advance the raw event times to the read's own resolved start (collapsing its own snapshot), net per (row, time) via consolidate_pact (the arrangement batcher without a retained trace), pack each netted (row, time, diff) as the append (pack(row, time, diff), max(time, as_of), +1) — append-only, matching SUBSCRIBE's diff format — then apply the MFP to the extended rows. A direct read of the same import advances the raw times to the as_of (yielding the input's contents), so a query may read a collection both directly and via CHANGES, and multiple one-off reads of one input collapse their own snapshots at their own starts. A maintained import includes the snapshot only when a direct read shares it; the changelog reads drop times at or below the start themselves. The one-off start travels as resolved_start on the MIR Changes node (the bound stays as written, keeping EXPLAIN deterministic). The changelog Get's LIR MFP stays on the Get and is not hoisted into the persist source operators (guard in Plan::refine_source_mfps).
• Optimizer (src/transform/): Changes is an opaque barrier leaf in every analysis; each analysis must give it an explicit value rather than reading a (nonexistent) input — e.g. the NonNegative analysis (whose catch-all otherwise underflows for a leaf at post-order index 0). One pushdown is wired up: non-temporal predicates on input columns commute with the changelog reinterpretation, so optimize_dataflow_filters pushes them into the changelog source import (only when common to every read of the input), where persist_source applies them before the reinterpretation — part-stats pruning over retained history. Predicates on the appended mz_timestamp/mz_diff columns stay above the leaf.
• Maintained materialized views (see "Maintained materialized views: sliding execution" and the "Implementation map" in the design doc): imports carry ChangelogMode::Maintained { window, start }; the MV optimizer extracts the constant window lag and wraps each changelog read in the temporal filter mz_now() < mz_timestamp + lag; the read start is resolved as join(since, as_of − window) at creation, environment restart (as_of_selection), and replica reconnect (command-history reduction); the compute controller lags the import's read hold by the window so a restart can reproduce the persisted contents exactly; rendering skips the snapshot and replays deltas from the start. The window is capped by the new changes_max_window system variable (default 1 day), enforced at creation only so lowering the cap cannot wedge existing objects at bootstrap. A strict sliding bound (AS OF, vs AS OF AT LEAST) requires a fully retained window: the sequencer advances the view's as_of to since + window when the input retains that much history, and errors otherwise — never silently serving a partial window. Start resolution reads the inputs' collection since via fresh read holds (the reused transaction holds sit at the query timestamp and hid retained history, over-clamping advisory starts too).

Verification

• test/sqllogictest/changes.slt: feature gating; argument validation (persist-backed only; views and CTEs rejected, by name and via subquery); bound validation (non-constant / out-of-range rejected); durable-lifetime gating; advisory clamp; sliding mz_now() plan + run; maintained-MV creation and rejection cases (strict sliding, month intervals, REFRESH); the changes_max_window cap; predicate-pushdown EXPLAIN shapes (pushed, not pushable, blocked by an unfiltered sibling read); mixed direct + CHANGES reads (direct, scalar subqueries, through an inlined view, MV creation).
• test/testdrive/changes.td: data round-trips — snapshot at as_of plus later inserts / deletes / updates as appends with correct mz_diff; aggregation composition; sliding bound; subquery form; CHANGES over a materialized view; strict AS OF reading back into a RETAIN HISTORY window; a maintained sliding-window MV aging in and appending deltas; pushed and non-pushable filters in one-off and maintained reads; the mixed direct + changelog data round-trip; same-input reads with distinct bounds collapsing their own snapshots; mixed maintained views (strict look-back where the direct side needs the snapshot, and an advisory join).
• ChangesMaterializedView platform check (misc/python/materialize/checks/all_checks/changes.py): end-to-end restart-exact reproduction — exact row multisets catch spurious snapshot rows, cross-phase mz_timestamp ordering catches a re-snapshot — across the restart/upgrade scenarios, including 0dt. Verified locally with RestartEntireMz.
• Parser round-trip tests; pack_changelog_row, NonNegative-leaf, and as_of_selection changelog-constraint unit tests.

Follows the design doc (doc/developer/design/20260602_changes_table_function.md).

Follow-ups (tracked in the design doc's "Remaining work")

• Indexes — same machinery, relax the gate and wire start resolution into index as-of selection.
• SUBSCRIBE — the one-off mode with a session-length hold.
• Hold introspection — surface the lagged dependency holds (adapter-level read-policy composition or an mz_internal relation).
• Arbitrary-expression CHANGES — generalize the subquery form beyond a bare read (the time-invariance / optimizer-barrier work).

https://claude.ai/code/session_018drjq1ZvKQgh931ix5vv1M

[antiguru]

Thorough review focused on correctness, dedup/refactoring, docs, and test gaps. This is a large, carefully-built PR — excellent design doc, focused regression tests for the two optimizer panics it flushed out, and unusually complete fan-out across the transform analyses. Findings below, ordered by severity.

Correctness

1. (High confidence) mz_now() in a one-off query body containing CHANGES resolves to the changelog start, not the query time

src/adapter/src/optimize/peek.rs (~L360–386). The dataflow as_of is set to the query timestamp and then overridden down to the changelog start:

df_desc.set_as_of(timestamp_ctx.antichain());              // as_of = query time
if let Some(changelog_as_of) = changelog_as_of {
    df_desc.set_as_of(Antichain::from_elem(changelog_as_of)); // as_of = start (earlier!)
}
let as_of = df_desc.as_of....into_option()...;             // = start
let style = ExprPrepOneShot { logical_time: EvalTime::Time(as_of), .. }; // mz_now() -> start

Before this PR as_of == timestamp_ctx.timestamp() invariably, so logical_time was the query logical time. The override breaks that invariant: any mz_now() elsewhere in the same one-off query now folds to the changelog start, while the peek still samples at the query timestamp. So SELECT mz_now() FROM CHANGES(t AS OF 5) LIMIT 1 would return 5, and temporal filters in the body would be evaluated at the start. The changelog's own bound is fine (resolved earlier from timestamp_ctx.timestamp()). Fix: derive logical_time from timestamp_ctx.timestamp(), not from the post-override as_of. No test exercises this combination.

2. (Medium) Multiple CHANGES reads with distinct bounds in one query collapse to the minimum as_of

peek.rs computes changelog_as_of = min(all starts) and pins the single dataflow as_of to it; every import is flagged ChangelogMode::OneShot, which carries no per-import start. render.rs's OneShot path then reads each flagged source at that one dataflow.as_of. So ... FROM CHANGES(c AS OF 100), CHANGES(d AS OF 50) reads c from 50 — spurious rows with mz_timestamp ∈ [50,100). Same for sliding bounds with different lags. Latent semantic bug, untested; at minimum note it in the design doc's "sharp edges," ideally give OneShot a per-import start like Maintained { start }.

Documentation inaccuracies

3. "Identical to SUBSCRIBE" overstates the equivalence

Design doc ("identical to SUBSCRIBE") and comments in relation.rs / query.rs ("matching SUBSCRIBE's diff format/shape"). Comparing to src/sql/src/plan/statement/dml.rs (~L1547–1566), the real SUBSCRIBE output differs three ways: column order (SUBSCRIBE prepends mz_timestamp, mz_diff; CHANGES appends), mz_timestamp type (SUBSCRIBE Numeric{scale 0} vs CHANGES MzTimestamp), and mz_diff nullability (SUBSCRIBE nullable(true) vs CHANGES nullable(false)). The CHANGES choices are reasonable, but say "the same columns, trailing, with mz_timestamp typed as mz_timestamp" rather than "identical."

4. pack_changelog_row doc comment is wrong for the maintained path

render.rs (~L1844): "time has already been advanced to max(time, as_of) by persist_source…". True only for OneShot. In Maintained, the source is read at start below as_of with the snapshot excluded, so the time written into mz_timestamp can be less than as_of — that's the point. Since this helper serves both paths, the comment is misleading exactly where it matters.

5. Gating error message names the wrong contexts

PlanError::ChangesRequiresSlidingBound displays "CHANGES in a materialized view or index requires a sliding bound", but plan_changes routes any maintained non-sliding lifetime here — including plain CREATE VIEW (tested at changes.slt:100), Subscribe, and Source (all is_maintained()). A CREATE VIEW user sees a message about MVs/indexes; and per the design's own lifetime matrix SUBSCRIBE is supposed to allow a fixed bound. Generalize the wording ("a maintained object") or special-case the view message.

Minor / pedantic

6. Double-planning the bound. plan_changes plans the bound via plan_changes_bound, then re-plans it via plan_as_of_or_up_to for fixed bounds purely to validate, discarding the result. Two passes over one expr; comment notes it's validation-only.
7. Subquery alias dropped. CHANGES((SELECT a AS x FROM t) AS OF …) reduces to Get(t); output column is a, not x (the rename lives only in the discarded scope).
8. Parser grab not feature-gated. parse_table_factor treats CHANGES ( specially regardless of enable_changes_table_function; a relation named changes immediately followed by ( would fail to parse even with the flag off. Low risk, but broader than the gated feature.
9. Zero-lag sliding bound. AS OF AT LEAST mz_now() (no subtraction) passes contains_temporal() → in an MV yields window = 0 and filter mz_now() < mz_timestamp. Degenerate, accepted, untested — reject or document.
10. Input column named mz_timestamp/mz_diff yields duplicate column names in the RelationDesc (no dedup). Same latent issue as SUBSCRIBE, so not a regression, but unvalidated/untested.

Test gaps

• No EXPLAIN test for the new Changes explain-text in either the HIR (sql/src/plan/explain/text.rs) or MIR (expr/src/explain/text.rs) formatter. An EXPLAIN SELECT … FROM CHANGES(…) slt case would lock in Changes <id> as_of=… / as_of_at_least=….
• No test for mz_now() in the body of a one-off CHANGES query (would surface #1).
• No test for multiple CHANGES with distinct bounds in one query (#2).
• No error-stream round-trip coverage for the reinterpretation (errs.delay(...) in render.rs).

Done well

• Every transform analysis gets an explicit Changes arm rather than relying on a catch-all, and the two real panics this surfaced (NonNegative leaf at post-order index 0; missing MreDiff arm) both have focused regression tests.
• The scalar bound is correctly excluded from the scalar visitors (relation.rs try_visit_scalars_mut1 et al.), so prep never resolves mz_now() inside the leaf prematurely — consistent with the opaque-leaf design and per-path resolution.
• The as_of_selection soft-constraint ordering (apply after the hard downstream constraint so a conflict degrades to the advisory fallback rather than overshooting) is subtle and well-tested (changelog_storage_constraints{,_conflict}).
• Creation-time-only window-cap enforcement, with the explicit rationale that bootstrap re-optimization must not wedge on a lowered cap, is a good call and documented in three places.

---

Generated by Claude Code

[antiguru]

Review findings addressed in a4fc08b.

#1 + #2 (one-off as_of override) — fixed together: ChangelogMode::OneShot now carries a per-import resolved start, mirroring Maintained. Rendering reads each changelog import at its own start (snapshot included); the dataflow as_of stays the query time, so mz_now() in the query body resolves correctly. The controller installs each import's read hold at its start; a strict bound below since still fails dataflow creation. Caveat: multiple one-off reads of the same input share the earliest start — one import, one snapshot collapse, and a per-read mz_timestamp filter cannot re-derive a later read's collapse — recorded as a sharp edge. New tests: mz_now() in the body, two CHANGES with distinct bounds in one query (asserting the later input's snapshot sits exactly at its own bound).

#3 — equivalence claims corrected in the design doc and plan_changes: same columns as SUBSCRIBE but appended, mz_timestamp-typed, non-null mz_diff.

#4 — pack_changelog_row comment rewritten for both paths (time may be below the as_of; that's the point).

#5 — message generalized to "CHANGES with a fixed bound is only supported in one-off SELECT queries", which stays accurate for views/sources and remains correct until SUBSCRIBE (where fixed bounds become allowed) is wired up.

#9 — zero-lag sliding bound kept (degenerate but coherent: empty window, only new changes enter) and locked in with an slt test. #7/#10 recorded as sharp edges in the design doc. EXPLAIN output locked in via slt for fixed and sliding bounds.

Not changed: #8 — the parser runs before the catalog exists, so it cannot consult feature flags; this matches every other gated syntax. A relation named changes is reachable by quoting. #6 — the second pass is plan-time validation of a constant expression; cheap and already commented. Error-stream round-trip test deferred: erroring a persist-backed input deterministically in testdrive (and then reading it back through CHANGES) needs more scaffolding than the slot warrants right now; the errs.delay path is exercised by the maintained MV tests structurally.