compute: two-runtime read isolation (interactive runtime) - #37770
compute: two-runtime read isolation (interactive runtime)#37770antiguru wants to merge 118 commits into
Conversation
antiguru
changed the title
f90cdae to
6728fc8
Compare
This is an architectural commitment, not just a featureBefore we land this, it's worth being explicit that two-runtime is not a "sure, let's do it" change. It commits the compute layer to a new shape and is effectively a one-way door. This comment lays out the trade-off honestly so the decision is made with eyes open. What it actually isTwo runtimes do not add CPU and do not magically "isolate" reads — both runtimes share the same cores. What the second runtime buys is one precise thing: a separate, OS-preemptible run loop, so an interactive read is no longer trapped behind a long, run-to-completion maintenance operator step on the same timely worker. The right way to frame it is separation of concerns, not isolation:
A single runtime cannot be both without compromising one of them. Two runtimes let each be pure. That is the real argument. Why the obvious alternatives don't apply"Just use a read replica." Fails on the use case that matters most: introspection. A replica's introspection describes that replica, so it fundamentally cannot be served from another one — and you need it exactly when the maintenance runtime is pinned (hydration, batchy work), which is when it blocks. Today we fly dark precisely then. Nothing but an in-process second runtime fixes that. Separately, replicas mostly redline on memory, not CPU, so a second replica doubles the binding resource; the shared-arrangement approach here duplicates no memory. And because those boxes are memory-bound, the spare CPU the interactive runtime needs is exactly what is already idle — the CPU-saturated case (where two runtimes help least) is not the common one. "Just yield more finely in the one runtime." That is the anti-pattern above. It would degrade the maintenance runtime's core contract to buy interactivity it shouldn't be responsible for. The costs we are accepting
Why it's a one-way door
That irreversibility is the crux. It is acceptable only because the end-state — maintenance as a pure batch runtime, interactive as a pure reader — is the architecture we would choose deliberately, given the points above. The bar for merging is therefore not "we can back out." It is "we would design it this way on purpose." On the introspection-during-hydration and memory-bound-fleet facts, we would. Bottom lineWorth doing, but as a deliberate architectural commitment. The durable, defensible assets are (a) zero-copy cross-thread arrangement sharing as a primitive and (b) introspection that survives hydration — both survive every counter-argument. The framing to avoid leaning on is "reads get faster," which a CPU-bound box or the unsolved coordinator cap can each undercut. Reviewers should weigh this as a commitment to the two-runtime shape, not just as an isolated feature. |
| /// Export `index_id` as [`GlobalId::Transient`] rather than [`GlobalId::User`]. | ||
| /// | ||
| /// The multiplexer fronting a two-runtime `clusterd` (see | ||
| /// `mz_compute_client::multiplex`) routes a `CreateDataflow` to the interactive | ||
| /// runtime only when every export id is transient and the dataflow has no | ||
| /// subscribe sink; a `User` export always stays on maintenance. Setting this | ||
| /// is how a script drives a dataflow onto the interactive runtime, e.g. to | ||
| /// exercise a query dataflow that imports a maintenance index and is itself | ||
| /// served by the interactive slow path. | ||
| #[serde(default)] | ||
| transient: bool, |
There was a problem hiding this comment.
Alternative would be to allow setting transient global ids based on the name, using the canonical GloablId string representation.
| fn threshold_shared_trace<'scope, T: RenderTimestamp>( | ||
| arrangement: Arranged<'scope, SharedOksEnter<T>>, | ||
| name: &str, | ||
| ) -> Arranged<'scope, RowRowAgent<T, Diff>> { |
There was a problem hiding this comment.
Could this function be generic over traces surfacing rows?
There was a problem hiding this comment.
Spiked. This is the same reduce_abelian higher-ranked output-key normalization wall documented on threshold_local: the bound only normalizes when the input and output trace types are concrete through the function signature, so a generic Tr: TraceReader does not compile. The three helpers (_local, _trace, _shared_trace) share identical bodies but must stay concrete for that reason.
Unifying them needs mz_reduce_abelian's key-container bounds reworked (the same class of fix as differential #797/#798's named key-container parameter), which is a broader reduce-plumbing change than this file. A macro could remove the textual duplication but adds indirection for three ~10-line functions without touching the type constraint. Leaving concrete and tracking the genericization with the reduce_abelian work rather than here.
| } | ||
| PeekStatus::NotReady => Some(peek), | ||
| PeekStatus::UsePeekStash => { | ||
| unreachable!("the interactive peek is never peek-stash eligible") |
There was a problem hiding this comment.
This is a limitation that we shouldn't have (and don't document).
6c243c1 to
5abfb74
Compare
Architectural reviewRead against the design doc, the released differential-dataflow (not the fork), and the two-runtime discussion threads. Summary: the thesis is right and well argued, the skeleton (role split, multiplexer, placeholder-plus-adopt, notification-driven wakeups) is sound, but there is one place where a differential invariant is reimplemented with half of it dropped, and the form of the specialization is broader than the goal needs in ways that are cheap to narrow now and expensive to narrow later. 1.
|
Directional follow-upsOrdered work list derived from the review above, written so it can be picked up cold. Grouped by whether it blocks merge. File and line references are against BlockingB1. Make
B2. Restore peek stash on the interactive path, or stop routing stash-eligible peeks there.
Add a test that a peek result above the stash threshold still returns rows with the feature on. Today CI cannot catch this because test results never approach B3. Restore replica-side peek observability. The constraint from the design is that interactive must not run its own introspection indexes, since it serves introspection from maintenance's published copies. Peek events are a different thing. Either forward interactive's B4. Narrow peek routing. Where the signal lives is the open design question. The peek carries its timestamp, and the runtime knows the target index's published B5. Explain the arrangement-size 2x. If confirmed, the reporting bug and a real retention issue are the same bug, and the fix is to have the publisher refresh or drop its chain when the spine merges rather than only on stream activation. Revert the threshold bump once fixed. Non-blocking but wanted before this is on by defaultN1. Write down the reconciliation index-replacement argument, or add slot epochs. N2. Land the SQL-level introspection-during-hydration acceptance test. N3. Drop N4. Delete the machinery with no production caller. N5. Make the routing tripwire real. N6. Document the memory coupling. Documentation correctionsD1. PR body says an unpublished-import query dataflow is deferred and built later. The design doc says the opposite and correctly, and no deferred-build code remains. Fix the PR body, and fix the header comments in D2. PR body and design doc both describe multiplexer peek dedup with D3. Design doc: replace the "99.9% reduction in query latency" framing with the sealing-dependency scope. Two runtimes remove the serving cost from the maintenance step loop and do not remove the dependency on maintenance sealing the read timestamp ( D4. Design doc: soften " D5. Note wherever the flag is documented that flipping D6. Explicitly not in scopeCore affinity pins maintenance worker i and interactive worker i to the same core ( Generated by Claude Code |
Addendum: SUBSCRIBE is a stronger motivation than the doc allowsFollow-up to the two comments above. The value. A subscribe's initial snapshot is a full arrangement walk at The non-goal text is wrong. The doc says "all interactive work is single-time, so the shared import applies no What is actually missing is a live-following import, which was deleted as dead code in 740e4b7 ("remove the dead live import(), keep import_snapshot_at"). So the real prerequisite is reinstating an unbounded import whose capability tracks the publisher's frontier instead of dropping at A cheaper path that gets the stated win without that. Split the subscribe at
The work becomes stitching the two into one ordered The same shape applies to Suggested doc change: reframe SUBSCRIBE from "out of scope" to "the next intended consumer, snapshot first", and record the split-at- Generated by Claude Code |
B4 design note capturedWrote up the routing-policy question rather than leaving B4 as "resolve with a short design note first". On branch
Cherry-pick or copy it into this branch, whichever is less friction. It says up front to fold the resolution into Two things in it revise what I wrote in the follow-up list. The signal already flows through the multiplexer. I claimed the multiplexer has no frontier view. It does: every The default direction matters more than the signal. Route to interactive only on a positive hint, and leave maintenance as the fallback. That keeps real traffic on the maintenance peek path, which is the concrete answer to the doc's own "the capability we lean on will atrophy" argument, and it means peek stash and the peek introspection relations keep working for the reads that use them. Gated on a dyncfg with The note also sharpens why unconditional routing is the wrong default, beyond the observability losses: an unsealed read waits either way, but waiting on interactive is slightly worse than waiting on maintenance, because maintenance answers in the same step that advances the frontier past Same policy question applies to Generated by Claude Code |
Benchmark plan, and a correction to the routing noteFair challenge on the previous comment. My last open question ("should introspection reads be forced to interactive") was posed as a judgement call and it is not one. Replaced it with experiments.
Seven experiments, each with a pre-registered prediction, plus a decision table mapping every outcome to a consequence. Highlights below, but the methodology section matters more than the individual experiments and is where the tail question lands. On p99, p99.9, p99.99 and maxThree things worth flagging up front. The harness will report tail percentiles it cannot support. For the extreme tail, measure the mechanism rather than the percentile. The reason p99.99 and beyond matter here is specific: interactive is a single step loop, so a read arriving mid-scan waits for that step, and a parked read is re-examined only when the worker next reaches the top of its loop. The floor under the read tail is the interactive runtime's step duration distribution, and step duration has millions of samples per minute rather than thousands. So the tractable route to a claim about max latency is to measure step duration directly and treat it as the bound. That needs a per-runtime step-duration histogram, which does not exist. The Percentiles read off an overloaded open-loop arm are queueing artifacts. When the offered rate exceeds service capacity, latency grows roughly linearly with elapsed time and every percentile becomes a statement about where in the run the sample fell. That is the 80 second baseline p50. It establishes that the control was driven past capacity, and it cannot separate "isolation improves latency" from "we overloaded the baseline". Every latency arm should measure capacity first, then run at about 30% of it, and report utilization next to the percentiles. Your race example, worked throughThe two error directions turn out to be asymmetric, which narrows things considerably. Reported frontiers never regress, so the multiplexer's tracked frontier is never ahead of that process's true frontier. Therefore false negatives come from report lag and are the only error on a single-process replica, and a false negative costs a lost isolation opportunity rather than a park. False positives come only from the cross-process meet, because only process 0 receives a There is a stronger structural result. The multiplexer sees the same response stream the controller does, and sees it earlier being upstream. So for any read whose timestamp comes from the controller's read frontier (serializable, stale, introspection), the tracked frontier is at or beyond the controller's view, which is at or beyond the chosen If that holds, the dynamic hint earns its keep only by catching strict-serializable reads whose timestamp got sealed during the flight from timestamp selection to the replica. That single fraction decides whether to build the hint or just classify statically, and it is a measurement. It also means the obvious worry, that a frontier hint self-defeats under saturation because the same stalled workers emit the reports, does not apply to the reads that matter, since their timestamps derive from the same stalled reports and so hint and timestamp move together. The experiments
Correction to my earlier recommendationTwo changes to I wrote that waiting on interactive is "slightly worse" than waiting on maintenance for an unsealed read. More precisely the penalty is exactly one interactive step, which is small when the lane is quiet and unbounded when it is not, since the lane has no admission control. So that penalty is really a question about lane congestion rather than about routing, and it is a tail phenomenon rather than a median one. B-B measures it. And the recommendation itself ("default to maintenance, route to interactive only on a positive hint") is now marked as a hypothesis the benchmarks are meant to overturn, with the case against it stated explicitly: if misroute penalties concentrate in the tail, all-interactive with the collateral losses fixed beats a hinted policy that is right on average. Reversibility is a real consideration but it does not outrank a measured tail regression. Generated by Claude Code |
Correction to B1: no differential change needed, and the fix is smallerI said B1 needed the merged-versus-pending distinction reconstructed, and implied that might not be reachable from released differential since Why the distinction does not need reconstructingThe publisher cannot read the trace's physical frontier. But it does not need to read it, because it can bound it from above. Now chain that with the promotion rule.
Add the precondition the spine itself asserts, So the merged-versus-pending branch is unnecessary. The precondition alone makes it fall out. The current code's bug is not a missing branch, it is a missing precondition: the shared handle accepts cuts below the trace's physical frontier, which the local spine explicitly forbids. The deferral is conservative in the right directionThe remaining worry was that a reader's physical hold reaches the agent only on the publisher's next activation, since the trace handle is
The publisher always forwards a hold at or below what the reader subsequently cuts at. The one-activation lag is a correctness margin, not a hazard. Revised B1
Item 3 is the one that actually bites today, and it is roughly a one-line change. The commit that introduced it, 67c6ba0 ("seed shared-trace importers with the full snapshot, not a frontier-filtered one"), fixed a real data-loss bug by removing the The regression test from the original B1 still stands and is still the important part: publish, keep writing past What upstream would buy, if anyone is planning differential work anywayNeither of these is needed and neither should gate this PR.
Since the fix is local and the upstream items are optional, this should not need differential release planning. Generated by Claude Code |
5abfb74 to
51e690e
Compare
Staging evaluation: what the numbers say, and three fixes that came out of itMeasured on a staging region across builds Fixes in this pushKubernetes rejected the new port name.
An offloaded walk can divert to the peek response stash. The gate was avoidable. Also added E1: the offload removes head-of-line blocking between peeksPoint-lookup latency behind three concurrent scans, one
The inline tail tracks the walk it is queued behind; the offloaded tail is flat. 33x at the largest walk. Engagement is confirmed by the counter, E2: the offload alone captures itThe same sweep single-runtime gives max latencies of 107.5 / 109.2 / 180.4 / 167.6 ms. V1 equals V3 within noise at every scan cost, and V0 equals V2. The second runtime contributes nothing to peek tail latency. What removes the blocking is the substrate the walk runs on, and that is a dyncfg needing no restart, no new port and no fleet roll. E7: temporary dataflows are what justify the second runtimeThey cannot use the offload at all, since that is a peek walk substrate. Sampled during one 60M-row index hydration, both arms as replicas of one cluster sharing that hydration (possible because of the scoping fix above):
Tail halves and introspection stops being collateral damage of hydration. p50 barely moves, which is what an isolation claim predicts. So the two mechanisms are not competing: the offload for peeks, the second runtime for rendered temporary dataflows. The routing policy should follow that split. One caveat worth more than the ratio: late materialization has an 850-950 ms floor in every cell, quiet or loaded, either runtime, for ~120 rows. That is temporary-dataflow creation and teardown. It is larger than the tail placement recovers, and it is the bigger prize. E6: publication does not cost resident memoryImporting a published trace does not duplicate it. 48 concurrent interactive dataflows over a 95 MiB published index moved the resident set by 4.5 MiB on one replica and 1.4 MiB on the other. The arrangement-size doubling seen earlier on a 16-worker replica did not reproduce. Merge blocker cleared. Methodology notes, including what went wrongRecorded because each of these first produced a wrong answer.
Still open
Full write-up, including the failed fixtures and the pre-registered predictions, is in |
Swap: no regression, and the largest margin measuredFollow-up to the swap item left open above. The first attempt failed on the fixture rather than the question: building one huge index spikes the working set while it hydrates, which killed a replica and left the two arms at unequal swap depth (3.2 vs 6.4 GB), so it was not a matched control. This environment runs That settles at ~3.8 GB resident and 5.05 GB of swap on one arm against 5.08 GB on the other, about 2.2x memory, with zero restarts on both. Equal depth is what the earlier run lacked. Point-lookup latency on the resident index, two concurrent walks of a swap-resident index:
No regression under swap, and the margin is larger than anywhere else measured. Inline shows a 29 second worst case in one repeat and a 2.2 second median in the other; the offloaded arm stays at its quiet latency in both. The walks themselves agree: one swapped walk takes 3.6, 4.7 and 56.4 seconds inline against 2.3 seconds three times over offloaded. Faster and far more predictable on the same data at the same depth. Why the margin is largest here: a walk faulting on swapped pages spends its time blocked rather than computing, so the thread it blocks is not doing useful work either way. Moving it off the serving worker costs nothing and recovers everything, which is not true in the CPU-bound case where an offloaded walk still competes for a core. Disk pressure is the regime with the least to lose and the most to gain. One correction to the earlier comment: the inconclusive first attempt reported an inline walk failing to complete while the offloaded one finished, at unequal swap depths. With depths matched that asymmetry persists in latency but neither arm restarts, so the restart in the first attempt should be attributed to the hydration spike rather than to the walk substrate. |
The console under load: answers stay fast, and stay about a second staleThis is the case that motivates the second runtime for a UI. A console polls introspection, dataflow sizes and Everything below runs in
Two findings, and they answer different questions. Latency splits the arms hard. On the single-runtime replica a console poll takes seconds while an index hydrates, up to 13.3 s for Staleness does not split them. Both arms answer from a timestamp about one second behind real time while hydrating, against roughly zero when idle, and they are indistinguishable on this axis (170-1589 ms vs -141 to 1456 ms). So the second runtime buys prompt answers at the same freshness, not fresher ones. That is worth stating precisely in the design doc, because "a console showing data about a second old during hydration" is a weaker and more accurate claim than "a console showing live data". Two reading notes: values straddling zero are clock skew between the probe host and the environment plus round trip, so idle staleness means "no measurable lag" rather than a negative number. And Write-up in |
The skewed point lookup: one bad key stops stalling everyoneA real customer pattern, and the sharpest case for the offload — it needs none of the stash work to be reachable, because Shape: Isolated: a normal lookup is 105 ms (round trip), the hot key is 2020 ms — finding the minimum of six million values for one key is one worker's walk. Load is open loop: arrivals fire on a fixed wall-clock schedule regardless of outstanding work, from a pre-opened pool. A closed loop throttles itself and hides exactly this queueing. Client-side queue delay stayed at 1.2 ms with zero drops in every run, so the pile-up is entirely server side. Three skewed lookups injected into a steady 10/s stream:
Same at 25/s, two injections:
The inline timeline shows the mechanism, not just its size. After a skewed lookup arrives at t=5.0, the normals behind it complete at a fixed instant rather than after a fixed delay: 2019.9 ms for the arrival at 5.0, then 1921, 1821, 1721, 1622, 1522 … down to 235.8 ms for the arrival at 6.8. Each one waits out the remaining walk. Identical pattern at 12.0 and 19.0. That is the complaint exactly: one lookup on a bad key stalls every lookup behind it, and the victim count is arrival rate times walk duration — which is why it presents as a cluster-wide latency spike rather than one slow query. The offloaded arm has no such window. The skewed lookups still take ~2 s (1971, 2021, 2003 ms) because the work is unchanged; they just stop being in anyone else's way. This is the strongest argument here for defaulting the offload on: the cost is a thread, and the benefit is that a single skewed key stops being a cluster-wide latency event. Happy to encode this as a parallel benchmark so the pattern is regression-protected — say the word and I'll wire it into the existing harness rather than leaving it as a one-off script. |
2f73ff3 to
5826fea
Compare
Introduce `mz_row_spine::ArcBatch`, a local newtype around `Arc<B>` carrying differential's batch traits (the orphan rule forbids the blanket impl on a bare `Arc<B>`), and switch the production spines and their builders (`RowRowSpine`, `RowValSpine`, `RowSpine`, `ValRowSpine`, `ColValSpine`, `ColKeySpine`) from `Rc`/`RcBuilder` to `ArcBatch`/`ArcBuilder`. An `Arc`-backed batch whose contents are `Send + Sync` can be read from a thread other than the one maintaining the trace, which `Rc` cannot do. Only the batch handle becomes atomic; the batch contents are unchanged. Also add generic `ArcOrdVal`/`ArcOrdKeySpine` aliases for callers outside `mz_compute`, adapt batch-size logging (`log_arrangement_size_inner`) to reach through the newtype to the inner `Arc`, and switch the storage sink trace to the `Arc`-backed spine. Two consumers of the sink trace follow from that switch. The iceberg sink stashes input batches while it waits for their batch description, so its `VecDeque` and the `with_ready_batches` helper now hold `ArcBatch` rather than `Rc`. And the arrangement-size operator's cache comment named the `RcBox` allocation its `Weak` keeps reserved, which is an `ArcInner` once the batch handle is atomic. The invariant it documents is unchanged. This is the foundational primitive the two-runtime read-isolation work (MaterializeInc#37770) builds on, extracted here for standalone review. It builds against released differential-dataflow with no fork or `[patch.crates-io]`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
5826fea to
306799b
Compare
Introduce `mz_row_spine::ArcBatch`, a local newtype around `Arc<B>` carrying differential's batch traits (the orphan rule forbids the blanket impl on a bare `Arc<B>`), and switch the production spines and their builders (`RowRowSpine`, `RowValSpine`, `RowSpine`, `ValRowSpine`, `ColValSpine`, `ColKeySpine`) from `Rc`/`RcBuilder` to `ArcBatch`/`ArcBuilder`. An `Arc`-backed batch whose contents are `Send + Sync` can be read from a thread other than the one maintaining the trace, which `Rc` cannot do. Only the batch handle becomes atomic; the batch contents are unchanged. Also add generic `ArcOrdVal`/`ArcOrdKeySpine` aliases for callers outside `mz_compute`, adapt batch-size logging (`log_arrangement_size_inner`) to reach through the newtype to the inner `Arc`, and switch the storage sink trace to the `Arc`-backed spine. Two consumers of the sink trace follow from that switch. The iceberg sink stashes input batches while it waits for their batch description, so its `VecDeque` and the `with_ready_batches` helper now hold `ArcBatch` rather than `Rc`. And the arrangement-size operator's cache comment named the `RcBox` allocation its `Weak` keeps reserved, which is an `ArcInner` once the batch handle is atomic. The invariant it documents is unchanged. This is the foundational primitive the two-runtime read-isolation work (MaterializeInc#37770) builds on, extracted here for standalone review. It builds against released differential-dataflow with no fork or `[patch.crates-io]`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
The spec described the design as drafted, which the implementation diverged from three ways. A model describing a system nobody built reads as assurance, so it is worse than no model. Brought into line: the acquired hold follows its reader instead of sitting at the as_of, reclaim is driven by an explicit release the rendering runtime records rather than by watching a registration appear and go away, and the drop and the release are two commands on that runtime's stream in that order rather than one. Three things fell out of doing it. I1 was wrong for the shipped design. It would have flagged a correct downgrade, because once the reader advances the collection may legitimately compact past the original as_of. It is now two windows: protected at the as_of before the dataflow is built on a process, at the reader's current hold afterwards. Sabotaging the downgrade to overshoot its reader violates it, so it discriminates. The no-permanent-pin property cannot be an invariant. Reclaim and downgrade are separate steps, so there is always a legal state where the release has been applied and the hold is still there, and a safety version flags exactly that state. It is now a liveness property against a fair spec. Checking it surfaced a dependency nobody had written down: the release reaches the owning runtime only through the rendering runtime applying a command, so a runtime that stops draining its queue strands the hold forever. Fairness now states that the runtimes drain, which is what each worker's server loop does. MaintStep deliberately treats only the command-acquired hold as a bound and ignores the reader's own registration. That registration is forwarded through one agent whose setter joins, so it cannot represent a frontier below where the agent already sits. Treating it as a bound would assume away the ratchet the acquired hold exists for. HoldsReleaseOnMaint.cfg is new and keeps run one's refutation live: the release on the owning runtime's stream, which the asymmetry in the code exists to avoid. Both refutations still violate I1 and the runner fails if either stops.
Formatting only. The resolution of the upstream index-peek-rows metrics against the peek offload restructure left one initializer wrapped that fits on a line.
Upstream MaterializeInc#38170 removed mz-compute-client from clusterd as an unused dependency, which it was on main. This branch uses it: `lib.rs` constructs the two-runtime `Multiplexer` from it. A semantic conflict git cannot see. Both sides are individually correct and the files do not overlap, so the rebase reported nothing and the breakage only showed up building clusterd, which no per-crate check of the compute crates reaches.
`bin/doc --document-private-items` runs with `-D warnings`, so `rustdoc::private-intra-doc-links` is an error: a public module doc may not link to a private or crate-private item, because the link resolves only under `--document-private-items` and breaks without it. Both references become plain code spans. They point at internal mechanics that the surrounding prose already explains, so nothing is lost by not linking.
A published arrangement forwarded the published `since` as its physical compaction target. That conflates the two compaction kinds: logical compaction decides which times stay distinguishable, physical compaction decides which batches may merge. Holding physical down to `since` kept a batch boundary at every frontier at or above the controller's read frontier, so the spine stopped merging, and it did so for every published index whether or not anything ever imported it. Measured on an index with no importer at all: 5 to 7 batches where an unshared index reached 1. A reader does not need a boundary at its `as_of`. `import_snapshot_at` seeds it with the whole chain and wraps the handle in `TraceFrontier`, which advances times rather than cutting, so a batch straddling `as_of` is harmless. Batches are immutable, so a merge after a seed is captured cannot disturb it, and `acknowledged` starts at that seed's coverage and only rises. What a reader does need is a boundary at each frontier it later passes to `cursor_through`. `mz_join_core` already declares exactly that through `set_physical_compaction`, as differential intends, but `SharedTraceHandle` recorded the request in a local field and dropped it. Mirror it into the publication point as `physical_holds` instead, and forward their meet, falling back to the chain coverage when no reader is registered. A floor starts at the coverage at registration, never at `since` (a logical frontier, which is the conflation this commit removes) and never at `upper`, which can lead the coverage and so sit above the seed a reader registering now receives. The straddle check in `SharedTraceHandle::batches_through` stays. It is the detector for a reader that needs a boundary this forwarding merged away, and deleting it would leave a consumer silently double counting updates at times not before its cut. `stale_as_of_import_over_merged_chain_matches_direct` asserted that no batch straddles the `as_of`. That was a restatement of the old forwarding rather than a property a reader depends on, so it now asserts the straddle, which makes it the regression test for reintroducing a collective floor. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
Every hold command in this design reconstructs an ordering that routing destroyed: `CreateDataflow` goes only to the rendering runtime while `AllowCompaction` goes to the owning one, so the two are no longer on a single ordered stream. The alternative is to stop routing compaction and hand it to both runtimes, which restores that ordering directly. Model both variants so TLC decides between them rather than leaving the argument in a comment. `broadcast` hands the frontier to both runtimes and adds nothing else. It violates I1 in five states: the controller creates a dataflow and immediately drops it, as a cancelled peek does, so its own read hold is gone before the rendering runtime has applied either command. Compaction then advances on the owning runtime while the create is still queued, and the dataflow is later built at an `as_of` the collection has compacted past. Restoring the ordering within each stream does not restore it between them. `broadcast-standing` adds the standing per-collection hold: the rendering runtime holds every shared collection at the last frontier it has applied, so the owning runtime cannot realize a frontier that runtime has not reached. It holds over 19940 distinct states. The invariant is that a shared arrangement compacts only as fast as the slowest runtime's stream position, which is derived from the importer rather than from the controller's per-dataflow bookkeeping, and it needs no acquisition, no release and no reclaim. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
Records the conclusion the model reached and what follows from it. The diagnosis in read-holds.md stands: splitting compute keeps the controller's global floor and loses the single ordered command stream. The mechanism does not. Every version of it reconstructs that ordering on top of the routing that lost it, where sending compaction to both runtimes restores it directly. Adds broadcast-compaction.md with the design, the refutation of broadcast without a hold, the invariant it buys, the list of what it deletes, and the two unmeasured costs. Marks read-holds.md as superseded while keeping its diagnosis and its record of what each attempt cost, since that is what makes the simpler mechanism defensible rather than merely simpler. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
An index's `since` must not pass the `as_of` of a dataflow importing it. A single command stream ordered the create against every later compaction. Splitting compute across two runtimes routed those two commands to different streams, which lost that ordering. Broadcast `AllowCompaction` for maintenance-owned collections to the interactive runtime as well. It applies the frontier as a standing hold on the shared arrangement, and the publisher bounds its logical target by that hold, so a shared arrangement compacts only as fast as the slowest runtime's stream position. An importing dataflow whose `CreateDataflow` is still queued on the interactive runtime has registered no reader hold yet, so this is the only thing keeping the arrangement at or below the `as_of` it is about to read at. The hold is seeded at the publisher's compaction frontier at adoption rather than at the minimum time. The controller offers no `as_of` below a collection's own `since`, so no importer can need a frontier below it, and the seed keeps a collection the controller never compacts again from being pinned forever. Tests at three levels, each verified by sabotaging the mechanism it covers: the publication point, the registry and import path, and the multiplexer's routing. The publisher also carries a `debug_assert` that the published `since` never passes the hold. The acquisition layer (`AcquireHolds`/`ReleaseHolds`) is untouched and now redundant with this. Its deletion follows in a separate commit, so the invariant is asserted before anything is removed on the strength of it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
The standing hold makes every acquisition redundant. It bounds the publisher's compaction by the importing runtime's own stream position, so the publisher's agent never sits above an `as_of` that runtime can still present, which is the ratchet the acquisition existed to escape. Deleted: `ComputeCommand::AcquireHolds`/`ReleaseHolds` and `HoldRequest`, `compute_state/command_hold.rs`, `ComputeState::command_holds` and its maintenance pass, the registry's release records (`release_holder`/`reclaim_holder`/`released_holders`/`clear_released`), the multiplexer's `held_exports` and hold synthesis, and the `history.rs` arms. Two more things fell out, both of which existed only because a command hold was a second writer of the published `since`. `SharedTraceState::writer_since` and `refresh_since` go, because with one writer left the publisher assigns `since` directly. So do the four `Published` methods that recorded a hold the point did not own, and with them the distinction between a hold that is a request forwarded through the publisher's agent and a hold that is a grant backed by someone else's handle. Every hold is a request again. The replica no longer discards anything at a reconnection. A standing hold is per collection, carries no dataflow identity, and only rises, so clearing it would drop the bound to the minimum time until the replayed compactions raised it again. Four render tests covering the acquisition go. What they tested, a read at an `as_of` below where the publisher's agent already sits, is covered at the same level by `standing_hold_pins_until_the_importing_runtime_applies`. The multiplexer's ordering test keeps its other half, that compaction is forwarded uncapped, and its `Hello` test now asserts what `reset` still does. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
The spec described an acquisition protocol the code no longer has. It now models what the code does: compaction broadcast to both runtimes, and the rendering runtime's applied frontier bounding the publisher. Gone with the acquisition: the `acquire` and `release-on-maint` mechanisms and their variables, the `cap` mechanism and the frontier report it retired on, and the properties that had them as their subject. `Holds.cfg` is the shipped design rather than a retired one, and `HoldsRouted.cfg` replaces `HoldsCap.cfg` as the second refutation: compaction routed to the owning runtime alone, which is the raw defect the runtime split introduced. It fails in the same six states as broadcast-alone, so the pair shows that broadcasting the command restores the ordering within each stream and buys nothing for that counterexample. Applying a compaction and publishing one are now separate actions. The previous spec recomputed the published frontier only when a command arrived, which models a publisher that never notices its standing hold catching up. Splitting them is what makes the liveness question expressible, and `CompactionNotStalled` states it: the bound a lagging rendering runtime imposes is temporary. It holds only because `Fairness` says the runtimes drain, which is the coupling this design introduces. The reader's own registration is a bound now. It could not be one under the acquisition mechanism, since a registration forwarded through a single joining agent cannot represent a frontier below where that agent sits. Bounding the agent by the standing hold removes that ratchet, which is the same reason the acquisition became unnecessary. `Holds.cfg` holds over 7845 distinct states, depth 19. Both refutations violate I1. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
The broadcast reached the interactive runtime for every maintenance-owned collection, including the transient ones. Maintenance renders its own transient collections, its subscribes and copy-tos, and those are sinks with no arrangement for anything to import. The interactive runtime has installed nothing under such an id, so applying the frontier for one took the local path and the drop in that sequence panicked in `drop_collection` with "dropped untracked collection". Two fixes, each of which is sufficient on its own. Narrow the broadcast to the collections the interactive runtime can import, which are the non-transient ones maintenance publishes. That is the truthful scope: a standing hold exists to keep an importable arrangement readable. Decide local work from `collections` rather than from the id. The id was the wrong question in both directions: the interactive runtime holds empty local copies of the peer's introspection indexes, whose ids are the peer's to publish, and the peer renders transient collections this runtime has never seen. So a stray broadcast can no longer reach the drop path at all. Reproduced with `bin/sqllogictest --optimized -- --system-parameter-default=enable_two_runtime_compute=true` over subscribe_outputs.slt and cursor.slt, which panic before either fix and pass with either one. 836 assertions pass over the nine SUBSCRIBE-carrying files. Also drop a rustdoc link from a public module doc to a private field, which `ci/test/lint-doc.sh` rejects under `-D rustdoc::private-intra-doc-links`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
A refusal says the arrangement compacted past the dataflow's `as_of`, and the existing diagnostics cannot say whether anything on the replica could have prevented it. The standing hold is the value that decides: equal to the refusing `since` means the importing runtime had already applied that compaction before it built the dataflow, so the create was ordered behind it on that runtime's own stream and the arrangement was legitimately compacted, with no replica-side hold able to help. Below it means the publisher escaped its bound, which is a defect here. The publisher's `debug_assert` covers the second case but compiles out under `--optimized`, which is what sqllogictest and mzcompose run, so the panic path has to carry it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
`enable_two_runtime_compute` named a topology, two runtimes, rather than the thing it turns on. What a reader wants from the name is which mechanism appears, and the mechanism is an interactive runtime that serves reads off the arrangements the maintenance runtime builds. The new name also joins the established `enable_compute_*` family alongside `enable_compute_peek_response_stash` and `enable_compute_replica_expiration`, so it sorts and reads with its siblings. Renames the const and the parameter name together with all three Python registration sites, since `check-test-flags.sh` requires the mzcompose variable parameters and the parallel-workload flag list to agree with the compiled set. NOTE: the parameter name is also the LaunchDarkly key. The old key carries a staging rule enabling the feature for the read-placement evaluation environment, and nothing reads that key after this commit, so the new key has to be created before the next staging measurement or the environment silently falls back to the compiled default of off. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
Placement and preemption are independent: a walk off the timely worker can still have yield points, which buys somewhere to observe a cancellation request and a bound on how long one walk pins its batches. So S1 and S5 are not two values of one setting, and the fourth cell of that grid is an untried candidate rather than a contradiction. Records what each cell has measured and names the two that decide the question, both empty: S1 on M1 is predicted throughout and never measured, and nothing at all is known about an off-worker walk that yields. The deciding experiment runs S1 against S5 on E1, E11 and E8b, with the rule registered up front, and needs neither the interactive runtime nor this branch. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
The design of record, the evaluation record and the two machine-checked protocol models land separately in MaterializeInc#38239, along with their CI checks. They have no dependency on this code and reviewing 4,700 lines of prose against 7,000 lines of compute is worse for both. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
The offload walks a fast-path index peek's cursor on a blocking task instead of on the serving worker. It is orthogonal to the interactive runtime: it needs no second runtime, the interactive runtime needs no offload, and the two were only ever in one branch because they were written together. Keeping them together also argues two things at once, and one of them is now a question rather than a claim. Cooperative peek yielding is a second candidate for the same defect, placement and preemption turn out to be independent axes so there is a fourth untried combination, and the experiment that decides between them has not run. Landing a mechanism that measurement may retire is the expensive order, since a merged mechanism is harder to remove than an unmerged one. What this leaves is the interactive runtime alone, whose case is E7 and E9: temporary dataflows under maintenance load, and a replica that stays introspectable while it hydrates. E1 and E11 go with the offload, which is where E2 showed they belonged. Interactive peeks are unaffected. They resolve through `shared_index_peek_response`, and the offload was a conditional in front of that rather than the path itself. Removes `local_snapshot.rs`, `PendingPeek::IndexOffload` and its walk plumbing, the two dyncfgs, `mz_index_peek_walks_total`, `PeekStash::upload_blocking`, and the in-flight accounting. `static_assertions` goes with them, having been used only for the `Send` assertion on the owned snapshot. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
…nct work Resolutions for the rebase onto main after `ComputeRuntimeRole` and the error-multiplicity collapse landed separately. The role enum arrived on main with `Interactive` gated behind `cfg(test)`, since nothing outside tests could construct it yet. This branch constructs it, so the enum and its impl keep this branch's copy, which drops the gate and adds `publishes`. Main's copy is removed rather than merged: the two differ only in the gate and the accompanying notes about it. `distinct_errs` and `bundle_errs` both match exhaustively on `ArrangementFlavor`, and neither knew about `SharedTrace`. A shared trace is an imported arrangement like `Trace`, so it takes the same treatment in both: `distinct_errs` leaves it alone because this dataflow cannot rewrite an imported error trace in place, and `bundle_errs` yields its error collection so a delta join propagates those errors once rather than once per path.
0cfb53f to
3e03995
Compare
Introduce `mz_row_spine::ArcBatch`, a local newtype around `Arc<B>` carrying differential's batch traits (the orphan rule forbids the blanket impl on a bare `Arc<B>`), and switch the production spines and their builders (`RowRowSpine`, `RowValSpine`, `RowSpine`, `ValRowSpine`, `ColValSpine`, `ColKeySpine`) from `Rc`/`RcBuilder` to `ArcBatch`/`ArcBuilder`. An `Arc`-backed batch whose contents are `Send + Sync` can be read from a thread other than the one maintaining the trace, which `Rc` cannot do. Only the batch handle becomes atomic; the batch contents are unchanged. Also add generic `ArcOrdVal`/`ArcOrdKeySpine` aliases for callers outside `mz_compute`, adapt batch-size logging (`log_arrangement_size_inner`) to reach through the newtype to the inner `Arc`, and switch the storage sink trace to the `Arc`-backed spine. Two consumers of the sink trace follow from that switch. The iceberg sink stashes input batches while it waits for their batch description, so its `VecDeque` and the `with_ready_batches` helper now hold `ArcBatch` rather than `Rc`. And the arrangement-size operator's cache comment named the `RcBox` allocation its `Weak` keeps reserved, which is an `ArcInner` once the batch handle is atomic. The invariant it documents is unchanged. This is the foundational primitive the two-runtime read-isolation work (MaterializeInc#37770) builds on, extracted here for standalone review. It builds against released differential-dataflow with no fork or `[patch.crates-io]`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
Introduce `mz_row_spine::ArcBatch`, a local newtype around `Arc<B>` carrying differential's batch traits (the orphan rule forbids the blanket impl on a bare `Arc<B>`), and switch the production spines and their builders (`RowRowSpine`, `RowValSpine`, `RowSpine`, `ValRowSpine`, `ColValSpine`, `ColKeySpine`) from `Rc`/`RcBuilder` to `ArcBatch`/`ArcBuilder`. An `Arc`-backed batch whose contents are `Send + Sync` can be read from a thread other than the one maintaining the trace, which `Rc` cannot do. Only the batch handle becomes atomic; the batch contents are unchanged. Also add generic `ArcOrdVal`/`ArcOrdKeySpine` aliases for callers outside `mz_compute`, adapt batch-size logging (`log_arrangement_size_inner`) to reach through the newtype to the inner `Arc`, and switch the storage sink trace to the `Arc`-backed spine. Two consumers of the sink trace follow from that switch. The iceberg sink stashes input batches while it waits for their batch description, so its `VecDeque` and the `with_ready_batches` helper now hold `ArcBatch` rather than `Rc`. And the arrangement-size operator's cache comment named the `RcBox` allocation its `Weak` keeps reserved, which is an `ArcInner` once the batch handle is atomic. The invariant it documents is unchanged. This is the foundational primitive the two-runtime read-isolation work (#37770) builds on, extracted here for standalone review. It builds against released differential-dataflow with no fork or `[patch.crates-io]`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi
…nt sharing (MaterializeInc#38396) Replaces MaterializeInc#37881, whose head branch lives on a fork and so cannot be the base of a stacked PR in this repository. Same commits, same tree, on an upstream branch instead. This is the root of the stack MaterializeInc#38386 through MaterializeInc#38393, which splits MaterializeInc#37770. ### Motivation Cross-runtime arrangement sharing (the two-runtime read-isolation work, MaterializeInc#37770) needs batches readable from a thread other than the one maintaining the trace. Differential's default spines reference-count batches with `Rc`, which is worker-local. ### Description Introduce `mz_row_spine::ArcBatch`, a local newtype around `Arc<B>` that carries differential's batch traits (the orphan rule forbids the blanket impl on a bare `Arc<B>`), and switch the production spines and their builders — `RowRowSpine`, `RowValSpine`, `RowSpine`, `ValRowSpine`, `ColValSpine`, `ColKeySpine` — from `Rc`/`RcBuilder` to `ArcBatch`/`ArcBuilder`. An `Arc`-backed batch whose contents are `Send + Sync` can be read across threads, which `Rc` cannot do. Only the batch handle becomes atomic; the batch contents are unchanged, so the cost is a marginally more expensive refcount. Also adds generic `ArcOrdVal`/`ArcOrdKeySpine` aliases for callers outside `mz_compute`, adapts batch-size logging (`log_arrangement_size_inner`) to reach through the newtype to the inner `Arc`, and switches the storage sink trace to the `Arc`-backed spine. Builds against released differential-dataflow 0.25 with no fork or `[patch.crates-io]`. ### Verification `cargo check --workspace` passes with no `Cargo.lock` churn. `relations.slt`'s golden is rewritten because the spine type name appears in operator names. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
1 participant
A second in-process "interactive" compute runtime that serves reads against the arrangements the maintenance runtime maintains, isolating latency-sensitive reads from CPU-bound maintenance. Implements the two-runtime design (#37747).
The cross-runtime arrangement-sharing primitive lives entirely in Materialize (
mz-compute::shared_trace,mz-row-spine::ArcBatch), so it builds against released differential-dataflow with no fork or[patch.crates-io]. The diff also carries the production Rc→Arc spine migration that cross-thread sharing needs. The publisher follows the controller's compaction throughAllowCompactionrather than reading trace internals.How it works:
oks/errsarrangements into a per-processArrangementSharingRegistry. AComputeRuntimeRole { Solo, Maintenance, Interactive }distinguishes runtimes.Solo(default, single-runtime) takes the same code paths as before, with no registry, no second runtime, and norolemetric label. It is not a byte-identical deployment: the Rc→Arc spine migration is unconditional and shows in the goldens.PeekResponse-per-uuid contract is upheld below and above it, by each process's per-workerPartitionedComputeStateand by the controller's per-process one.ArrangementFlavor::SharedTracearrangements (not re-derived collections) and render joins/reduces over them; the query's own transient output is itself published so its result peek is served the same way.Maintenancerole, since those indexes bypass the normalexport_indexpublish path). The interactive runtime runs with logging disabled and serves introspection peeks from maintenance's published copies, so introspection queries during hydration return promptly instead of blocking behind maintenance. The logging dataflows sit on the same stalled workers, so "promptly" means promptly with stale data, which is the useful property during an incident.SyncActivator+ dirty-id inbox. A read whose dependency is not yet published or sealed is enqueued, not blocked. Publication (insert) and seal (note_frontier, fired from the maintenance export's frontier probe) mark the id dirty and wake the interactive worker, which re-examines only the affected pending work. A query dataflow whose imports are not yet published is still built immediately, binding a real but empty publication point that a publisher later adopts in place. Builds are never deferred or reordered, because every worker must construct dataflows in the same order. The per-stepprocess_peeksscan is removed on the interactive runtime.ENABLE_TWO_RUNTIME_COMPUTE(dyncfg, off by default in production) makes the controller launch replicas with a second interactive runtime (--interactive-compute-timely-config, its own worker ports). Enabled by default in the variable CI system parameters so the suite exercises the two-runtime path broadly. Flipping it changesServiceConfig::ports, so it rolls every compute replica in the environment.Acceptance: a
clusterd-test-driverworkflow drives an interactive query dataflow that imports an unpublished maintenance index, scheduled before the index publishes (mirroring controller ordering), and its result peek resolves correctly only after publication, proving the bind→fill→resolve read path is served off the maintenance worker.A
TwoRuntimeReadIsolationparallel-benchmark scenario measures read latency while the maintenance runtime is saturated by hydration churn. With two runtimes on, point-read p50 stays flat (~11ms) through the churn while the single-runtime baseline backlogs without bound. The scenario reads atstrict_serializable=False, which is the population that benefits: an interactive peek still waits for maintenance to seal its read timestamp, so stale and serializable reads get the full win and strict serializable gets close to none.Design: #37747 and
doc/developer/design/20260720_two_runtime_compute/design.md(the single design of record).Scope / follow-ups:
until/as_ofcoalescing (correct for one-shot peeks; a future subscribe migration must add it).IndexPeekMetrics, somz_active_peeks,mz_peek_durations_histogram, and theindex_peek_*histograms are empty while the feature is on.test/testdrive/introspection-sources.tdcarries the raised bound and points at the design doc entry.GlobalIdwhile the interactive runtime reconciles independently, so the slot binding wants either a written unreachability argument or an epoch.index_peek_total_seconds; pre-existingprintln!debug lines inrender.rsNonearms.🤖 Generated with Claude Code
https://claude.ai/code/session_019G29DBfgE8LXpE5jamm2Zi