adapter: use occ for read-then write, with incremental retries - #35192
Closed
aljoscha wants to merge 51 commits into
Closed
adapter: use occ for read-then write, with incremental retries#35192aljoscha wants to merge 51 commits into
aljoscha wants to merge 51 commits into
Conversation
Contributor
|
Thanks for opening this PR! Here are a few tips to help make the review process smooth for everyone. PR title guidelines
Pre-merge checklist
|
aljoscha
force-pushed
the
adapter-read-then-write-occ-incremental
branch
4 times, most recently
from
February 27, 2026 13:59
1675464 to
2cb8e6d
Compare
aljoscha
force-pushed
the
adapter-read-then-write-occ-incremental
branch
from
March 16, 2026 15:30
865091f to
22a54a2
Compare
aljoscha
force-pushed
the
adapter-read-then-write-occ-incremental
branch
11 times, most recently
from
April 23, 2026 10:33
179deb6 to
8b8d81c
Compare
aljoscha
force-pushed
the
adapter-read-then-write-occ-incremental
branch
10 times, most recently
from
April 28, 2026 14:39
aff3f2e to
f444d6d
Compare
aljoscha
force-pushed
the
adapter-read-then-write-occ-incremental
branch
2 times, most recently
from
May 4, 2026 10:48
9ad320f to
fc3f78c
Compare
The frontend read-then-write path had its own copy of the subscribe optimizer, `optimize_mir`, because its input arrives already lowered to MIR. The copy duplicated the local transform, the view import, the sink description, the expression prep, the global transform, the explain trace and the duration accounting, and it had dropped the comments that explain why the sink has no `FORCE NOT NULL` and no `REFRESH`. A prep or metainfo step added to one of the two would have silently shipped a differently optimized dataflow on the other. Fold both into `optimize_source`, which takes a `SubscribeSource` saying whether the subscribe reads an existing collection or a query that has already been lowered. `SUBSCRIBE FROM <SELECT>` lowers and then calls it, the read-then-write path calls `optimize_query` directly. The sink's column order becomes a parameter, empty for the raw-diff output the read-then-write subscribe wants. The `SubscribeFrom::Id` arm previously shared the pipeline tail with the query arm, so it goes through the same function now. Lowering is timed separately at the call site so it still counts towards the reported end-to-end optimization time.
`run_occ_loop` owned both the retry loop and the terminal write submission, which made it carry the six-arm `WriteResult` match twice, once for the write at an observed frontier and once for the blind write, along with two copies of the comment explaining why a raced write target is reported as a retryable conflict. The blind copy also needed a `soft_panic_or_log!` for an outcome that path cannot produce, and `defer_write` had to be threaded in as a sixteenth parameter just so the loop could decide between submitting and handing the diffs back. The loop now reports what it found: `OccOutcome::Blind` carries the diffs of a selection that reads no persisted state, which the subscribe running to completion proves are frontier-independent. The caller decides where they go, buffering them as session write ops inside a multi-statement transaction and otherwise submitting them through `submit_blind_write`. Both submission sites map the coordinator's answer through `classify_write_result`, which splits the one retryable conflict from the outcomes that end the statement, so the loop only has to know how to retry. The submission still happens while the OCC permit is held, so the permit keeps bounding in-flight statements from subscribe to durable write. `defer_write`'s remaining use, refusing a read-dependent write inside a transaction, moves next to where `defer_write` is computed. That check is defense in depth for the read-dependency check in `SessionClient::try_frontend_read_then_write`, and refusing before we run a dataflow is the last point where refusing is possible at all. `conn_id` and `target_global_id` stay parameters of the loop: the write at an observed frontier is still submitted from inside it.
`commit_timestamped` re-implemented `write_to_txns`: the catalog upper advance and its metric, the append and its metric, the `unwrap_or_terminate` on an unexpected storage error, the runaway-timestamp check and the oracle apply were all duplicated. The genuine difference is one policy decision, whether an upper conflict is retried at a fresh oracle timestamp or reported to the caller, because an OCC write's diffs are only valid at the timestamp they were computed for. Both now call `attempt_write_to_txns`, which does one attempt at a given `WriteTimestamp` and classifies the result as applied, upper conflict, or worker gone. The retry loop stays in `write_to_txns` and the conflict report stays in `commit_timestamped`, so each path keeps its own policy while the mechanics live in one place. `commit_timestamped` also returns `ControlFlow` now, the same "the committer survived" convention `commit` uses, instead of a `bool` that reads like success. `commit_timestamped` deliberately skips the wall-clock throttle, the merge loop, the group commit permit and write locks. All four are intentional and none of them was written down, so a reader comparing the two functions would have taken the omissions for oversights. Its doc comment now lists them with the reason for each. Move the write's target off the response channel. `InternalWriteResponder` carried `expected_target_global_id`, which is request data, and because the guard lived there while `writes` is a general map, commit staging had to hand-unpack a single-table shape and report "wrong number of tables" as `WriteResult::TargetChanged`. A client would read that as a retryable DDL race, which it is not. The request now carries a `WriteTarget`, staging reads the guard directly, and the unpacking and its soft panic are gone. Also record why the blind path needs a target re-check at all: it keys the append on the `CatalogItemId` and lets group commit resolve the generation, where the timestamped path names the `GlobalId` it validated. And state plainly that `InternalWriteResponder`'s `Drop` prevents a session panic, since the session task `expect`s a reply.
`FrontendWriteAttemptState` recorded the cancellation reason as an `AtomicU8` over three free `CANCELLATION_*` constants, with a `compare_exchange` to write it and a decode to read it back. The value has exactly one writer, the line after the `select!` in `try_frontend_read_then_write_with_cancel`, and the wrapper and the future it wraps are polled by the same task, so a `Mutex<Option<..>>` is all the `Send` bound needs. The struct doc now says that, so the next reader does not have to infer it from the atomics. The reason-to-error mapping existed twice, in `requested_error` and again at the `select!` site. It is now a `From` impl used by both. Also say which error must win where the retry loop checks for cancellation after a conflict. The check is redundant with the one at the top of the loop except when cancellation and an exhausted retry budget coincide, and there we want to report the cancellation.
The frontend read-then-write path registered its cancel watch under a fresh operation id and unregistered it from a drop guard, which forced an `Option<Uuid>` into every `connection_cancel_watches` entry, including the ones coordinator staged sequencing owns, where it means nothing and `sequence_staged` had to destructure a slot it never reads. Nothing can observe a watch left behind, so the unregister earns none of that. Every installation is a fresh channel: `RegisterConnectionCancelWatch` always replaces, and `sequence_staged` always inserts. `handle_execute` removes the entry when a statement starts, and `clear_connection` removes it when the connection's state is cleared, so the only reader of a previous entry, the "was already canceled" check in `sequence_staged`, only ever sees a watch installed during the statement it is running. Map growth was already bounded by live connections, since entries are keyed by connection id. So the command, the drop guard and the `Option<Uuid>` go, and `connection_cancel_watches` is a sender and receiver again. The frontend keeps its `borrow()` pre-check, which closes the window between the coordinator's insert and the session task's first read.
`take_over` and `take_over_sql_execute` were the same eight-argument body, differing only in whether they set `counted`. They are one function now, with a `TakeOver` argument that names the distinction: the statement the session task will run, versus a SQL `EXECUTE` it unrolls into an inner statement that gets counted wherever it ends up running. Rename `counted` to `coordinator_must_not_run`, which is what the flag is used for, so its doc no longer needs to spell out that taking over an `EXECUTE` is an exception. Also say why `retire` returns early on a slot with no logging id. It looks redundant with the emit path, which no-ops on a `None` id, but it is what keeps the end-reason mapping's soft panic from firing for a defused slot.
`build_success_response` said it builds the response after a successful write. It is called before the write, which is the whole point of the result-size bail inside it: that rejection happens before anything is written. The `write_ts: None` case of `Command::AttemptWrite` claimed the write "does not fail and will be retried until the write succeeds". It can come back read-only, target-changed, canceled or indeterminate, and every one of those is handled by the caller. What is true of that mode is only that its timestamp cannot be passed, since the oracle picks it. `view::Optimizer` had three constructors: `new`, `new_with_prep` and `new_with_prep_no_limit`, identical apart from a field. Keep `new_with_prep`, make `new` delegate to it, and turn the constant-folding limit into a `without_fold_constants_limit` modifier.
…atch `try_frontend_read_then_write_with_cancel` registered a connection cancel watch on the coordinator, a send plus oneshot await through the serialized command loop, before anything checked whether the statement was one this path handles. The OCC flag and the statement-kind match live in the callee, so every statement the frontend peek path bails on paid an extra synchronous coordinator round-trip: BEGIN, COMMIT, SET, all DDL, blind INSERT ... VALUES, FETCH, COPY FROM, DECLARE, and every SELECT when peek sequencing is off. With the OCC flag off, the production default, that is pure overhead, and because it lands on the serialized loop it also adds queueing latency for other sessions. Hoist the cheap part of the gate, the flag plus the statement kind, into the wrapper so it returns `Ok(None)` before the registration. The statement kinds are now named once, in `is_read_then_write_statement`, which both the gate and the callee use. The pre-check on the freshly registered watch stays where it is: it closes a real window between the coordinator's insert and the session task's first read, and only statements that actually proceed reach it.
`INSERT INTO t SELECT 1 WHERE false RETURNING x` takes no fast path on either side, since the constant-INSERT blind write requires no RETURNING. Both paths sequence it as a read-then-write, and they disagreed when no row matched. The coordinator builds its RETURNING rows from the diffs and `send_diffs` branches on those rows being non-empty, so it falls through to `Inserted(0)` and pgwire sends the tag with no result set. The frontend branched on the RETURNING expressions instead and produced an empty `SendingRowsImmediate`, so pgwire sent a row description and `SELECT 0`. A driver sees `fetchall()` return `[]` on one path and raise "the last operation didn't produce a result" on the other, and the HTTP and WebSocket endpoints return a `rows` payload against an `ok` tag. Postgres returns an empty result set here, so the frontend behavior was the better end state, but changing `send_diffs` changes the path that ships today and that decision does not belong in this change. So make the frontend bug-compatible instead, and pin it with a zero-row RETURNING case in the DML logging parity table, which so far only covered a RETURNING insert that always produces a row.
`ready_to_write` is decided on a progress message, and at that instant every accumulated diff has `ts < progress`. The loop then drains whatever else is queued with `try_recv` and feeds it to `process_message`, which appends data rows but only advances `current_upper` on a progress message. `write_ts` is read after the drain and `consolidate(write_ts)` rewrites every diff's timestamp to it, including a diff whose original timestamp is at or past `write_ts`. The window is real. A subscribe sends a batch's data and its following progress as two separate channel messages, and a batch's data is strictly below its upper, so the channel can hold `[data(<u1), progress(u1), data(in [u1,u2))]` and the drain can consume the third message and then see `Empty`. It is usually caught downstream, because a table write at `t` means the oracle advanced past `t` and `commit_timestamped` rejects the attempt as `TimestampPassed`, but a read dependency whose frontier advances without an oracle write is exactly the residual case, and a `REFRESH`-scheduled materialized view is such a dependency and is allowed by `validate_read_then_write_dependencies`. There the write would succeed at a timestamp earlier than the state it reflects. Track the maximum data timestamp next to where data rows are pushed, reset it in `consolidate`, and require it to be strictly below `write_ts` after the drain. That makes the invariant local instead of resting on a three-step argument about the oracle. When it does not hold we wait for the next progress message rather than writing, which re-establishes it.
The coordinator sequences a read-then-write's read as a real peek, so the optimizer's notices reach the session and, with `emit_timestamp_notice` set, so does `AdapterNotice::QueryTimestamp`. The frontend path emitted neither, which made `SET emit_timestamp_notice = on; DELETE FROM t WHERE ...` produce a notice with the flag off and silence with it on. The frontend peek path already does both, so this was an omission specific to read-then-write. The optimizer notices come from the `df_meta` the code discarded when unapplying the global LIR plan. The timestamp determination and the id bundle are already in hand at that point, which is what `ExplainTimestamp` needs, and it runs before the subscribe is created so the read holds still cover the collections it explains.
`add_active_compute_sink` made only the builtin table update conditional on
the sink's `internal` flag. The `mz_active_subscribes` gauge was bumped for
every subscribe, and decremented for every subscribe, so it stayed balanced
but each frontend DML transiently raised
`mz_active_subscribes{session_type="user"}`. A dashboard keyed on that gauge
then shows subscribes that no user subscribe stands behind, and disagrees
with `mz_subscriptions`, which correctly shows nothing.
Move the increment and the decrement inside the `internal` checks, so an
internal subscribe is invisible in both places, and say so at the creation
site.
The coordinator checks, in order: the transaction-state gate, planning, read-only mode, cluster resolution, cluster restrictions, RBAC, and then `allows_writes` and bounded staleness. The frontend checked bounded staleness and read-only before planning, so when two errors applied the winner differed. In a bounded-staleness transaction `DELETE FROM missing_table` reported the unknown item with the flag off and the bounded-staleness error with it on, and a privilege violation reported RBAC with the flag off and bounded staleness with it on. In read-only mode a planning error was reported with the flag off and `ReadOnly` with it on. Read-only now sits directly after planning, where `sequence_plan` has it, and bounded staleness sits after the cluster and RBAC block, where `sequence_insert` and `sequence_read_then_write` have it. Both still cover the constant-INSERT dispatch as well as the read-then-write path, which is why the read-only check was early in the first place. `allows_writes` comes with it, because `sequence_insert` rejects a transaction that cannot take a write before it rejects the isolation level. It is guarded on being inside a multi-statement transaction, the only state where it is defined and also the only one where it can be false, so autocommit statements still get their answer from `PeekClient::frontend_read_then_write`. That also aligns the constant-INSERT sub-path, which previously reported whatever `add_transaction_ops` raised for the transaction's existing ops (`SubscribeOnlyTransaction`, `DDLOnlyTransaction`) where the coordinator reports `ReadOnlyTransaction`. The parity table gains a bounded-staleness case. It pins the new placement: both paths now record the statement's cluster on the error row, where before the frontend rejected the statement without having resolved one. A read-only case is not expressible there, since read-only is a property of the server the whole table runs against.
…g out Two arms of `try_frontend_peek_inner`, for an unexpected EXPLAIN FILTER PUSHDOWN plan kind and an unexpected plan kind, logged a soft panic and then returned `Ok(None)`. Both sit below the point where the statement's log entry has been taken over, so in a release build the statement fell through to the coordinator, which began a second execution for it and counted it in `mz_query_total` twice. There is no legitimate coordinator fallback left for these arms, so report an internal error.
…tion Only `OccCounterUpdateAction` tolerated "read-then-write exceeded maximum retry attempts under contention". `UpdateAction`, `DeleteAction`, `InsertSelectAction` and `InsertReturningAction` all route through the OCC path now that the flag defaults on for mzcompose, so any of them can hit it and would fail the run. It takes extreme contention to reach, so today the statement timeout usually wins the race first, which is luck rather than design. `InsertReturningAction` had no error list at all: a constant INSERT is a blind write, but RETURNING takes it off that fast path and makes it a read-then-write.
The governing requirement for the frontend OCC path is that a user cannot tell which path sequenced their statement. Four differences are deliberate: the extra `optimization-finished` lifecycle event, the `max_result_size` accounting that counts distinct rows rather than diff entries, the write-timeline throttle a timestamped write skips, and the zero-row `INSERT ... RETURNING` response that is bug-compatible with the coordinator rather than with Postgres. Listing them keeps the next reader from filing them as bugs, and records the Postgres divergence as a thing we chose.
The permit-release test blocked its victim by sleeping inside the selection's dataflow. Cancelling the statement does not stop that operator, so the cluster's only worker stayed busy and the follow-up write waited on the worker rather than on the permit. Under a loaded 48-way test run that wait reached 26 seconds and the test failed for a reason it was not testing. The victim now parks in read linearization instead, waiting on a far-future REFRESH materialized view. That holds a permit while occupying no worker, so the follow-up write's own statement_timeout is the only thing standing between a leaked permit and a passing test, and its latency no longer depends on how loaded the machine is. The wall-clock assertion is gone with the mechanism that needed it, and the subscribe-teardown check moves to the cancellation test, which does run a real dataflow.
Comments and docs only, no behavior change. Removes the em dashes and the prose semicolons this branch introduced, and the chronology: comments that described what the code used to do, or what "the fix" changed, now state how the system behaves and why. The module docs of frontend_read_then_write no longer walk the steps of a function body. In their place they write down the load-bearing assumption that was missing: "does this write read persisted state" is answered twice, syntactically from depends_on() on the planned selection before anything runs, and dynamically from the subscribe channel ending on its own. Both answers have to agree, and what they decide is whether the diffs may be buffered until COMMIT. The call sites that used to explain that in full now point at the module docs. Deletions where a comment carried no weight: the duplicated account of the end-of-execution obligation in client.rs, which ExecutionLogging's type doc already gives, the sibling-function design argument in the doc of notice_if_startup_only, the call-site restatement on GroupCommitWriteLocks::insert_lock, and a few labels that restated the next line. The prose TODO about row cloning becomes a TODO, and so does the N.B. about write submission going through the coordinator. Two additions in appends.rs: the group commit path had lost the reason its partial-lock panic must panic, and the internal write arm never said that its lock acquisition is all-or-nothing, which is what keeps it deadlock-free. The QA findings doc was a narrative of a bug and its fix. It now records the findings themselves: why statement_timeout is enforced around the whole operation rather than the OCC loop, why permit starvation has a wider blast radius than a per-table write lock, why max_concurrent_occ_writes is constrained to at least 1, and the pre-existing coordinator-path hang that this change does not address. The design doc's Timeouts section described the statement timeout and the retry bound as work still to do.
The variable system parameter list is ordered by name, and the flag sat between two entries beginning with c.
ManySmallUpdates steps memory_clusterd up by about 56% alongside its wallclock regression, for the same reason: the operation installs a subscribe dataflow on the cluster where the fast-path peek it replaces holds nothing. The wallclock and throughput figures move a little between runs, so they are stated as the ranges observed rather than as single measurements.
Triggering a group commit after failing to acquire the write locks looks like the obvious improvement, and it would spin: the holder is not waiting on us, so the immediate retry finds the lock held, re-queues, and triggers again for as long as the holder keeps it. Riding a trigger someone else raises costs at most one timeline advancement tick and no CPU. Also records the statement_timeout = 0 corollary of permit starvation, and why permit acquisition sits before the read holds rather than after linearization, which would swap a common bounded cost for a rare unbounded one.
A statement sequenced from its session task reports the end of execution as a message to the coordinator, and carried the timestamp it took when it finished. The handler discarded it and read the clock itself, so finished_at and the ExecutionFinished lifecycle event charged the statement for however long that message waited in the coordinator queue. Since avoiding that queue is the point of frontend sequencing, the error grows with exactly the load the path exists to tolerate. end_statement_execution now takes the end timestamp from its caller. The coordinator path passes its own clock reading, which is what it was doing inside.
The helper waited for mz_query_total to reach the iteration number, but that counter is process-wide and the two setup INSERTs had already pushed it past the target, so the first probe always succeeded and the cancel fired against a statement that might still be connecting. It now reads a baseline and waits for a bump above it, then settles, because the bump lands a few steps before the permit is acquired. An unsynchronized cancel could also be lost outright, since a cancel that arrives before the frontend registers its watch is discarded. The victim then ran to the default 60s statement_timeout and failed with QUERY_CANCELED, which is the same SQLSTATE a cancel produces, so the assertion passed while the cancel path went untested. Both halves now check the error message, not just the code. The timeout half was unsynchronized too, and its 1s deadline could fire before a permit was ever taken. It waits for the permit like the cancel half and gets a 5s deadline so the settle cannot outlast it.
The module docs credited the pre-dataflow check with catching disagreement between the syntactic and dynamic predicates for whether a write reads persisted state. It cannot: it re-evaluates the syntactic predicate, so it catches a caller that skipped the client-side gate and nothing else. In the direction that matters, the syntactic predicate claiming the selection reads nothing while the subscribe then reads persisted state, the check passes and the write commits mid-transaction. The loop's Committed arm now asserts it has no write timestamp to apply inside a transaction, which is that disagreement made observable. It is a soft assertion because the write is durable by then, and it is conditioned on the timestamp rather than on deferral alone, since a committed write with no timestamp is legitimate for a selection whose diffs fully cancel. Also restores the contract that an empty portal is logged but not counted, which is the parity with the coordinator that count_statement exists for, and corrects a QA finding that called the coordinator path's far-future block unconditional. That path arms statement_timeout around its row stream, so only a zero timeout makes it permanent.
…er list" This reverts commit 47b3039. Moving an entry changes what CI_SYSTEM_PARAMETERS=random gives every parameter after it, because the walk draws one value per entry in list order from a seeded RNG. Previously reproduced seeds stop reproducing. The list is not strictly name-ordered to begin with, so the tidy-up bought nothing for that price.
The helper read its own baseline, and it runs after the victim's thread is spawned. When takeover won that race the baseline already counted the victim, so no later bump could satisfy the wait and the test failed with the counter stuck. The baseline is now taken at the call site, before the spawn, which is the only point where it is guaranteed not to include the statement being waited for.
aljoscha
force-pushed
the
adapter-read-then-write-occ-incremental
branch
from
July 28, 2026 13:48
194e1bc to
6a39f9c
Compare
An internal subscribe's dataflow is optimized against a catalog snapshot on the session task and shipped later, when the coordinator handles the message. A DROP of one of its dependencies can land in that window, and creation then fails with CollectionMissing. It went through ship_dataflow, whose contract is that creation cannot fail, so the coordinator panicked and took the process with it. Nightly reproduced this in the parallel-workload rename scenario. The coordinator's own read-then-write path has no such window, since it optimizes and ships in one turn of the loop, which is why treating creation as infallible is sound there and not here. This is the same window the frontend peek path already handles by reporting ConcurrentDependencyDrop, so we report that too. Shipping now precedes registering the sink, so a failure has nothing to unwind. That costs no concurrency: an internal subscribe writes no introspection row, so its registration notify is already resolved.
Contributor
Author
|
superseded by #37923 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
We now use SUBSCRIBE instead of PEEK to maintain the desired set of
updates that need to be written. We also don't acquire locks on tables
but instead optimistically try and write our updates at the timestamp
right at our current subscribe frontier.
Additionally, we take the opportunity this provides and move the
sequencing code from the coordinator main loop to the frontend, similar
to how we have done that for peeks in frontend_peek.rs.
Work towards https://github.com/MaterializeInc/database-issues/issues/6686
Implementation of https://github.com/MaterializeInc/materialize/blob/63645b72e83ee26d2cfa99d25d773a06e6accb74/doc/developer/design/20260210_incremental_occ_read_then_write.md