adapter: let the oracle choose the read-then-write write timestamp

The OCC loop wrote at the frontier its subscribe reported, which conflates two
jobs. A frontier certifies what the loop has a complete view of. Choosing the
timestamp to write at is a separate decision, and the two coincide only because
the target table is usually what pins the frontier.

They come apart when another input does. A materialized view with a `REFRESH`
option settles until its next refresh, so a selection over one reports a
frontier hours or days ahead of the clock while the target table's upper is
still near it. Writing there ratchets the timeline's oracle into the future,
and the oracle is monotone and durable, so every later write and
strict-serializable read blocks until the clock catches up. Under
`serializable` the write is simply invisible.

So the oracle chooses and the frontier certifies. The target `T` is one step
above the oracle's write timestamp, the smallest value `commit_timestamped`
accepts, and a conflict hands back the next eligible one. The loop writes at
`T` once the frontier reaches `T`, with every diff from strictly below `T` as
the payload and everything at or after it held back for a later target.

Three boundaries carry the correctness. A progress message at `F` certifies
completeness below `F`, so readiness is `F >= T`. The payload is `t < T`
strictly, because a diff at `T` is concurrent with the write. And `T > as_of`,
because the snapshot arrives at `as_of` and has to be in the payload, which
the pre-read linearization guarantees by leaving the oracle at or above
`as_of`.

Readiness gives `T <= F`, not equality, so the payload can exclude diffs the
subscribe already delivered in `[T, F)`. For a selection that reads the target
table those say the table moved past `T`, the compare-and-append refuses, and
the loop retries at the timestamp the refusal names. Persist arbitrates, not
the frontier. This path produces that window itself: the committer appends at
the target and applies it to the oracle only afterwards, so while one write
sits in between, a second statement's target is a step behind the table's
upper. Retries converge because both refusals name a strictly higher timestamp
and the loop waits for the frontier to certify each one.

A zero-row answer no longer waits on a frontier either. It reports the timestamp
the emptiness holds as of, one below whichever of the frontier and the target
certifies less. For the common `UPDATE ... WHERE <no match>` that is the
statement's own `as_of`, already linearized before the subscribe started, so the
answer costs no group commit, and the two cases the loop used to separate turn
out to be the same rule.

Tests: the far-future refresh-MV write commits near the clock instead of being
refused, and a `serializable` session reads back its own such write, which
pins the anomaly this closes. Unit tests cover the fold's boundaries, where
the off-by-ones live. The zero-row linearization workflow now parks its winner
with a blind write rather than a timestamped one, so the answer is reached
without a round trip through the group committer and the wait for the oracle is
observable at all.

Author: aljoscha

doc/developer/design/20260210_incremental_occ_read_then_write.md

@@ -49,9 +49,9 @@ 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.
-- 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
+- Mixed read/write transactions. A write that reads persisted state commits at a
+  timestamp the oracle handed out while the statement ran, 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
@@ -65,12 +65,13 @@ subscribe-based OCC loop:
 1. Open a subscribe on the read expression (the `selection` from the
    `ReadThenWrite` plan), starting at the timestamp determined by the oracle
 2. Accumulate diffs from the subscribe
-3. When the subscribe frontier advances to T (meaning we have a consistent
-   snapshot), attempt to write the accumulated diffs at timestamp T
-4. If the write succeeds, done
-5. If the write fails because another writer already committed at timestamp T,
-   the subscribe will deliver the new state; go back to step 3 with the updated
-   diffs
+3. Take the write timestamp T from the timeline's oracle, one step above its
+   write timestamp, which is the smallest value the group committer accepts
+4. Once the subscribe frontier has advanced to T, so the accumulated diffs below
+   T are complete, attempt to write those diffs at T
+5. If the write succeeds, done
+6. If the write fails because another writer already took T, adopt the timestamp
+   the committer reports as next eligible and go back to step 4 with it
 
 This approach is correct by construction: the subscribe always reflects the
 committed state of the data, and the timestamped write mechanism ensures that
@@ -144,18 +145,37 @@ Session Task                         Coordinator
   |                                      |
   |   +-- OCC Loop ------------------+   |
   |   | receive diffs from subscribe |   |
-  |   | on frontier advance:         |   |
-  |   |   consolidate diffs          |   |
+  |   | target T from the oracle     |   |
+  |   | once frontier >= T:          |   |
+  |   |   consolidate diffs below T  |   |
   |   |   AttemptTimestampedWrite -> |-->|-- group_commit()
   |   |   <-- Success/Failed --------|<--|
-  |   |   if Failed: continue loop   |   |
+  |   |   if Failed: T = next, loop  |   |
   |   |   if Success: break          |   |
   |   +------------------------------+   |
   |                                      |
   |-- DropInternalSubscribe -----------> |
   |                                      |
 ```
 
+The target `T` comes from the oracle, not from the subscribe's frontier. A
+frontier certifies what the loop has a complete view of, which is a different
+question from which timestamp to write at, and the two coincide only when the
+selection's inputs are caught up with the oracle. An input can legitimately sit
+far in the future, a materialized view with a `REFRESH` schedule for instance, so
+taking `T` from the frontier would move the write timeline to that future and
+keep it there. The frontier gates the write, the oracle chooses it, and three
+invariants hold for every write the loop makes: the frontier is at or above `T`
+before it submits, the payload is every diff strictly below `T`, and `T` is above
+the statement's `as_of`, which is where the snapshot arrives.
+
+Note that the frontier being at or above `T` does not make the two equal. The
+frontier is a minimum over the selection's inputs, so it bounds neither `T` nor
+the target table's upper. Where it runs above `T`, a selection that reads the
+target table sees diffs the payload excludes, the compare-and-append refuses, and
+the loop retries at a higher target. Persist arbitrates the timestamp, not the
+frontier.
+
 ### Timestamped writes
 
 A timestamped write is a write that must be committed at a specific timestamp.
@@ -230,16 +250,20 @@ selection.
 
 ### The timestamped write ensures atomicity
 
-The write is submitted at the timestamp corresponding to the subscribe's
-frontier. The group commit machinery checks that this timestamp hasn't been
+The write is submitted at a timestamp taken from the timeline's oracle, once the
+subscribe's frontier has reached it so that the accumulated diffs below it are
+complete. The group commit machinery checks that this timestamp hasn't been
 passed by the oracle:
 
 - If the timestamp is still valid: the write is committed at exactly that
   timestamp, and the oracle is advanced past it. Any concurrent OCC loops that
   were targeting the same timestamp will fail and retry.
 - If the timestamp has already passed (another write committed first): the
-  write also fails. The OCC loop continues, the subscribe delivers the updates
-  from the intervening write, and the loop retries at the new frontier.
+  write also fails, and the reply names the next eligible timestamp. The OCC
+  loop adopts that as its new target, waits for the subscribe's frontier to
+  reach it, which folds the intervening writes' updates into the payload, and
+  retries. The reported timestamp is always strictly above the rejected one, so
+  the retries make progress.
 
 This ensures that the write is always based on the state of the data at exactly
 the write timestamp. There is no window for lost updates: either the write
@@ -256,6 +280,22 @@ 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.
 
+A statement that matches no rows performs no write, so nothing advances the
+oracle for it. It reports the timestamp its view is complete through and waits
+for the oracle read timestamp to reach that. A selection already empty at `as_of`
+is complete through `as_of`, which the statement linearized before its subscribe
+started, so the wait is satisfied at once. A selection that empties later reports
+the later timestamp and waits for a group commit.
+
+Reporting the frontier the subscribe observed would also be correct, but the
+subscribe follows Persist, which runs ahead of the oracle, so that frontier is
+usually a timestamp the oracle has not published and every zero-row answer would
+wait for a commit. The lower timestamp is not weaker: the answer describes the
+selection at the timestamp it reports, later reads land at or above it, and rows
+written after it are later in the serial order, as they would be for a SELECT
+there. Answering from state the oracle has not published is the hazard, and that
+is the direction the wait covers.
+
 ### Single timestamped write per group commit round
 
 Only one timestamped write is processed per group commit round. This is correct
@@ -324,13 +364,19 @@ that the next reader does not take them for bugs.
   When inputs are caught up the lock path's window is milliseconds wide and also
   needs a materially conflicting write plus a reader inside it, which is
   presumably why it went unnoticed.
-- **A lagging dependency blocks rather than waits.** This is the price of the
-  strengthening above. A selection dependency that persistently lags by more
-  than about one `default_timestamp_interval` makes every attempt conflict,
-  because the observed frontier is bounded by the lagging input while the write
-  timestamp keeps advancing with the oracle. The statement then burns retries
-  until `statement_timeout` instead of committing, where the lock path's peek
-  simply waited for the input to catch up.
+- **A lagging dependency delays rather than being read stale.** This is the
+  price of the strengthening above. The write timestamp comes from the oracle,
+  and the loop waits for the subscribe's frontier to certify it before
+  submitting, so a lagging selection dependency delays the statement by its lag.
+  A dependency that catches up commits normally. One that persistently lags by
+  more than about one `default_timestamp_interval` never lets an attempt land:
+  every wait ends with the oracle already past the target, the committer refuses
+  it and names a newer one, and the next wait is again bounded by the lagging
+  input. Each round costs one of `max_occ_retries`, but the rounds are paced by
+  frontier advances rather than spinning, so what ends the statement is
+  `statement_timeout`, which it runs out while holding an OCC permit and its
+  subscribe. The lock path's peek simply waited for the input to catch up and
+  then committed.
 - **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
@@ -357,11 +403,19 @@ that the next reader does not take them for bugs.
   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.
+- **The write-timeline throttle.** A blind write's group commit sleeps in the
+  committer until the wall clock catches up with the oracle's write timestamp,
+  keeping the timeline from running ahead of the clock. A timestamped write
+  cannot be throttled that way: its timestamp is fixed before it reaches the
+  committer, so sleeping would only delay a write that already has to land at
+  that timestamp. The committer instead refuses a target above
+  `write_ts_upper_bound(now)` outright. That refusal is unreachable in normal
+  operation, since the target is one step above the oracle's write timestamp and
+  the oracle clamps itself to the clock. It fires only for a write timeline that
+  has already run away from the clock, which is an environment-level invariant
+  violation rather than something a statement can provoke. 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
@@ -398,10 +452,10 @@ throughput (left) and latency (right). Key observations:
   a subscribe sees only progress from another table's write. They do still
   contend, in three ways: the concurrency semaphore is process-global across
   tables and clusters, the conflict predicate is the global oracle plus the
-  shared txns-shard upper, so two writers that observed the same frontier refuse
-  each other, and each timestamped write is its own committer round rather than
-  merging into a shared group commit. Every write benchmark is single-table, so
-  the cross-table case is unmeasured.
+  shared txns-shard upper, so two writers that took the same target timestamp
+  refuse each other, and each timestamped write is its own committer round rather
+  than merging into a shared group commit. Every write benchmark is single-table,
+  so the cross-table case is unmeasured.
 
 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
@@ -417,16 +471,24 @@ three runs) and `Update` 1.4x slower (33-45%), and the scalability
 `ManySmallUpdates` is the worst case for this design, and the reason is worth
 recording. Its statements set every matched row's `f1` to one shared random
 value, which merges a whole residue class, so the class count only shrinks and
-roughly 90% of its 100 updates end up matching no rows. A statement that matches
-nothing still has to linearize its read, and the oracle advances only when a
-group commit applies, so each of those statements needs a commit that has nothing
-to write. We ask for one rather than waiting for the periodic keepalive, which
-costs a commit round trip per statement instead of up to a full
-`default_timestamp_interval`. That is the difference between 3.5x and 157x, but
-it is not free, and a workload dominated by zero-row writes pays it on every
-statement. The residue is the price of the linearization guarantee rather than a
-defect: correctness requires the oracle to advance, and only a commit advances
-it.
+roughly 90% of its 100 updates end up matching no rows. Such a statement has
+nothing to write, and the oracle advances only when a group commit applies, so
+linearizing a timestamp the oracle has not reached costs a commit that carries
+nothing. Asking for one rather than waiting for the periodic keepalive is the
+difference between 3.5x and 157x. A selection already empty at its `as_of` needs
+no commit at all, which is this workload's shape, and what remains is a selection
+that empties later. That residue is the price of the linearization guarantee
+rather than a defect, since only a commit advances the oracle.
+
+Note: the figures above were measured before the write timestamp came from the
+oracle and a zero-row answer reported the timestamp its view is complete through.
+Re-measured on `ManySmallUpdates` with the path forced on for both builds, those
+two are worth 1.6x to 2.3x, 0.59-0.88s against 1.33-1.41s over four comparisons,
+with a control holding the path off on both sides at 0.336s against 0.337s. Group
+commits per statement fall from about 1.09 to between 0.13 and 0.37, which is the
+mechanism rather than a proxy for it. The scenario remains about 2x the lock path,
+the rest of the gap being the subscribe dataflow installed and torn down for every
+statement.
 
 The PoC's large-write win does not survive here. `Update` is itself a large
 mutation, a full-table update over 10^6 rows, and it is 1.4x slower. An `UPDATE`

src/adapter/src/coord/appends.rs

@@ -445,11 +445,11 @@ impl GroupCommitter {
     ///
     /// What [`Self::commit`] does that this skips, and why that is safe:
     ///
-    /// * The wall-clock throttle. `target_timestamp` is the caller's to choose, so
-    ///   instead of sleeping until the clock catches up we refuse a target above
-    ///   [`write_ts_upper_bound`] outright. Sleeping is the wrong answer for a caller
-    ///   whose target can be hours out, and committing there would advance the oracle
-    ///   with it.
+    /// * The wall-clock throttle. `target_timestamp` is the caller's to choose, and a
+    ///   target above [`write_ts_upper_bound`] is refused rather than slept off.
+    ///   Committing there would advance the oracle with it, and a caller that took its
+    ///   target from the oracle cannot exceed the bound unless the timeline has already
+    ///   run away, which sleeping would not resolve.
     /// * A [`GroupCommitPermit`]. The caller bounds how many of these are in
     ///   flight, and that is the backpressure for this path.
     /// * Merging queued commits. There is nothing to merge into: these diffs

src/adapter/src/error.rs

@@ -144,6 +144,10 @@ pub enum AdapterError {
     /// The write timestamp ran past what the write timeline may be advanced to,
     /// so nothing was appended. See `coord::timeline::write_ts_upper_bound` for
     /// the bound and why exceeding it is not recoverable.
+    ///
+    /// The timestamp comes from the timeline's oracle, so reaching this means the
+    /// oracle has run away from the wall clock rather than that the statement
+    /// asked for anything unusual.
     ReadThenWriteTimestampTooFarAhead {
         target_timestamp: mz_repr::Timestamp,
         limit: mz_repr::Timestamp,
@@ -929,7 +933,10 @@ impl AdapterError {
             }
             AdapterError::ReadThenWriteContention => SqlState::T_R_SERIALIZATION_FAILURE,
             AdapterError::ReadThenWriteTimestampTooFarAhead { .. } => {
-                SqlState::FEATURE_NOT_SUPPORTED
+                // An invariant violation in the environment rather than a property of
+                // the statement: the write timeline has run away from the wall clock.
+                // Nothing the client sends can produce or avoid it.
+                SqlState::INTERNAL_ERROR
             }
             AdapterError::CollectionUnreadable { .. } => SqlState::NO_DATA_FOUND,
             AdapterError::NoClusterReplicasAvailable { .. } => SqlState::FEATURE_NOT_SUPPORTED,