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,14 @@ 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 on this path commits at the frontier it
+  observed, which it cannot postpone until COMMIT, so it runs only as a single
+  statement.
 
 ## Overview
 
@@ -128,7 +127,7 @@ Session Task                         Coordinator
   |                                      |
   |-- acquire OCC semaphore              |
   |                                      |
-  |-- CreateReadThenWriteSubscribe ----> |
+  |-- CreateInternalSubscribe ---------> |
   | <------------ subscribe channel -----|
   |                                      |
   |   +-- OCC Loop ------------------+   |
@@ -141,14 +140,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 +192,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 +241,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 +256,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 +290,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 +332,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`
@@ -318,6 +387,12 @@ write locks). We therefore must make the flag sticky per `environmentd` process
 lifetime (check on bootstrap only) to avoid this, and keep the current
 `confirm_leadership` checks.
 
+In CI the flag defaults to enabled for versions that carry it, so the mzcompose
+suites exercise the OCC path even though production keeps it off. The version
+gate leaves it disabled for the older versions an upgrade test runs, and
+`CI_SYSTEM_PARAMETERS=random` can pick either value, which is how both paths
+stay covered.
+
 Once the OCC path is fully rolled out and validated:
 
 1. Remove the old `sequence_read_then_write` code path

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",

misc/python/materialize/parallel_workload/action.py

@@ -62,6 +62,7 @@
     MAX_TABLES,
     MAX_VIEWS,
     MAX_WEBHOOK_SOURCES,
+    OCC_CONTENTION_EXHAUSTED_ERROR,
     Cluster,
     ClusterReplica,
     Column,
@@ -773,7 +774,6 @@ def run(self, exe: Executor) -> bool:
             )
         all_column_values = ", ".join(f"({v})" for v in column_values)
         query = f"INSERT INTO {table} ({column_names}) VALUES {all_column_values}"
-        # TODO: Use INSERT INTO {} SELECT {} (only works for tables)
         if self.rng.choice([True, False]):
             self.stmt_id += 1
             self.exe_prepared(query, f"insert{self.stmt_id}", exe)
@@ -796,6 +796,7 @@ def errors_to_ignore(self, exe: Executor) -> list[str]:
         result.extend(
             [
                 "canceling statement due to statement timeout",
+                OCC_CONTENTION_EXHAUSTED_ERROR,
                 # A random expression can evaluate to NULL (e.g. a map-key
                 # miss) even for a NOT NULL column, which is a legitimate
                 # rejection. The base list only ignores it for DDL complexity.
@@ -909,6 +910,11 @@ def run(self, exe: Executor) -> bool:
 
 
 class InsertReturningAction(Action):
+    def errors_to_ignore(self, exe: Executor) -> list[str]:
+        # A constant INSERT is a blind write, but RETURNING takes it off that
+        # fast path and makes it a read-then-write.
+        return [OCC_CONTENTION_EXHAUSTED_ERROR] + super().errors_to_ignore(exe)
+
     def run(self, exe: Executor) -> bool:
         table = None
         if exe.insert_table is not None:
@@ -946,7 +952,6 @@ def run(self, exe: Executor) -> bool:
             )
         all_column_values = ", ".join(f"({v})" for v in column_values)
         query = f"INSERT INTO {table} ({column_names}) VALUES {all_column_values}"
-        # TODO: Use INSERT INTO {} SELECT {} (only works for tables)
         returning_exprs = []
         if self.rng.random() < 0.5:
             returning_exprs += [
@@ -1012,6 +1017,7 @@ def errors_to_ignore(self, exe: Executor) -> list[str]:
         result.extend(
             [
                 "canceling statement due to statement timeout",
+                OCC_CONTENTION_EXHAUSTED_ERROR,
                 # A random SET expression can evaluate to NULL (e.g. a map-key
                 # miss) even for a NOT NULL column. That is a legitimate
                 # rejection, not a bug, and the column type can't be coerced
@@ -1070,6 +1076,9 @@ class ReadThenWriteCounterUpdateAction(Action):
     def errors_to_ignore(self, exe: Executor) -> list[str]:
         return [
             "canceling statement due to statement timeout",
+            # Extreme contention on one row is what this action creates, so
+            # exhausting the retry budget is an expected outcome here.
+            OCC_CONTENTION_EXHAUSTED_ERROR,
         ] + super().errors_to_ignore(exe)
 
     def run(self, exe: Executor) -> bool:
@@ -1096,6 +1105,7 @@ class DeleteAction(Action):
     def errors_to_ignore(self, exe: Executor) -> list[str]:
         errors = [
             "canceling statement due to statement timeout",
+            OCC_CONTENTION_EXHAUSTED_ERROR,
         ] + super().errors_to_ignore(exe)
         if exe.db.scenario == Scenario.Rename:
             errors += ["does not exist"]
@@ -2038,6 +2048,10 @@ def __init__(
         # behavior, you should add it. Feature flags which turn on/off
         # externally visible features should not be flipped.
         self.uninteresting_flags: list[str] = [
+            # Read once at environmentd startup, so an ALTER SYSTEM SET only
+            # takes effect after a restart. Flipping it here would be a no-op
+            # for the running process.
+            "enable_adapter_frontend_occ_read_then_write",
             "enable_compute_half_join2",
             "enable_mz_join_core",
             "enable_compute_correction_v2",

misc/python/materialize/parallel_workload/database.py

@@ -899,11 +899,24 @@ def __str__(self) -> str:
 # end-of-run check is guaranteed to find the table.
 READ_THEN_WRITE_COUNTER_NAME = "materialize.public.pw_rtw_counter"
 
+# The frontend read-then-write path gives up after its OCC retry budget when a
+# statement keeps losing the race for the write timestamp. That is a
+# user-visible consequence of contention, not a bug, so every action whose
+# statement is a read-then-write (DELETE, UPDATE, INSERT ... SELECT,
+# INSERT ... RETURNING) has to tolerate it. It is still counted in the error
+# statistics.
+OCC_CONTENTION_EXHAUSTED_ERROR = (
+    "read-then-write exceeded maximum retry attempts under contention"
+)
+
 # Error texts that prove an increment did not land.
 #
-# A concurrently modified dependency is what the coordinator reports when it
-# revalidates a plan before sequencing it, which is before any write. The other
-# is a cluster-resolution failure during planning, which the workload provokes
+# An exhausted retry budget is checked right after an attempt the group
+# committer rejected, so nothing was appended. A concurrently modified
+# dependency is what the coordinator reports when it revalidates a plan before
+# sequencing it, and what the frontend path reports for a changed write target,
+# both before any write. The last is a cluster-resolution failure during
+# planning, which the workload provokes
 # on purpose by pointing the default cluster at a nonexistent one, so the
 # statement never reaches a write path at all. The trailing quote keeps it from
 # matching "unknown cluster replica size" errors.
@@ -912,6 +925,7 @@ def __str__(self) -> str:
 # included, because either can race a commit that did happen. A wrong entry here
 # makes healthy runs fail, an unnecessary unknown only widens the upper bound.
 DEFINITELY_NOT_COMMITTED_ERRORS = (
+    OCC_CONTENTION_EXHAUSTED_ERROR,
     "was concurrently modified",
     "unknown cluster '",
 )

src/adapter-types/src/dyncfgs.rs

@@ -422,6 +422,13 @@ pub const DEFAULT_HYDRATION_BURST_LINGER: Config<Duration> = Config::new(
     "The burst-replica linger duration written when an AUTO SCALING STRATEGY omits LINGER DURATION.",
 );
 
+pub const FRONTEND_READ_THEN_WRITE: Config<bool> = Config::new(
+    "enable_adapter_frontend_occ_read_then_write",
+    false,
+    "Use frontend sequencing (with optimistic concurrency control) for \
+     DELETE, UPDATE, and INSERT operations.",
+);
+
 /// Adds the full set of all adapter `Config`s.
 pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
     configs
@@ -474,4 +481,5 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&CATALOG_INFO_METRICS_RECONCILE_INTERVAL)
         .add(&PG_TIMESTAMP_ORACLE_STATEMENT_TIMEOUT)
         .add(&ENABLE_SCOPED_SYSTEM_PARAMETERS)
+        .add(&FRONTEND_READ_THEN_WRITE)
 }

src/adapter/src/catalog/open.rs

@@ -317,8 +317,9 @@ impl Catalog {
         // `SystemConfiguration` values applied via `pre_item_updates` live in
         // the `SystemVars` value map by now. Mirror their effective values into
         // the dyncfg `ConfigSet`, so that startup-only reads observe configured
-        // values rather than compile-time defaults, `ENABLE_EXPRESSION_CACHE`
-        // just below being one of them. `apply_updates` only
+        // values rather than compile-time defaults. Those reads are the
+        // `ENABLE_EXPRESSION_CACHE` read just below and
+        // `FRONTEND_READ_THEN_WRITE` in coord bootstrap. `apply_updates` only
         // syncs the `ConfigSet` when a `SystemConfiguration` update is present,
         // and a deployment configured purely via `system_parameter_default` has
         // none, so sync explicitly here.