adapter: use occ for read-then write, with incremental retries

doc/developer/design/20260210_incremental_occ_read_then_write.md

@@ -41,10 +41,13 @@ a subscribe that continually tracks the current state of the data.
 - 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,7 +144,7 @@ Session Task                         Coordinator
   |   |   if Success: break          |   |
   |   +------------------------------+   |
   |                                      |
-  |-- DropReadThenWriteSubscribe ------> |
+  |-- DropInternalSubscribe -----------> |
   |                                      |
 ```
 
@@ -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
 
@@ -295,17 +298,27 @@ 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).
+
 ## 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,119 @@
+# QA findings: incremental OCC read-then-write
+
+Adversarial QA pass over the `adapter-read-then-write-occ-incremental` branch
+(frontend OCC sequencing for DELETE / UPDATE / INSERT...SELECT, gated by
+`enable_adapter_frontend_occ_read_then_write`).
+
+Tests live in `src/environmentd/tests/server.rs`, prefixed `qa_occ_` /
+`qa_legacy_`. Run them with the OCC binary already built:
+
+```
+METADATA_BACKEND_URL=postgres://root@localhost:26257/materialize \
+  cargo nextest run -p mz-environmentd -E 'test(/qa_occ_/)'
+```
+
+All `qa_occ_` tests — including the two former bug repros
+(`qa_occ_far_future_refresh_mv_respects_statement_timeout` and
+`qa_occ_far_future_rtw_starves_permit_pool`) — are now active and passing; they
+verify the fix described below. The only `#[ignore]`d test is
+`qa_legacy_far_future_refresh_mv_statement_timeout`, which documents the
+pre-existing far-future hang on the legacy coordinator path (out of scope for
+this OCC-only fix).
+
+## What was verified correct (passing tests)
+
+These exercise behaviors the design must preserve; all pass under OCC:
+
+- `qa_occ_concurrent_delete_no_overdelete` — N concurrent `DELETE`s of a
+  multiset row of multiplicity M sum to exactly M deleted, table ends empty,
+  multiplicity never goes negative. OCC retry prevents over-delete.
+- `qa_occ_duplicate_row_multiplicity_counts` — DELETE/UPDATE/INSERT...SELECT
+  affected-row counts respect multiset multiplicity.
+- `qa_occ_not_null_constraint_enforced` — `NOT NULL` violations via UPDATE and
+  INSERT...SELECT error out and leave the table unchanged (no partial commit).
+- `qa_occ_returning_values_correct` — INSERT RETURNING (constant and
+  read-dependent INSERT...SELECT) returns the right rows/expressions. (Note:
+  Materialize only parses `RETURNING` for INSERT, so the DELETE/UPDATE
+  RETURNING branches in `build_success_response` are effectively dead code.)
+- `qa_occ_update_moves_overlapping_rows` — row-moving UPDATEs
+  (`SET id = id + 1`) produce the correct final set through the
+  Let/Negate/map MIR transform and consolidation.
+- `qa_occ_insert_select_from_mv` — INSERT...SELECT reading a materialized view
+  (TimestampDependent timeline; linearization defaults to EpochMilliseconds).
+- `qa_occ_concurrent_mixed_dml_no_internal_error` — concurrent
+  UPDATE/DELETE/INSERT mix produces no internal errors / coordinator panics.
+
+The core OCC retry/consolidation logic, the timestamped-write/oracle
+interaction, and the snapshot/progress `NoRowsMatched` reasoning were also
+reviewed statically and found sound.
+
+## Fixed: OCC read-then-write now honors `statement_timeout` everywhere
+
+**Original bug.** `frontend_read_then_write::ensure_read_linearized` loops
+`tokio::time::sleep` until the oracle reaches `as_of`, with no
+`statement_timeout` check. When a read-then-write's `as_of` was far in the
+future — e.g. the read depends on a `REFRESH AT <far future>` materialized
+view — the session hung indefinitely; only client cancellation / disconnect
+freed it. `statement_timeout` was only enforced inside `run_occ_loop` (via
+`tokio::time::timeout` on `recv`), which is reached *after* linearization, so
+the pre-loop phases (planning, OCC permit acquisition, timestamp determination,
+linearization) were all unbounded. This contradicted the call-site comment,
+which claimed a pathological RTW "hits `statement_timeout` and returns".
+
+The OCC-specific amplification was the real concern: the parked op holds an OCC
+semaphore permit (`max_concurrent_occ_writes`, default 4) for its entire
+lifetime. A handful of such ops exhausted the global permit pool and wedged
+**all** read-then-writes process-wide — including ones on unrelated tables —
+because those victims block on permit acquisition *before* the OCC loop, so
+they too ignored `statement_timeout`. The legacy path only blocks writes to the
+*target* table (per-table write lock), so OCC widened the blast radius from one
+table to the whole RTW pipeline.
+
+**The fix.** `statement_timeout` is now enforced **centrally**, in the
+`tokio::select!` of `SessionClient::try_frontend_read_then_write_with_cancel`
+(`src/adapter/src/client.rs`) — the one place that already owns the whole
+operation's lifetime and handles cancellation. A new select arm sleeps for
+`*session.vars().statement_timeout()` (treating `Duration::ZERO` as "off /
+wait forever" via `futures::future::pending`, mirroring `run_occ_loop`'s
+`effective_timeout`). When it fires it returns `AdapterError::StatementTimeout`
+and forwards `Command::PrivilegedCancelRequest` to the coordinator (mirroring
+the existing `cancel_future` arm) to clean up any in-flight coordinator-owned
+work. Because the whole `try_frontend_read_then_write` future is dropped, the
+OCC permit, read holds, and `SubscribeHandle` (whose `Drop` sends
+`DropInternalSubscribe`) are all released.
+
+This single deadline now bounds *every* phase: planning, OCC permit
+acquisition, timestamp determination, `ensure_read_linearized`, **and** the OCC
+loop. The far-future op times out promptly and releases its permit, so victims
+behind a starved pool also honor their own `statement_timeout`. The in-loop
+`run_occ_loop` timeout is kept as defense-in-depth (it also bounds a single
+blocking `recv`); the call-site comment in `frontend_read_then_write.rs` was
+corrected to point at the central enforcement.
+
+Verified by `qa_occ_far_future_refresh_mv_respects_statement_timeout` (the
+far-future op returns a timeout error well within budget) and
+`qa_occ_far_future_rtw_starves_permit_pool` (the unrelated victim times out on
+permit acquisition within its own `statement_timeout` instead of hanging).
+Both are now active (no longer `#[ignore]`d) and passing.
+
+### Out of scope / unchanged
+
+- **Legacy path.** The legacy (lock-based) coordinator path also hangs on a
+  far-future read past `statement_timeout`
+  (`qa_legacy_far_future_refresh_mv_statement_timeout`, still `#[ignore]`d).
+  This pre-existing hang is intentionally left unchanged; the fix is OCC-only.
+
+### Corollary: `max_concurrent_occ_writes` is constrained to `>= 1`
+
+The same "block before the timeout is enforced" structure meant that
+`max_concurrent_occ_writes = 0` sizes the semaphore to zero permits and bricks
+every read-then-write after restart. With the central timeout such operations
+at least *time out* instead of hanging forever, but zero permits is still a
+footgun, so the parameter now carries a domain constraint requiring at least
+1. That covers both ways of setting it, `ALTER SYSTEM SET` and
+`system_parameter_default`.
+
+`ALTER SYSTEM SET`/`RESET` of the parameter is allowed, since the value is
+sampled once at boot and the running process cannot observe a later change.
+The statement emits a warning that the change only takes effect when
+`environmentd` restarts.

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",
+            ["true", "false"],
+        ),
         VariableSystemParameter(
             "enable_cast_elimination",
             "true",

misc/python/materialize/parallel_workload/action.py

@@ -1931,6 +1931,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",

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

@@ -313,6 +313,17 @@ impl Catalog {
             .await;
         builtin_table_updates.extend(builtin_table_update);
 
+        // Both the `system_parameter_defaults` set above and the durable
+        // `SystemConfiguration` values just applied via `pre_item_updates` now
+        // live in the `SystemVars` value map. Mirror their effective values into
+        // the dyncfg `ConfigSet` so startup-only reads — the
+        // `ENABLE_EXPRESSION_CACHE` read just below and `FRONTEND_READ_THEN_WRITE`
+        // in coord bootstrap — observe configured values, not compile-time
+        // defaults. `apply_updates` only syncs the `ConfigSet` when a
+        // `SystemConfiguration` update is present; deployments configured purely
+        // via `system_parameter_default` have none, so sync explicitly here.
+        state.system_config().dyncfg_updates();
+
         // Ensure mz_system has a password if configured to have one.
         // It's important we do this after the `pre_item_updates` so that
         // the mz_system role exists in the catalog.