adapter: buffer a transaction's blind read-then-write until commit

A read-then-write inside a multi-statement transaction was refused
outright, because a write that commits immediately cannot be rolled back
at transaction end. That is the right answer only for a write that reads
persisted state. One whose selection reads nothing produces diffs that are
valid at any timestamp, so it can be buffered as a session write op and
land with the rest of the transaction, which is what a constant INSERT
already does.

Two predicates decide it and they have to agree. Before planning, the
syntactic one on `depends_on()` refuses a read-dependent write while
refusing is still possible. During execution, the subscribe answers the
same question dynamically, and the loop's `Committed` arm 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.

`max_concurrent_occ_writes` is sampled once at startup, so `ALTER SYSTEM
SET` on it silently did nothing. The statement is accepted and now warns
that the change takes effect when environmentd restarts. `RESET ALL` names
every parameter rather than one, so it compares values instead of names.
The parameter also gets a domain constraint of at least 1, since zero
permits would leave every read-then-write waiting out its
`statement_timeout`.

Author: aljoscha

doc/developer/design/20260210_incremental_occ_read_then_write.md

@@ -41,9 +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.
-- 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.
+- 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
 

src/adapter/src/client.rs

@@ -1909,27 +1909,21 @@ impl SessionClient {
             }
         };
 
-        // Only single-statement (`Started`) transactions may enter the OCC
-        // loop, its writes commit immediately and cannot be rolled back at
-        // transaction end. Multi-statement transactions reach this point only
-        // for AST-constant INSERTs whose planned expression turned out
-        // non-constant. Match the coordinator's error precedence: `mz_now()`
-        // gets its dedicated error, everything else is prohibited in a
-        // transaction block. The coordinator's lock-based path additionally
-        // supports INSERTs of volatile constants (for example `random()`) in
-        // transaction blocks by buffering the diffs until commit, which the
-        // OCC path cannot do.
+        // The syntactic predicate for "reads persisted state", see the module
+        // docs on `frontend_read_then_write`. Inside a transaction, only a write
+        // that reads nothing can run on this path.
+        //
+        // The AST gate above is not enough to establish this. It admits
+        // INSERTs whose source is constant in the AST, and such a statement can
+        // still plan to a selection with `Get` nodes, because SQL-implemented
+        // builtins (`pg_get_viewdef`, `text` to `reg*` casts, ...) read system
+        // relations. So decide on the planned selection, and do it before we
+        // execute a dataflow for a statement we would then refuse.
         {
             let session = self.session.as_ref().expect("SessionClient invariant");
-            if !matches!(session.transaction(), TransactionStatus::Started(_)) {
-                let contains_temporal = rtw_plan.selection.contains_temporal()
-                    || rtw_plan.assignments.values().any(|e| e.contains_temporal())
-                    || rtw_plan.returning.iter().any(|e| e.contains_temporal());
-                if contains_temporal {
-                    return Err(AdapterError::Unsupported(
-                        "calls to mz_now in write statements",
-                    ));
-                }
+            if session.transaction().is_in_multi_statement_transaction()
+                && !rtw_plan.selection.depends_on().is_empty()
+            {
                 return Err(prohibited_in_transaction(&stmt));
             }
         }

src/adapter/src/coord/sequencer/inner.rs

@@ -20,7 +20,9 @@ use futures::{Future, StreamExt, future};
 use itertools::Itertools;
 use mz_adapter_types::compaction::CompactionWindow;
 use mz_adapter_types::connection::ConnectionId;
-use mz_adapter_types::dyncfgs::{ENABLE_PASSWORD_AUTH, READ_THEN_WRITE_MAX_DEPENDENCIES};
+use mz_adapter_types::dyncfgs::{
+    ENABLE_PASSWORD_AUTH, FRONTEND_READ_THEN_WRITE, READ_THEN_WRITE_MAX_DEPENDENCIES,
+};
 use mz_catalog::memory::error::ErrorKind;
 use mz_catalog::memory::objects::{
     CatalogItem, Connection, DataSourceDesc, Sink, Source, Table, TableDataSource, Type,
@@ -75,7 +77,7 @@ use mz_sql::plan::{
 use mz_sql::session::metadata::SessionMetadata;
 use mz_sql::session::user::UserKind;
 use mz_sql::session::vars::{
-    self, IsolationLevel, NETWORK_POLICY, OwnedVarInput, SCHEMA_ALIAS,
+    self, IsolationLevel, MAX_CONCURRENT_OCC_WRITES, NETWORK_POLICY, OwnedVarInput, SCHEMA_ALIAS,
     TRANSACTION_ISOLATION_VAR_NAME, Var, VarError, VarInput,
 };
 use mz_sql::{plan, rbac};
@@ -4208,6 +4210,7 @@ impl Coordinator {
         };
         self.catalog_transact(Some(session), vec![op]).await?;
 
+        Self::notice_if_startup_only(session, &name);
         session.add_notice(AdapterNotice::VarDefaultUpdated {
             role: None,
             var_name: Some(name),
@@ -4224,6 +4227,7 @@ impl Coordinator {
         self.is_user_allowed_to_alter_system(session, Some(&name))?;
         let op = catalog::Op::ResetSystemConfiguration { name: name.clone() };
         self.catalog_transact(Some(session), vec![op]).await?;
+        Self::notice_if_startup_only(session, &name);
         session.add_notice(AdapterNotice::VarDefaultUpdated {
             role: None,
             var_name: Some(name),
@@ -4238,15 +4242,79 @@ impl Coordinator {
         _: plan::AlterSystemResetAllPlan,
     ) -> Result<ExecuteResponse, AdapterError> {
         self.is_user_allowed_to_alter_system(session, None)?;
+        // Which parameters `RESET ALL` changes has to be read before the
+        // transaction applies it, afterwards they all read as their default.
+        let startup_only_changed = self.startup_only_vars_changed_by_reset_all();
         let op = catalog::Op::ResetAllSystemConfiguration;
         self.catalog_transact(Some(session), vec![op]).await?;
+        for name in startup_only_changed {
+            session.add_notice(AdapterNotice::StartupOnlyVarUpdated {
+                var_name: name.to_string(),
+            });
+        }
         session.add_notice(AdapterNotice::VarDefaultUpdated {
             role: None,
             var_name: None,
         });
         Ok(ExecuteResponse::AlteredSystemConfiguration)
     }
 
+    /// System parameters whose value `environmentd` samples once at startup.
+    ///
+    /// `enable_adapter_frontend_occ_read_then_write` selects between the
+    /// lock-based and the OCC read-then-write path. Both are never live in one
+    /// process, so the choice is fixed at boot and every session inherits it.
+    /// `max_concurrent_occ_writes` sizes the OCC semaphore at boot.
+    ///
+    /// `ALTER SYSTEM` on one of these is allowed to go through. The catalog
+    /// value is what the next process start reads, and the running process
+    /// cannot observe it, so there is no window where two code paths are live at
+    /// once.
+    fn startup_only_vars() -> [&'static str; 2] {
+        [
+            FRONTEND_READ_THEN_WRITE.name(),
+            MAX_CONCURRENT_OCC_WRITES.name(),
+        ]
+    }
+
+    /// Warns that `name` is only read at startup, so the running process keeps
+    /// the value it sampled at boot.
+    fn notice_if_startup_only(session: &Session, name: &str) {
+        if Self::startup_only_vars()
+            .iter()
+            .any(|n| n.eq_ignore_ascii_case(name))
+        {
+            session.add_notice(AdapterNotice::StartupOnlyVarUpdated {
+                var_name: name.to_string(),
+            });
+        }
+    }
+
+    /// The startup-only parameters whose value `ALTER SYSTEM RESET ALL` would
+    /// change. Parameters already at their effective default are untouched, so
+    /// they are not reported.
+    fn startup_only_vars_changed_by_reset_all(&self) -> Vec<&'static str> {
+        // Value-based, unlike `notice_if_startup_only`, which warns whenever an
+        // operator names one of these parameters. `RESET ALL` names every
+        // parameter, so only a value that actually moves is worth a warning.
+        let config = self.catalog().system_config();
+        let defaults = config.defaults();
+        Self::startup_only_vars()
+            .into_iter()
+            .filter(|name| {
+                // Both names are registered system vars, a lookup failure
+                // would mean the definitions and this list have drifted apart.
+                let current = config
+                    .get(name)
+                    .expect("startup-only parameter is a registered system var")
+                    .value();
+                defaults
+                    .get(*name)
+                    .is_some_and(|default| default != &current)
+            })
+            .collect()
+    }
+
     // TODO(jkosh44) Move this into rbac.rs once RBAC is always on.
     fn is_user_allowed_to_alter_system(
         &self,

src/adapter/src/frontend_read_then_write.rs

@@ -67,12 +67,13 @@ use mz_expr::Eval;
 use mz_expr::row::RowCollection;
 use mz_expr::{CollectionPlan, Id, LocalId, MirRelationExpr, MirScalarExpr, RowSetFinishing};
 use mz_ore::cast::CastFrom;
-use mz_ore::soft_panic_or_log;
+use mz_ore::{soft_assert_or_log, soft_panic_or_log};
 use mz_repr::optimize::OverrideFrom;
 use mz_repr::{CatalogItemId, Diff, GlobalId, RelationDesc, Row, RowArena, Timestamp};
 use mz_sql::catalog::CatalogError;
 use mz_sql::plan::{self, MutationKind, QueryWhen};
 use mz_sql::session::metadata::SessionMetadata;
+use mz_storage_client::client::TableData;
 use prometheus::Histogram;
 use timely::progress::Antichain;
 use tokio::sync::mpsc;
@@ -87,7 +88,7 @@ use crate::coord::{Coordinator, TargetCluster};
 use crate::error::AdapterError;
 use crate::optimize::Optimize;
 use crate::optimize::dataflows::{ComputeInstanceSnapshot, EvalTime, ExprPrep, ExprPrepOneShot};
-use crate::session::{Session, TransactionOps};
+use crate::session::{Session, TransactionOps, WriteOp};
 use crate::statement_logging::{StatementLifecycleEvent, StatementLoggingId};
 use crate::{PeekClient, PeekResponseUnary, TimelineContext, optimize};
 
@@ -318,13 +319,17 @@ impl PeekClient {
         // write ops once we know them.
         session.add_transaction_ops(TransactionOps::Writes(vec![]))?;
 
-        // A write on this path commits immediately and cannot be rolled back at
-        // transaction end, so only a single-statement transaction may reach it.
-        // The check lives in `SessionClient::try_frontend_read_then_write`, and
-        // this is defense in depth for it: rejecting here, before we run a
-        // dataflow, is the last point where refusing is still possible.
-        if session.transaction().is_in_multi_statement_transaction() {
-            soft_panic_or_log!("read-then-write reached the OCC path inside a transaction");
+        // Inside a transaction only a write that reads nothing gets here, see
+        // the module docs. The syntactic check lives in
+        // `SessionClient::try_frontend_read_then_write`.
+        let defer_write = session.transaction().is_in_multi_statement_transaction();
+        if defer_write && !depends_on.is_empty() {
+            // Defense in depth for the check named above. Rejecting here, before
+            // we run a dataflow, is the only place left where refusing is still
+            // possible: past the OCC loop the write may already be durable.
+            soft_panic_or_log!(
+                "read-dependent read-then-write reached the OCC path inside a transaction"
+            );
             return Err(AdapterError::Internal(
                 "read-then-write cannot be run inside a transaction block".into(),
             ));
@@ -497,10 +502,36 @@ impl PeekClient {
         let response = match result {
             Ok(OccOutcome::Committed { response, write_ts }) => {
                 if let Some(write_ts) = write_ts {
+                    // A committed write timestamp inside a transaction means the
+                    // two predicates disagreed: the syntactic one let us defer,
+                    // the subscribe then read persisted state. The write is
+                    // already durable, so there is nothing to refuse, and
+                    // `apply_write` still has to run to keep the session's read
+                    // timestamps ahead of it.
+                    soft_assert_or_log!(
+                        !defer_write,
+                        "read-then-write committed a write inside a transaction"
+                    );
                     session.apply_write(write_ts);
                 }
                 Ok(response)
             }
+            Ok(OccOutcome::Blind { response, diffs }) if defer_write => {
+                // NOTE: A buffered session write carries no target-generation
+                // guard. The immediate path pins `target_global_id` and group
+                // commit re-validates it, but a `WriteOp` only names the
+                // `CatalogItemId` and commit staging resolves whatever global
+                // id is current then. So an `ALTER TABLE ... ADD COLUMN` that
+                // lands between here and COMMIT appends rows of the old arity
+                // under the new schema. This holds for every buffered write,
+                // not just ours.
+                session
+                    .add_transaction_ops(TransactionOps::Writes(vec![WriteOp {
+                        id: target_id,
+                        rows: TableData::Rows(diffs),
+                    }]))
+                    .map(|()| response)
+            }
             Ok(OccOutcome::Blind { response, diffs }) => {
                 match self
                     .submit_blind_write(

src/adapter/src/notice.rs

@@ -132,6 +132,12 @@ pub enum AdapterNotice {
         role: Option<String>,
         var_name: Option<String>,
     },
+    /// An `ALTER SYSTEM` statement named a system parameter that
+    /// `environmentd` only reads at startup, so the running process keeps the
+    /// value it sampled at boot.
+    StartupOnlyVarUpdated {
+        var_name: String,
+    },
     Welcome(String),
     PlanInsights(String),
     IntrospectionClusterUsage,
@@ -211,6 +217,7 @@ impl AdapterNotice {
             AdapterNotice::DroppedInUseIndex { .. } => Severity::Notice,
             AdapterNotice::PerReplicaLogRead { .. } => Severity::Notice,
             AdapterNotice::VarDefaultUpdated { .. } => Severity::Notice,
+            AdapterNotice::StartupOnlyVarUpdated { .. } => Severity::Warning,
             AdapterNotice::Welcome(_) => Severity::Notice,
             AdapterNotice::PlanInsights(_) => Severity::Notice,
             AdapterNotice::IntrospectionClusterUsage => Severity::Warning,
@@ -323,6 +330,7 @@ impl AdapterNotice {
             AdapterNotice::WebhookSourceCreated { .. } => SqlState::SUCCESSFUL_COMPLETION,
             AdapterNotice::PerReplicaLogRead { .. } => SqlState::SUCCESSFUL_COMPLETION,
             AdapterNotice::VarDefaultUpdated { .. } => SqlState::SUCCESSFUL_COMPLETION,
+            AdapterNotice::StartupOnlyVarUpdated { .. } => SqlState::WARNING,
             AdapterNotice::Welcome(_) => SqlState::SUCCESSFUL_COMPLETION,
             AdapterNotice::PlanInsights(_) => SqlState::from_code("MZ001"),
             AdapterNotice::IntrospectionClusterUsage => SqlState::WARNING,
@@ -508,6 +516,11 @@ impl fmt::Display for AdapterNotice {
                     "{vars} updated for {target}, this will have no effect on the current session"
                 )
             }
+            AdapterNotice::StartupOnlyVarUpdated { var_name } => write!(
+                f,
+                "changes to {} only take effect when environmentd restarts",
+                var_name.quoted()
+            ),
             AdapterNotice::Welcome(message) => message.fmt(f),
             AdapterNotice::PlanInsights(message) => message.fmt(f),
             AdapterNotice::IntrospectionClusterUsage => write!(