timely-util: spill young chunk generations uncompressed

misc/python/materialize/mzcompose/__init__.py

@@ -595,6 +595,7 @@ def get_default_system_parameters(
     "linear_join_yielding",
     "enable_column_paged_batcher",
     "enable_column_paged_batcher_spill",
+    "column_chunk_compress_min_depth",
     "column_paged_batcher_budget_fraction",
     "column_paged_batcher_lz4",
     "column_paged_batcher_swap_pageout",

misc/python/materialize/parallel_workload/action.py

@@ -3055,6 +3055,11 @@ def __init__(
             "0.02",
         ]
         self.flags_with_values["enable_upsert_paged_spill"] = BOOLEAN_FLAG_VALUES
+        self.flags_with_values["column_chunk_compress_min_depth"] = [
+            "0",  # compress every spilled body
+            "1",  # the default: fresh chunks store uncompressed
+            "4",  # exempt several young generations
+        ]
         # 0 forces the estimated-size path for every table, the default forces
         # the exact COUNT(*) path for workload-sized tables.
         self.flags_with_values["mysql_source_snapshot_exact_count_max_rows"] = [

src/compute-types/src/dyncfgs.rs

@@ -86,6 +86,28 @@ pub const ENABLE_COLUMN_PAGED_BATCHER_SPILL: Config<bool> = Config::new(
     ParameterScope::Replica,
 );
 
+/// The youngest chunk generation whose spilled bodies are compressed.
+///
+/// A chunk at generational depth `d` is rewritten with frequency
+/// proportional to `2^-d` under geometric merging, so compressing shallow
+/// generations buys pool bytes back for only a short stay at a guaranteed
+/// near-term codec round-trip. Generations below the floor spill under the
+/// identity codec: fully budgeted and swap-backed, with encode and decode
+/// reduced to copies. The default exempts only fresh (depth 0) chunks. A
+/// chunk that outlives a merge untouched ages a generation regardless, and
+/// an identity-coded body at or past the floor is re-spilled compressed at
+/// its next survival, so key-disjoint input cannot hold its backlog
+/// uncompressed indefinitely. Lowering this at runtime therefore migrates
+/// bodies that already spilled, rather than applying only to new ones.
+/// `0` compresses every spilled body.
+pub const COLUMN_CHUNK_COMPRESS_MIN_DEPTH: Config<u32> = Config::new(
+    "column_chunk_compress_min_depth",
+    1,
+    "The youngest chunk generation whose spilled bodies are lz4-compressed in the buffer \
+     pool; younger generations store uncompressed. 0 compresses every spilled body.",
+    ParameterScope::Replica,
+);
+
 /// Resident-bytes budget fraction for chunk spilling. Two consumers read
 /// it: the column pager's tiered policy multiplies it against the
 /// announced memory limit, and the buffer pool (`mz_ore::pool`)
@@ -691,4 +713,5 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&COLUMN_PAGED_BATCHER_SPILL_WORKER_COUNT)
         .add(&COLUMN_PAGED_BATCHER_EAGER_BACKING)
         .add(&COLUMN_PAGED_BATCHER_POOL_RSS_TARGET_FRACTION)
+        .add(&COLUMN_CHUNK_COMPRESS_MIN_DEPTH)
 }

src/compute/src/compute_state.rs

@@ -428,6 +428,13 @@ impl ComputeState {
                     warn!("chunk spill: buffer pool unavailable; chunks stay resident");
                 }
             }
+
+            // The generational depth floor below which spilled bodies store
+            // uncompressed. Subsystem-independent, so applied here alongside
+            // the rest of the process-wide chunk configuration.
+            let compress_min_depth =
+                u8::try_from(COLUMN_CHUNK_COMPRESS_MIN_DEPTH.get(config)).unwrap_or(u8::MAX);
+            mz_timely_util::columnar::chunk::set_compress_min_depth(compress_min_depth);
         }
 
         // Remember the maintenance interval locally to avoid reading it from the config set on

src/ore/src/pool.rs

@@ -126,6 +126,28 @@ pub trait ExtentCodec: std::fmt::Debug + Send + Sync {
     fn decode(&self, stored: &[u8], body: &mut [u8]);
 }
 
+/// The identity [`ExtentCodec`]: the stored form is the body. Encode and
+/// decode are copies, and range reads copy the range directly, so a chunk
+/// stored under this codec pays no compression work in either direction
+/// while remaining fully budgeted and swap-backed like any other extent.
+#[derive(Debug)]
+pub struct IdentityCodec;
+
+/// The [`IdentityCodec`] instance to pass to [`Pool::insert_with`].
+pub static IDENTITY_CODEC: IdentityCodec = IdentityCodec;
+
+impl ExtentCodec for IdentityCodec {
+    fn encode(&self, body: &[u8], out: &mut Vec<u8>) {
+        out.clear();
+        out.extend_from_slice(body);
+    }
+
+    fn decode(&self, stored: &[u8], body: &mut [u8]) {
+        assert_eq!(stored.len(), body.len(), "identity stored form is the body");
+        body.copy_from_slice(stored);
+    }
+}
+
 /// The largest stored form [`ExtentCodec::encode`] may produce for a
 /// `body_len`-byte body: an incompressible-input expansion matching lz4's
 /// worst case plus a four-byte length prefix. The extent store's size-class
@@ -3703,6 +3725,24 @@ mod tests {
         assert_eq!(pool.stats().resident_bytes, 0);
     }
 
+    /// The identity codec stores the body verbatim: eviction and reads,
+    /// whole and by range, reconstruct it unchanged.
+    #[mz_ore::test]
+    fn identity_codec_round_trips() {
+        let pool = test_pool(usize::MAX);
+        let want = payload(SMALL, 601);
+        let h = pool.insert_with(SMALL, ChunkHints::default(), &IDENTITY_CODEC, |dst| {
+            dst.copy_from_slice(&want);
+        });
+        assert_eq!(read(&h), want);
+        pool.evict(&h);
+        assert_eq!(read(&h), want, "round-trips through the extent");
+        pool.evict(&h);
+        let mut range = Vec::new();
+        h.read_range_into(8..24, &mut range);
+        assert_eq!(range, want[8..24], "range reads copy the range directly");
+    }
+
     #[mz_ore::test]
     fn insert_with_fills_in_place() {
         let pool = test_pool(usize::MAX);