CHANGES design doc: reduce declaration, snapshot skips, sharing analysis

Extend the changelog-algebra section with three design outcomes: the
next step of declaring the netting as an ordinary count(*) reduce over a
ChangesRaw Z-set leaf (with the consolidate_pact rendering as the
time-in-key lowering pattern, the fusion case becoming stock reduce
algebra, and the analysis/escape-guard costs); the conditions under
which the snapshot fetch can be skipped (maintained by construction,
one-off via predicate pushdown into SnapshotMode, SUBSCRIBE's explicit
spelling); and the two sharings across reads of one input (raw
arrangement for the netting across starts, SemigroupVariable-bounded
windowed packed arrangement across maintained readers — the prerequisite
shape for indexes on changelogs). Mark per-consumer reinterpretation as
merged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: antiguru

doc/developer/design/20260602_changes_table_function.md

@@ -239,6 +239,28 @@ Prefer composing these from idiomatic differential pieces — `SemigroupVariable
 The pattern is established twice: the self-correcting persist sink (`correction = desired ∪ negate(written feedback)`) and the monotonic TopK retraction loop (`src/compute/src/render/top_k.rs`, `SemigroupVariable` + delayed negated feedback), which keeps arrangement state proportional to the *output* rather than the input.
 The same feedback shape is the tool whenever a changelog consumer needs "current window" or "latest tick" state rather than full accumulation.
 
+**Declaring the netting as a reduce (next step).**
+Make the algebra explicit in the IR: a `ChangesRaw` leaf (schema `(cols, mz_timestamp)`, deltas in the multiplicity) with an *ordinary* `count(*) GROUP BY all columns` reduce above it, planted by the planner.
+Today's consumer-site rendering (`consolidate_pact` + pack) becomes the lowering pattern for "reduce whose group key contains the reified time, over `ChangesRaw`" — chosen because the generic reduce is unbounded there (profile 1).
+The payoff is that the fusion case stops being bespoke: `sum(mz_diff [* f]) BY k` is ordinary reduce-of-reduce algebra — `ReductionPushdown` collapses it and the surviving reduce lowers as a stock accumulable reduce, diff-weighted natively (the `mz_message_counts` plan, profile 3) — and filters on input columns *and* `mz_timestamp` push through the count by reduce algebra rather than changelog-specific reasoning.
+The costs, eyes open: the analysis knowledge inverts (`ChangesRaw` is genuinely non-monotonic and non-non-negative; the packed form's append-only-ness must be *derived* from the reduce-with-time-in-key pattern, or the monotonic min/max rendering win regresses), and `ChangesRaw` must never escape un-reduced to a peek, sink, TopK, or non-fused consumer (planner invariant plus conservative analyses — no keys, no reduction elision — with a test that tries to break it).
+The timestamp advancement stays confined to the leaf's rendering, which must reify the event time and advance differential times to the `as_of` so the generic semantics line up while the specialized lowering nets *before* reification — not nice, but it checks out.
+
+**When the snapshot can be skipped.**
+A maintained read never needs it, by construction: the window is `(start, now]` and the snapshot sits at or below `start` (the current `SnapshotMode::Exclude`, unless a direct reader shares the import).
+A direct read always needs it — its accumulation *is* the contents.
+A one-off read can skip it exactly when a predicate provably discards it, which is a pushdown into `SnapshotMode`: `mz_timestamp > <start>` (the snapshot collapses to exactly `start`); any predicate implying `mz_diff < 1` (snapshot diffs are net contents, always ≥ 1 — a deletions-only audit skips the whole fetch); and the future `SUBSCRIBE` integration's `WITH (SNAPSHOT = false)` is the explicit spelling.
+Worth having: the snapshot fetch+consolidate dominates a one-off changelog read over a large input, and in the reduce framing the `mz_timestamp > start` case is just a filter pushed below the count into the leaf's read.
+
+**Sharing across reads of one input.**
+Two distinct sharings, with different tools.
+Sharing the *netting* across different starts needs no feedback loop: the raw arrangement's trace is self-bounding (deletes cancel, `~|input|`) and answers accumulation-at-any-time, so one arrangement serves every start — trace at `s'` for the snapshot, batch stream for `t > s'`.
+Sharing the *packed output* across maintained readers is where `SemigroupVariable` earns its keep: the packed form has the time reified into data, so arranging it naively is the unbounded profile-1 trace, but a delayed negated feedback retracting entries as they age past the *widest live window* bounds it to `~window` — the monotonic TopK shape, state proportional to what can still matter.
+N maintained reads (and a future *index on a changelog*) then share one windowed packed arrangement, each applying its own narrower window filter at read-out, instead of N temporal-filter copies each feeding its own downstream arrangement.
+Caveats: the feedback retraction *defines* window contents, so it must be exact and `event_delay(widest_window)`-driven (the TopK loop's wall-clock `system_delay` suffices there only because its retractions are an optimization), and it must be the single owner of the trailing edge — today's per-read `mz_now() < ts + lag` temporal filters are the same loop in MFP form, and two mechanisms must not both own it.
+One-off reads gain nothing (no maintenance; `consolidate_pact` holds ~nothing).
+The endgame shape: one input → one raw trace + one windowed packed trace + N readers.
+
 ### Remaining work
 
 Recorded with enough detail to be picked up in a later session.
@@ -250,7 +272,7 @@ Recorded with enough detail to be picked up in a later session.
 * **Optimizer integration.** `Changes` is deliberately an opaque barrier leaf: every analysis assigns it a conservative explicit value, and the changelog `Get`'s LIR MFP is withheld from the persist source operators (guard in `Plan::refine_source_mfps`), so beyond the pushed filters the full shard is read and processed only afterwards. With predicate pushdown done and the append-only nature now reflected in the `Monotonic` and `NonNegative` analyses (every update is emitted with diff +1 and nothing is retracted at the node — aging out of a maintained window retracts in the temporal filter *above* it, which the analyses handle), the remaining sound relaxations, roughly in value order: projection/demand pushdown (only demanded input columns need be read; `mz_timestamp`/`mz_diff` are appended by the reinterpretation); per-analysis precise values (arity/types are exact, but e.g. uniqueness and cardinality are maximally conservative); statistics for join planning. Bounding the shard read by the window's upper edge is the same idea as `start` bounds the lower edge. This is also groundwork for arbitrary expressions below — a true IR operator needs the optimizer to reason through `Changes`, not around it.
 * **Arbitrary expressions.** `CHANGES((SELECT ...) AS OF ...)` beyond a bare read of a persist-backed collection. The subquery form is the syntactic entry point, but anything that filters or transforms reopens the time-invariance problem deferred in "Alternatives" ("True table function in the IR"): we know how to *produce* changelog output from uncompacted data, not how to optimize queries with `Changes` in arbitrary positions while meeting user expectations. `plan_changes` currently peels an identity projection / empty map and requires a global `Get`; generalizing means either materializing the derived query first (today's workaround: `CREATE MATERIALIZED VIEW`, then `CHANGES` over it) or making `Changes` a true IR operator over arbitrary inputs, with the optimizer-barrier consequences catalogued there.
 * **Hold introspection (option A).** The lagged holds live inside the compute controller and are invisible in the catalog. The recorded future direction is adapter-level `ReadPolicy` composition (`ReadPolicy::Multiple` with per-dependent bookkeeping) or, more cheaply, an `mz_internal` relation exposing per-dataflow dependency holds, so the cost a `CHANGES` MV imposes on its input is attributable. Tracked outside this design as [CLU-104](https://linear.app/materializeinc/issue/CLU-104) — the ask (attribute read holds in introspection) is not specific to `CHANGES`.
-* **Per-consumer reinterpretation.** Prototyped on this branch; the paragraphs below record the design and where the prototype deviates. Rendering used to reinterpret a changelog import *once, at the import site*, and bind only the packed stream under the input's id — which forced two restrictions. First, reading a collection both directly and via `CHANGES` in one dataflow is rejected (`ChangesMixedRead`): both reads lower to indistinguishable LIR `Get`s of the same id, and the direct read would consume packed changelog rows (a real bug before the rejection: `SELECT (SELECT count(*) FROM r), (SELECT count(*) FROM CHANGES (r AS OF ...))` returned the changelog count twice). Second, multiple one-off reads of one input share the *earliest* start outright. Neither restriction is information-theoretic: the raw import stream read at the earliest start determines everything — advance times to `as_of` for a direct read, advance to a read's own `start` + consolidate + pack for its changelog (post-*packing* filtering cannot re-derive a later snapshot collapse, but pre-packing time advancement can). Lifting both means moving the reinterpretation from the import site to the consumer side. What that entails:
+* **Per-consumer reinterpretation.** Implemented (merged from the prototype); the paragraphs below record the design and where the implementation deviates. Rendering used to reinterpret a changelog import *once, at the import site*, and bind only the packed stream under the input's id — which forced two restrictions. First, reading a collection both directly and via `CHANGES` in one dataflow is rejected (`ChangesMixedRead`): both reads lower to indistinguishable LIR `Get`s of the same id, and the direct read would consume packed changelog rows (a real bug before the rejection: `SELECT (SELECT count(*) FROM r), (SELECT count(*) FROM CHANGES (r AS OF ...))` returned the changelog count twice). Second, multiple one-off reads of one input share the *earliest* start outright. Neither restriction is information-theoretic: the raw import stream read at the earliest start determines everything — advance times to `as_of` for a direct read, advance to a read's own `start` + consolidate + pack for its changelog (post-*packing* filtering cannot re-derive a later snapshot collapse, but pre-packing time advancement can). Lifting both means moving the reinterpretation from the import site to the consumer side. What that entails:
   * *Import.* Read raw at the **minimum** start over all reads of the input. `SnapshotMode::Include` iff any consumer needs the snapshot — any direct read or any one-off changelog read; `Exclude` only when every consumer is a maintained changelog read (preserving today's skip of the snapshot fetch). The error stream advances to the `as_of` once, at the import.
   * *Per-read specs.* The import's changelog metadata becomes a map from read site (the LIR `Get`'s `LirId`) to `{start, snapshot}` plus the import-level minimum. The three start-resolution sites (sequencer at creation, `as_of_selection` at restart, command-history reduction at reconnect) keep operating on import metadata only — no plan traversal; rendering looks the spec up by the `Get`'s own `LirId`. Controller holds and validation use the import-level minimum; maintained holds keep the widest window (unchanged).
   * *Consumers.* The changelog `Get` becomes distinguishable in LIR (a `GetPlan` variant or the spec lookup itself). A changelog read with start `s`: advance times to `s` (collapsing its snapshot), net per `(row, time)` (`consolidate_pact`), pack, advance emission times to the `as_of`; a maintained read drops times `<= s` instead of collapsing. A direct read: `advance_by(as_of)` and nothing else — no netting needed, accumulation handles unconsolidated fragments. The advance restores the times-advanced-by-`as_of` contract per consumer that the raw import no longer provides.