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 test for the parameter's domain constraint lands here too, though the
constraint itself arrives with the path that reads the parameter.

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(),
             ));
@@ -506,10 +511,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!(