storage/upsert: select the stash representation by dyncfg

enable_upsert_chunked_stash (off in production, on and randomized in CI)
selects between two instantiations of the upsert-v2 operator loop:

* Chunked (flag on): ChunkBatcher stash, chunk-spine feedback
  arrangement, bulk-probe drain. Spills committed chunk bodies through
  the process buffer pool.
* Paged (flag off, the default): paged columnar merge batcher stash,
  ValRowSpine feedback arrangement, cursor drain. Spills cold chains
  through the storage-owned column pager.

The operator loop itself is shared: build_upsert_operator is generic
over an UpsertStashArm, whose two impls carry the flavor-specific
pieces (stash batcher, feedback spine, flush, drain). UpsertStashFlavor
names the arms and resolves from the config set once, at operator
construction, mirroring compute's ArrangementBatcher. Both flavors'
spill paths stay gated by enable_upsert_paged_spill, so storage_state
applies that flag to both the chunk gate and the restored
upsert_stash_pager.

The unit-test harness runs every scenario under both flavors and
asserts they produce identical output.

Author: DAlperin

misc/python/materialize/mzcompose/__init__.py

@@ -308,6 +308,14 @@ def get_variable_system_parameters(
             "true",
             ["true", "false"],
         ),
+        # On by default so CI exercises the chunked stash flavor, which is
+        # off in production while it earns trust. Only meaningful when
+        # enable_upsert_v2 is true.
+        VariableSystemParameter(
+            "enable_upsert_chunked_stash",
+            "true",
+            ["true", "false"],
+        ),
         VariableSystemParameter(
             "enable_upsert_v2",
             "false",

misc/python/materialize/parallel_workload/action.py

@@ -3055,6 +3055,7 @@ def __init__(
             "0.02",
         ]
         self.flags_with_values["enable_upsert_paged_spill"] = BOOLEAN_FLAG_VALUES
+        self.flags_with_values["enable_upsert_chunked_stash"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["column_chunk_compress_min_depth"] = [
             "0",  # compress every spilled body
             "1",  # the default: fresh chunks store uncompressed

src/storage-types/src/dyncfgs.rs

@@ -392,21 +392,52 @@ pub const STORAGE_UPSERT_MAX_SNAPSHOT_BATCH_BUFFERING: Config<Option<usize>> = C
     ParameterScope::Replica,
 );
 
-/// Storage's leg of the process-wide chunk spill gate
-/// (`mz_timely_util::columnar::chunk`). The gate is the OR of a compute leg
-/// (`enable_column_paged_batcher_spill`) and this storage leg: chunks spill
-/// while either is set, so this flag cannot veto spilling that the compute
-/// flag has enabled. Spilled chunks draw on the one shared pool budget.
+/// Allow the upsert-v2 stash to spill out of RSS. Off by default; while off,
+/// the stash keeps everything resident.
 ///
-/// Off by default. Enabling it also installs the process buffer pool (via
-/// compute's config handler, which reads this flag from the aggregate dyncfg
-/// set), so storage-only spilling needs no compute-side gate.
+/// The spill mechanism depends on the stash flavor
+/// ([`ENABLE_UPSERT_CHUNKED_STASH`]):
+///
+/// * Chunked: sets storage's leg of the process-wide chunk spill gate
+///   (`mz_timely_util::columnar::chunk`). The gate is the OR of a compute
+///   leg (`enable_column_paged_batcher_spill`) and this storage leg: chunks
+///   spill while either is set, so this flag cannot veto spilling that the
+///   compute flag has enabled. Spilled chunks draw on the one shared pool
+///   budget, and the gate is consulted at every chunk commit, so flips
+///   apply to running dataflows.
+/// * Paged: gates the storage-owned column pager the stash and feedback
+///   arrangement route their chains through, independently of compute's
+///   `enable_column_paged_batcher_spill`. Captured at operator
+///   construction, so flips apply to dataflows created after the change.
+///
+/// Enabling it also installs the process buffer pool (via compute's config
+/// handler, which reads this flag from the aggregate dyncfg set), so
+/// storage-only spilling needs no compute-side gate.
 pub const ENABLE_UPSERT_PAGED_SPILL: Config<bool> = Config::new(
     "enable_upsert_paged_spill",
     false,
-    "Allow upsert-v2 chunks to spill to the shared buffer pool. Sets the storage leg of the \
-     process-wide spill gate, which is the OR of this flag and the compute \
-     `enable_column_paged_batcher_spill`.",
+    "Allow the upsert-v2 stash to spill out of RSS, through the buffer pool (chunked stash \
+     flavor) or the column pager (paged stash flavor).",
+    ParameterScope::Replica,
+);
+
+/// Use the chunked stash flavor for the upsert-v2 operator: differential's
+/// chunk merge batcher for the source stash and a spine of chunk batches for
+/// the feedback arrangement, with a bulk-probe drain. When `false` (the
+/// default), the paged flavor is used: the paged columnar merge batcher and a
+/// `ValRowSpine`, with a cursor-based drain. See
+/// `mz_storage::upsert_continual_feedback_v2::UpsertStashFlavor` for the
+/// comparison.
+///
+/// Read at operator construction time; flips take effect on dataflows created
+/// after the change. Only meaningful when [`ENABLE_UPSERT_V2`] is `true`.
+/// Spilling in either flavor is gated by [`ENABLE_UPSERT_PAGED_SPILL`].
+pub const ENABLE_UPSERT_CHUNKED_STASH: Config<bool> = Config::new(
+    "enable_upsert_chunked_stash",
+    false,
+    "Use the chunk batcher and chunk spine for the upsert-v2 stash and feedback arrangement, \
+     instead of the paged columnar merge batcher and ValRowSpine. Only meaningful when \
+     enable_upsert_v2 is true.",
     ParameterScope::Replica,
 );
 
@@ -533,6 +564,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&ENABLE_UPSERT_V2)
         .add(&SUSPENDABLE_SOURCES)
         .add(&ENABLE_UPSERT_PAGED_SPILL)
+        .add(&ENABLE_UPSERT_CHUNKED_STASH)
         .add(&WALLCLOCK_GLOBAL_LAG_HISTOGRAM_RETENTION_INTERVAL)
         .add(&WALLCLOCK_LAG_HISTORY_RETENTION_INTERVAL)
         .add(&crate::sources::sql_server::CDC_CLEANUP_CHANGE_TABLE)

src/storage/src/render/sources.rs

@@ -339,6 +339,13 @@ where
                         if dyncfgs::ENABLE_UPSERT_V2
                             .get(storage_state.storage_configuration.config_set())
                         {
+                            // Resolved here, at operator construction, so the
+                            // dataflow keeps one stash flavor for its whole
+                            // life even if the flag flips underneath it.
+                            let stash_flavor =
+                                crate::upsert_continual_feedback_v2::UpsertStashFlavor::from_config(
+                                    storage_state.storage_configuration.config_set(),
+                                );
                             crate::upsert::upsert_v2(
                                 upsert_input.enter(scope),
                                 upsert_envelope.clone(),
@@ -347,6 +354,7 @@ where
                                 previous_token,
                                 export_config,
                                 backpressure_metrics,
+                                stash_flavor,
                             )
                         } else {
                             crate::upsert::upsert(

src/storage/src/storage_state.rs

@@ -846,12 +846,14 @@ impl<'w> Worker<'w> {
                     STORAGE_SERVER_MAINTENANCE_INTERVAL
                         .get(self.storage_state.storage_configuration.config_set());
 
-                // Set storage's leg of the process-wide chunk spill gate.
-                // The buffer pool and its budget are the shared ones
-                // configured by compute's `apply_worker_config` (compute and
-                // storage run in the same process). The gate ORs this leg
-                // with compute's, so chunks spill while either subsystem's
-                // flag is set.
+                // Apply storage's upsert spill flag to both stash flavors'
+                // mechanisms: the storage leg of the process-wide chunk
+                // spill gate (chunked flavor) and the storage-owned column
+                // pager (paged flavor). The buffer pool, the pager pool, and
+                // their budgets are the shared ones configured by compute's
+                // `apply_worker_config` (compute and storage run in the same
+                // process). The chunk gate ORs storage's leg with compute's,
+                // so chunks spill while either subsystem's flag is set.
                 //
                 // The flag is replica-scoped: the storage controller merges
                 // per-replica overrides into the `UpdateConfiguration`
@@ -867,6 +869,7 @@ impl<'w> Worker<'w> {
                         enabled, "upsert stash spill: applying gate",
                     );
                     crate::upsert::upsert_stash_spill::set_enabled(enabled);
+                    crate::upsert::upsert_stash_pager::set_enabled(enabled);
                 }
             }
             InternalStorageCommand::StatisticsUpdate { sources, sinks } => self

src/storage/src/upsert.rs

@@ -296,9 +296,10 @@ macro_rules! upsert_source_time_unit {
 }
 upsert_source_time_unit!(GtidPartition, Lsn);
 
-/// Storage's leg of the process-wide chunk spill gate.
+/// Storage's leg of the process-wide chunk spill gate, used by the chunked
+/// upsert-v2 stash flavor.
 ///
-/// The upsert-v2 source stash and feedback arrangement spill through the
+/// In that flavor the source stash and feedback arrangement spill through the
 /// process buffer pool ([`mz_timely_util::columnar::chunk`]): committed chunk
 /// bodies land in the pool once compute's config handler has installed and
 /// budgeted it (storage and compute run in the same `clusterd` process).
@@ -315,6 +316,43 @@ pub mod upsert_stash_spill {
     }
 }
 
+/// Pager for the paged upsert-v2 stash flavor.
+///
+/// This draws from the same process-wide [`TieredPolicy`] budget pool as the
+/// compute column-paged batcher — there is one budget and one underlying
+/// `mz_ore::pager` — but whether the stash *uses* it is gated by storage's own
+/// `enable_upsert_paged_spill` flag, independently of compute's
+/// `enable_column_paged_batcher_spill`. The shared pool's budget / backend /
+/// codec are configured by compute's `apply_tiered_config` (storage and compute
+/// run in the same `clusterd` process).
+///
+/// Flipping the flag takes effect on dataflows created after the change: the
+/// paged flavor captures the pager once at operator construction.
+///
+/// [`TieredPolicy`]: mz_timely_util::column_pager::policy::TieredPolicy
+pub mod upsert_stash_pager {
+    use std::sync::{LazyLock, RwLock};
+
+    use mz_timely_util::column_pager::{ColumnPager, shared_pager};
+
+    /// Active pager handed to upsert source-stash batchers. Defaults to
+    /// disabled (every chunk resident) until [`set_enabled`] turns it on.
+    static PAGER: LazyLock<RwLock<ColumnPager>> =
+        LazyLock::new(|| RwLock::new(ColumnPager::disabled()));
+
+    /// Enable or disable the stash's use of the shared column pager. When
+    /// enabled, the stash spills through the shared budget pool; when disabled
+    /// it keeps every chunk resident.
+    pub fn set_enabled(enabled: bool) {
+        *PAGER.write().expect("upsert stash pager poisoned") = shared_pager(enabled);
+    }
+
+    /// The current upsert-stash pager. Cheap: clones the inner `Arc`.
+    pub fn pager() -> ColumnPager {
+        PAGER.read().expect("upsert stash pager poisoned").clone()
+    }
+}
+
 impl Debug for UpsertKey {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "0x")?;
@@ -604,6 +642,7 @@ pub(crate) fn upsert_v2<'scope, T, FromTime>(
     previous_token: Option<Vec<PressOnDropButton>>,
     source_config: crate::source::SourceExportCreationConfig,
     backpressure_metrics: Option<BackpressureMetrics>,
+    stash_flavor: upsert_continual_feedback_v2::UpsertStashFlavor,
 ) -> (
     VecCollection<'scope, T, Result<Row, DataflowError>, Diff>,
     StreamVec<'scope, T, (Option<GlobalId>, HealthStatusUpdate)>,
@@ -630,10 +669,12 @@ where
     tracing::info!(
         worker_id = %source_config.worker_id,
         source_id = %source_config.id,
+        ?stash_flavor,
         "rendering upsert source (btreemap backend)"
     );
 
     upsert_continual_feedback_v2::upsert_inner(
+        stash_flavor,
         thin_input,
         upsert_envelope.key_indices,
         resume_upper,