storage/upsert: v2 stash on ChunkBatcher; feedback arrangement as ChunkSpine with the UnloadChunk drain

Author: DAlperin

misc/python/materialize/mzcompose/__init__.py

@@ -299,6 +299,11 @@ def get_variable_system_parameters(
             "true",
             ["true", "false"],
         ),
+        VariableSystemParameter(
+            "enable_upsert_paged_spill",
+            "true",
+            ["true", "false"],
+        ),
         VariableSystemParameter(
             "enable_upsert_v2",
             "false",
@@ -598,7 +603,6 @@ def get_default_system_parameters(
     "column_paged_batcher_spill_worker_count",
     "column_paged_batcher_eager_backing",
     "column_paged_batcher_pool_rss_target_fraction",
-    "enable_upsert_paged_spill",
     "enable_lgalloc_eager_reclamation",
     "lgalloc_background_interval",
     "lgalloc_file_growth_dampener",

src/storage-types/src/dyncfgs.rs

@@ -10,7 +10,7 @@
 //! Dyncfgs used by the storage layer. Despite their name, these can be used
 //! "statically" during rendering, or dynamically within timely operators.
 
-use mz_dyncfg::{Config, ConfigSet};
+use mz_dyncfg::{Config, ConfigSet, ParameterScope};
 use std::time::Duration;
 
 /// When dataflows observe an invariant violation it is either due to a bug or due to the cluster
@@ -328,22 +328,23 @@ pub const STORAGE_UPSERT_MAX_SNAPSHOT_BATCH_BUFFERING: Config<Option<usize>> = C
     "Limit snapshot buffering in upsert.",
 );
 
-/// Allow the upsert-v2 source stash's chunk batcher to spill cold chains out
-/// of RSS via the process buffer pool. The stash draws from the same shared
-/// pool budget as the compute chunk batchers — there is one budget — but
-/// this flag gates the stash's participation independently of the
-/// compute-side `enable_column_paged_batcher_spill`.
+/// 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.
 ///
-/// Off by default; the stash keeps every chunk resident until enabled.
-/// 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.
+/// 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.
 pub const ENABLE_UPSERT_PAGED_SPILL: Config<bool> = Config::new(
     "enable_upsert_paged_spill",
     false,
-    "Allow the upsert-v2 source stash to spill chunks to the shared buffer pool, gated \
-     independently of the compute `enable_column_paged_batcher_spill`.",
-);
+    "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`.",
+)
+.scoped(ParameterScope::Replica);
 
 // RocksDB
 

src/storage/src/storage_state.rs

@@ -846,17 +846,27 @@ impl<'w> Worker<'w> {
                     STORAGE_SERVER_MAINTENANCE_INTERVAL
                         .get(self.storage_state.storage_configuration.config_set());
 
-                // Gate the upsert source-stash's use of the column pager. The
-                // pager's budget pool, backend, and codec are the shared ones
+                // 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); storage only decides whether
-                // its stash participates, via its own dyncfg.
+                // storage run in the same process). The gate ORs this 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`
+                // commands it sends, so reading this worker's `ConfigSet`
+                // here observes them.
                 {
                     use mz_storage_types::dyncfgs::ENABLE_UPSERT_PAGED_SPILL;
 
                     let enabled = ENABLE_UPSERT_PAGED_SPILL
                         .get(self.storage_state.storage_configuration.config_set());
-                    crate::upsert::upsert_stash_pager::set_enabled(enabled);
+                    info!(
+                        worker = self.timely_worker.index(),
+                        enabled, "upsert stash spill: applying gate",
+                    );
+                    crate::upsert::upsert_stash_spill::set_enabled(enabled);
                 }
             }
             InternalStorageCommand::StatisticsUpdate { sources, sinks } => self

src/storage/src/upsert.rs

@@ -296,37 +296,22 @@ macro_rules! upsert_source_time_unit {
 }
 upsert_source_time_unit!(GtidPartition, Lsn);
 
-/// Pager for the upsert-v2 source stash.
+/// Storage's leg of the process-wide chunk spill gate.
 ///
-/// 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).
+/// The upsert-v2 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).
 ///
-/// [`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.
+/// The gate is process-wide with one leg per subsystem, and chunks spill
+/// while either leg is set. Storage sets its leg from
+/// `enable_upsert_paged_spill`, so that flag alone cannot veto spilling
+/// enabled by compute's leg. The gate is consulted at every chunk commit, so
+/// flips apply to running dataflows.
+pub mod upsert_stash_spill {
+    /// Enable or disable spilling of upsert chunk bodies to the buffer pool.
     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()
+        mz_timely_util::columnar::chunk::set_storage_spill_enabled(enabled);
     }
 }