adapter: sequence read-then-write from the session task with OCC

DELETE, UPDATE and INSERT ... SELECT run on the coordinator, which holds a
write lock on the target table across the read and the write. Every such
statement therefore serializes against every other one on that table, and
the coordinator loop is occupied for the duration.

This sequences them from the session task instead, using optimistic
concurrency control. The selection is read through an internal subscribe,
which streams the mutation's diffs directly rather than peeking every
matched row and recomputing them. The write is submitted at the timestamp
the diffs were observed at, and the group committer refuses it if another
writer got there first. A refusal is a retry with a fresh snapshot, not a
lost update, and the retry budget is `max_occ_retries`.

Concurrency is bounded by a semaphore of `max_concurrent_occ_writes`
permits, acquired before the read holds so that queued operations do not
pin compaction on their read dependencies while they wait.
`statement_timeout` is enforced in one place, the `select!` that owns the
whole operation, because every phase can block: permit acquisition,
linearization against a far-future `as_of`, and the retry loop itself.

The path is off by default and gated by
`enable_adapter_frontend_occ_read_then_write`, which is read once at
startup and fixed for the life of the process. A mixed-mode window would
be unsound: the lock-based path excludes concurrent writers, the OCC path
detects them afterwards, and the two do not synchronize.

Large mutations get faster because the subscribe streams diffs, small ones
get slower because each installs a dataflow where the old path used a
fast-path peek. That trade is deliberate and recorded in the design doc.

Author: aljoscha

doc/developer/design/20260210_incremental_occ_read_then_write.md

@@ -36,15 +36,18 @@ a subscribe that continually tracks the current state of the data.
 ## Non-Goals
 
 - High-performance writes under heavy contention. The current implementation
-  serializes writes behind a global lock; the new implementation serializes
-  them via OCC retries. Neither is designed for high write throughput.
+  serializes writes behind a global lock. The OCC implementation serializes
+  them via retries. Neither is designed for high write throughput.
 - Removing the in-process locks immediately. During rollout, the old lock-based
   path and the new OCC path coexist behind a feature flag. The locks can be
   removed once the OCC path is fully rolled out.
-- Multi-statement transactions. The OCC approach as described here applies to
-  single-statement implicit transactions. Explicit multi-statement write
-  transactions continue to use the existing path. And there are not plans to
-  support mixed read/write transactions.
+- Mixed read/write transactions. A write that reads persisted state commits at
+  the frontier it observed, which it cannot postpone until COMMIT, so it runs
+  only as a single statement. A write that reads nothing does compose with
+  transactions: its diffs are frontier-independent, so they are buffered as
+  session write ops and land when the transaction commits. That covers, for
+  example, `INSERT INTO t SELECT generate_series(1, 20000)`, whose values are
+  constant but too large to fold into a literal.
 
 ## Overview
 
@@ -128,7 +131,7 @@ Session Task                         Coordinator
   |                                      |
   |-- acquire OCC semaphore              |
   |                                      |
-  |-- CreateReadThenWriteSubscribe ----> |
+  |-- CreateInternalSubscribe ---------> |
   | <------------ subscribe channel -----|
   |                                      |
   |   +-- OCC Loop ------------------+   |
@@ -141,14 +144,14 @@ Session Task                         Coordinator
   |   |   if Success: break          |   |
   |   +------------------------------+   |
   |                                      |
-  |-- DropReadThenWriteSubscribe ------> |
+  |-- DropInternalSubscribe -----------> |
   |                                      |
 ```
 
 ### Timestamped writes
 
 A timestamped write is a write that must be committed at a specific timestamp.
-The group commit machinery has to be extended to supports this by:
+The group commit machinery has to be extended to support this by:
 
 1. Checking if the target timestamp is still valid (hasn't been passed by the
    oracle)
@@ -193,7 +196,7 @@ subscribe.
 The subscribes created for read-then-write are internal: they do not appear in
 `mz_subscriptions` or other introspection tables, and they don't increment the
 active subscribes metric. They are created and dropped via dedicated `Command`
-variants (`CreateReadThenWriteSubscribe`, `DropReadThenWriteSubscribe`).
+variants (`CreateInternalSubscribe`, `DropInternalSubscribe`).
 
 ## Correctness
 
@@ -242,7 +245,7 @@ oracle read timestamp. However, actually applying the write bumps the oracle
 read timestamp to at least the write timestamp, so at write time it holds that
 `write_ts <= oracle_read_ts`. The linearization invariant is maintained.
 
-### Single timestamped write write per group commit round
+### Single timestamped write per group commit round
 
 Only one timestamped write is processed per group commit round. This is correct
 because:
@@ -257,15 +260,22 @@ because:
 
 ### Timeouts
 
-We have to be careful about bounding the lifetime of the occ loop, both in
-wallclock time and number of retries. With the old approach, a read-then-write
-could take arbitrarily long, and block the rest of the system. With the new
-approach, the occ loop might try arbitrarily long, without ever succeeding. It
-will not block the rest of the system, though, which is a big benefit.
+The lifetime of the OCC loop has to be bounded, both in wallclock time and in
+number of retries. With the lock-based approach, a read-then-write could take
+arbitrarily long and block the rest of the system. With OCC it can retry
+arbitrarily long without ever succeeding, but it does not block the rest of the
+system, which is a big benefit.
 
-As a safety net, we should bound the lifetime of the occ loop with our existing
-statement timeout, and potentially add a hard upper limit on the number of
-attempts per occ loop.
+`statement_timeout` provides the wallclock bound. It is enforced in the session
+task, around the whole operation rather than around the loop alone, so it also
+covers planning, OCC permit acquisition, timestamp determination, and read
+linearization. Any of those can park indefinitely, and a parked operation holds
+an OCC permit, so a bound on the loop alone would leave the permit pool
+starvable.
+
+`max_occ_retries` provides the retry bound. A statement that keeps losing the
+race for its write timestamp fails with a contention error instead of retrying
+forever.
 
 ### Comparison with the old approach
 
@@ -284,6 +294,37 @@ The new approach is arguably easier to reason about: there is no global lock
 state to consider, no deferred operations, no lock merging. The correctness
 argument is local to the OCC loop and the group commit mechanism.
 
+## Deliberate differences from the lock-based path
+
+A user must not be able to tell which path sequenced their statement. These are
+the places where the two paths do differ, on purpose. They are listed here so
+that the next reader does not take them for bugs.
+
+- **Statement lifecycle events.** The frontend path records an
+  `optimization-finished` event for a DML, the coordinator path does not,
+  because it hands the read-then-write's inner peek a trivial logging context
+  and so logs nothing for it. We keep the extra event, it is real information
+  about a statement the user did run.
+- **`max_result_size` accounting.** The coordinator sums one row length per diff
+  entry before consolidation. The frontend recomputes the total from the
+  consolidated set, which counts one row length per distinct row and ignores
+  multiplicity. So a `DELETE` of a million copies of one row can exceed the
+  limit on the coordinator path and succeed on the frontend path. We keep the
+  frontend's accounting: it matches what the write actually appends, one entry
+  with a large diff.
+- **The write-timeline throttle.** A timestamped write does not go through the
+  throttle that a blind write's group commit applies, because its timestamp
+  comes from an observed subscribe frontier rather than from the clock. See the
+  doc comment on `GroupCommitter::commit_timestamped` for the full list of what
+  that path skips and why.
+- **Zero-row `INSERT ... RETURNING`.** Both paths report `INSERT 0 0` with no
+  result set when no rows match, because the coordinator decides the response
+  kind from the evaluated RETURNING rows and there are none. Postgres returns an
+  empty result set here, with a row description. The frontend path is
+  deliberately bug-compatible with the coordinator rather than correct on its
+  own: fixing it changes the behavior of the path that ships today, which is a
+  separate decision from this change.
+
 ## Performance
 
 The goal is not to make writes faster, but to not regress significantly.
@@ -295,17 +336,49 @@ Benchmarking a PoC-level implementation of the OCC approach against `main` for
 The benchmark varies concurrency (number of workers) on the x-axis and shows
 throughput (left) and latency (right). Key observations:
 
-- At low concurrency (1-7 workers), the OCC approach is comparable or _better_
-  than `main`. This is because the OCC path begins preparing the write (opening
-  the subscribe, receiving the snapshot) before the write timestamp is claimed,
-  whereas the old path only starts the peek after acquiring the lock.
+- At low concurrency (1-7 workers), the result depends on write size. A single
+  large `UPDATE`/`DELETE` is comparable or _better_ than `main`, because the
+  subscribe streams the mutation diffs directly whereas the old path peeks every
+  matched row and then recomputes the diffs. Small writes, however, _regress_:
+  every operation installs a subscribe dataflow, waits for its snapshot, and
+  tears it down, where the old path uses a cheap fast-path peek. This
+  per-operation subscribe overhead makes tiny `UPDATE`s roughly 1.5-2x slower at
+  low/no concurrency (observed in the nightly feature benchmark
+  `ManySmallUpdates` and the scalability `UpdateWorkload`).
 - At higher concurrency, performance degrades as expected due to the O(N^2)
   retry behavior: with more concurrent writers, more retries are needed. The
   concurrency semaphore (default 4 permits) bounds this in practice.
 - The benchmark is for a worst-case workload (all writers updating the same
   table). Real workloads with writes to different tables won't experience the
   contention.
 
+The chart above is from the PoC, which benchmarked `UPDATE t SET x = x + 1` over
+a larger table (the regime where OCC wins). It does not capture the small-write
+regression noted above, which is an accepted cost: high write throughput is a
+non-goal (see Non-Goals).
+
+Measured on the full implementation, with the OCC path on for every mzcompose
+suite, the small-write regression is at the bad end of that range. Across nightly
+runs the feature benchmark `ManySmallUpdates` is 1.7-1.9x slower and `Update`
+1.4x slower, and the scalability `UpdateWorkload` loses 36-39% throughput at
+concurrency 1 and about 22% at 8 and 32.
+
+`ManySmallUpdates` also steps `memory_clusterd` up by about 56%, from 56.8 MB to
+88.5 MB. Same cause as the wallclock step, from the other side: the subscribe
+dataflow each operation installs is arranged on the cluster, where the fast-path
+peek it replaces holds nothing. The absolute figures stay small because the
+dataflow lives only as long as the operation.
+
+The performance suites run the OCC path, because that is the configuration we
+intend to ship. The write benchmarks therefore record a one-time step, which we
+accept for the reasons above. Registering it is a follow-up once the change has
+landed and has a commit hash: `ManySmallUpdates` and `Update` go in
+`get_ancestor_overrides_for_performance_regressions` and `UpdateWorkload` in
+`ANCESTOR_OVERRIDES_FOR_SCALABILITY_REGRESSIONS`, both in
+`misc/python/materialize/version_ancestor_overrides.py`. That justification only
+applies when the comparison is against a released version, so until the step is
+inside the baseline these scenarios report a regression against `main`.
+
 ## Rollout
 
 The new path is controlled by a `enable_adapter_frontend_occ_read_then_write`

doc/developer/design/20260210_incremental_occ_read_then_write_qa_findings.md

@@ -0,0 +1,105 @@
+# QA findings: incremental OCC read-then-write
+
+Findings from an adversarial QA pass over frontend OCC sequencing for DELETE,
+UPDATE and INSERT ... SELECT, gated by
+`enable_adapter_frontend_occ_read_then_write`. The properties recorded here are
+pinned by the `qa_occ_`-prefixed tests in `src/environmentd/tests/server.rs`:
+
+```
+METADATA_BACKEND_URL=postgres://root@localhost:26257/materialize \
+  cargo nextest run -p mz-environmentd -E 'test(/qa_occ_/)'
+```
+
+## `statement_timeout` bounds the whole operation, not just the OCC loop
+
+Enforcement sits in the `tokio::select!` of
+`SessionClient::try_frontend_read_then_write_with_cancel`
+(`src/adapter/src/client.rs`). That frame owns the operation's entire lifetime
+and already handles cancellation, so a deadline placed there covers every phase:
+planning, OCC permit acquisition, timestamp determination, read linearization,
+and the retry loop. A `statement_timeout` of zero means "no deadline" and is
+represented by `futures::future::pending`.
+
+A deadline placed further in would leave phases unbounded, and the unbounded
+phases are the dangerous ones. `ensure_read_linearized` sleeps until the oracle
+reaches the read's `as_of`, so a read whose `as_of` lies far in the future, for
+example one depending on a `REFRESH AT '3000-01-01'` materialized view, parks
+there for years. Permit acquisition happens before the loop is entered at all,
+so a victim of such a parked operation would never reach an in-loop deadline
+either.
+
+When the deadline fires, the statement reports `AdapterError::StatementTimeout`
+and forwards `Command::PrivilegedCancelRequest` to the coordinator to clean up
+coordinator-owned work, mirroring the cancellation arm beside it. Dropping the
+`try_frontend_read_then_write` future releases the OCC permit, the read holds,
+and the `SubscribeHandle`, whose `Drop` sends `DropInternalSubscribe`.
+
+## Permit starvation has a wider blast radius than a per-table write lock
+
+A parked read-then-write holds its OCC semaphore permit
+(`max_concurrent_occ_writes`, default 4) for its whole lifetime. A handful of
+parked operations exhaust the pool and stall every read-then-write in the
+process, including ones on unrelated tables, because a waiter blocks on permit
+acquisition before doing anything else. The lock-based coordinator path cannot
+do that: it takes a write lock on the target table, so it only blocks writes to
+that table.
+
+That asymmetry is why the deadline above has to cover the permit wait. It bounds
+the victims of a starved pool, not just the operation that starves it.
+
+`statement_timeout = 0` removes that bound, so the starvation becomes permanent:
+four sessions with no deadline, each reading a far-future `REFRESH` materialized
+view, park in `ensure_read_linearized` holding the whole pool, and every
+read-then-write in the process fails or hangs until one of them is cancelled.
+
+Moving permit acquisition after linearization would fix that case, and we do not
+do it, because the permit is deliberately acquired before the read holds. A
+waiter that queued while holding read holds would pin compaction on its read
+dependencies for as long as it waits, which is what happens under ordinary write
+contention rather than only in the far-future case. So the ordering trades a
+rare unbounded case against a common bounded one. Sequencing it as read holds,
+then linearize, then permit would swap those, not remove the trade.
+
+## `max_concurrent_occ_writes` must be at least 1
+
+A value of 0 sizes the semaphore to zero permits, so every read-then-write in
+the process waits out its `statement_timeout` and then fails. The parameter
+carries a domain constraint requiring at least 1, which covers both ways of
+setting it, `ALTER SYSTEM SET` and `system_parameter_default`.
+
+`ALTER SYSTEM SET`/`RESET` of the parameter is accepted, because the value is
+sampled once at boot and the running process cannot observe a later change. The
+statement warns that the change only takes effect when `environmentd` restarts.
+
+## The coordinator path blocks on a far-future read until its own timeout
+
+`INSERT INTO dst SELECT a FROM mv`, where `mv` is a `REFRESH AT '3000-01-01'`
+materialized view, blocks on the lock-based coordinator path while holding the
+target table's write lock. It does not block forever. That path arms
+`statement_timeout` around the row stream it reads the selection from, in
+`sequencer::inner`, with zero mapped to `Duration::MAX`, so the statement fails
+after the deadline and only `statement_timeout = 0` makes the block permanent.
+
+The blast radius is the target table rather than the whole process, per the
+argument above, which is the one respect in which the coordinator path is better
+behaved here.
+
+This was first recorded as an unconditional hang. If you see one with a non-zero
+`statement_timeout`, the cause is not a missing deadline and the note above is
+where to stop looking. The check is cheap: with the OCC flag off,
+`SET statement_timeout = '5s'` and then the INSERT above should fail in about
+five seconds.
+
+## `RETURNING` is only parsed for `INSERT`
+
+The parser rejects `RETURNING` on DELETE and UPDATE, so the DELETE and UPDATE
+arms of the RETURNING handling in `build_success_response` are unreachable. They
+exist because that code dispatches on `MutationKind`, but no test can exercise
+them.
+
+## Reviewed without findings
+
+The OCC retry and consolidation logic, the interaction between timestamped
+writes and the oracle, and the reasoning that distinguishes an empty snapshot
+from the initial subscribe progress were reviewed against the code and produced
+no finding.

doc/user/data/metrics.yml

@@ -1417,6 +1417,20 @@ metrics:
   visibility: public
   tags:
   - environment
+- name: mz_occ_read_then_write_retry_count_bucket
+  help: Number of OCC retries per read-then-write operation.
+  labels:
+  - le
+  source: src/adapter/src/metrics.rs
+  visibility: internal
+- name: mz_occ_read_then_write_retry_count_count
+  help: Number of OCC retries per read-then-write operation.
+  source: src/adapter/src/metrics.rs
+  visibility: internal
+- name: mz_occ_read_then_write_retry_count_sum
+  help: Number of OCC retries per read-then-write operation.
+  source: src/adapter/src/metrics.rs
+  visibility: internal
 - name: mz_optimization_notices
   help: Number of optimization notices per notice type.
   labels:

misc/python/materialize/mzcompose/__init__.py

@@ -240,6 +240,11 @@ def get_variable_system_parameters(
             "true",
             ["true", "false"],
         ),
+        VariableSystemParameter(
+            "enable_adapter_frontend_occ_read_then_write",
+            "true" if version >= MzVersion.parse_mz("v26.36.0-dev") else "false",
+            ["true", "false"],
+        ),
         VariableSystemParameter(
             "enable_cast_elimination",
             "true",