Add MADV_PAGEOUT support for eager eviction of compressed spill

misc/python/materialize/mzcompose/__init__.py

@@ -498,6 +498,7 @@ def get_default_system_parameters(
     "enable_column_paged_batcher_spill",
     "column_paged_batcher_budget_fraction",
     "column_paged_batcher_lz4",
+    "column_paged_batcher_swap_pageout",
     "enable_upsert_paged_spill",
     "enable_lgalloc_eager_reclamation",
     "lgalloc_background_interval",

misc/python/materialize/parallel_workload/action.py

@@ -1618,6 +1618,9 @@ def __init__(
             "0.25",
         ]
         self.flags_with_values["column_paged_batcher_lz4"] = BOOLEAN_FLAG_VALUES
+        self.flags_with_values["column_paged_batcher_swap_pageout"] = (
+            BOOLEAN_FLAG_VALUES
+        )
         self.flags_with_values["enable_upsert_paged_spill"] = BOOLEAN_FLAG_VALUES
 
         # If you are adding a new config flag in Materialize, consider using it

src/compute-types/src/dyncfgs.rs

@@ -93,6 +93,28 @@ pub const COLUMN_PAGED_BATCHER_LZ4: Config<bool> = Config::new(
      `enable_column_paged_batcher_spill = true`.",
 );
 
+/// Proactively evict the column-paged batcher's lz4-compressed spill chunks
+/// from RSS via `MADV_PAGEOUT` when spilling to the swap backend. Only
+/// meaningful when [`COLUMN_PAGED_BATCHER_LZ4`] is `true` and the active
+/// backend is swap (no scratch directory): on that path the compressed bytes
+/// stay resident in the process address space and currently receive no madvise
+/// at all, so the kernel reclaims them only lazily under LRU pressure.
+/// `MADV_PAGEOUT` instead swaps them out eagerly at spill time, holding RSS at
+/// the budget rather than letting it drift up to the pressure cliff. A later
+/// page-in re-faults the pages — cheap because lz4 shrank the byte volume,
+/// which is what makes eager eviction pay off on this path.
+///
+/// Off by default: the eager-reclaim syscall is the one kernel interaction the
+/// pager design singled out as risky, so it stays gated until proven on the
+/// target workload.
+pub const COLUMN_PAGED_BATCHER_SWAP_PAGEOUT: Config<bool> = Config::new(
+    "column_paged_batcher_swap_pageout",
+    false,
+    "Eagerly evict the column-paged batcher's lz4-compressed swap-backend spill chunks from RSS \
+     via `MADV_PAGEOUT` (they otherwise receive no madvise and are reclaimed only lazily). Only \
+     meaningful when `column_paged_batcher_lz4 = true` and the swap backend is active.",
+);
+
 /// Whether rendering should use `mz_join_core` rather than DD's `JoinCore::join_core`.
 pub const ENABLE_MZ_JOIN_CORE: Config<bool> = Config::new(
     "enable_mz_join_core",
@@ -498,4 +520,5 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&ENABLE_COLUMN_PAGED_BATCHER_SPILL)
         .add(&COLUMN_PAGED_BATCHER_BUDGET_FRACTION)
         .add(&COLUMN_PAGED_BATCHER_LZ4)
+        .add(&COLUMN_PAGED_BATCHER_SWAP_PAGEOUT)
 }

src/compute/src/compute_state.rs

@@ -323,6 +323,7 @@ impl ComputeState {
 
             let enabled = ENABLE_COLUMN_PAGED_BATCHER_SPILL.get(config);
             let codec = COLUMN_PAGED_BATCHER_LZ4.get(config).then_some(Codec::Lz4);
+            let swap_pageout = COLUMN_PAGED_BATCHER_SWAP_PAGEOUT.get(config);
 
             // Budget derivation: fraction × announced memory limit, with a
             // 128 MiB floor so the no-pressure case doesn't page per chunk.
@@ -344,12 +345,13 @@ impl ComputeState {
                 enabled,
                 ?backend,
                 ?codec,
+                swap_pageout,
                 fraction,
                 mem_limit,
                 budget_bytes = total,
                 "column-paged batcher: applying tiered config",
             );
-            apply_tiered_config(enabled, total, backend, codec);
+            apply_tiered_config(enabled, total, backend, codec, swap_pageout);
         }
 
         // Remember the maintenance interval locally to avoid reading it from the config set on

src/ore/src/pager.rs

@@ -21,6 +21,7 @@ mod file;
 mod swap;
 
 pub use file::set_scratch_dir;
+pub use swap::advise_pageout;
 
 use crate::pager::file::FileInner;
 use crate::pager::swap::SwapInner;

src/ore/src/pager/swap.rs

@@ -59,20 +59,62 @@ pub(crate) fn pageout_swap(chunks: &mut [Vec<u64>]) -> Handle {
     Handle::from_swap(SwapInner::new(taken))
 }
 
+/// Proactively reclaims (swaps out) the resident pages of `bytes` via
+/// `MADV_PAGEOUT`, holding RSS at the caller's budget right now rather than
+/// waiting for kernel LRU to reclaim under pressure the way `pageout_swap`'s
+/// `MADV_COLD` hint does.
+///
+/// Unlike `pageout_swap`, this takes a borrow and does **not** transfer
+/// ownership: the allocation stays addressable in the caller's address space,
+/// so a later read simply re-faults the swapped-out pages back in. That suits a
+/// buffer the caller must keep reachable — e.g. the column pager's
+/// lz4-compressed bytes kept in memory — but still wants evicted eagerly so the
+/// budget is real instead of a fiction the kernel only honors at the pressure
+/// cliff.
+///
+/// On non-Linux targets this is a no-op (matching `MADV_COLD`).
+pub fn advise_pageout(bytes: &[u8]) {
+    madvise_pageout(bytes);
+}
+
 #[cfg(target_os = "linux")]
 fn madvise_cold(chunk: &[u64]) {
-    if chunk.is_empty() {
-        return;
-    }
-    let page = page_size();
-    let base_ptr = chunk.as_ptr();
-    let base_addr = base_ptr.addr();
     // `Vec<u64>` cannot exceed `isize::MAX` bytes, so this multiplication
     // cannot overflow on any supported target. Use `checked_mul` for
     // defense-in-depth: a corrupted length should fail loudly, not wrap.
     let Some(len_bytes) = chunk.len().checked_mul(std::mem::size_of::<u64>()) else {
         return;
     };
+    // SAFETY: `(ptr, len_bytes)` describes the live `&[u64]` exactly.
+    unsafe { madvise_aligned(chunk.as_ptr().cast::<u8>(), len_bytes, libc::MADV_COLD) }
+}
+
+#[cfg(target_os = "linux")]
+fn madvise_pageout(bytes: &[u8]) {
+    // SAFETY: `(ptr, len)` describes the live `&[u8]` exactly.
+    unsafe { madvise_aligned(bytes.as_ptr(), bytes.len(), libc::MADV_PAGEOUT) }
+}
+
+/// Issues `madvise(advice)` over the page-aligned interior of the byte range
+/// `[base_ptr, base_ptr + len_bytes)`. `madvise` operates at page granularity,
+/// so the start rounds up and the end rounds down to page boundaries; a range
+/// that contains no whole page is skipped so we never advise pages we only
+/// partially own.
+///
+/// # Safety
+///
+/// `base_ptr` must point to the start of a live allocation of at least
+/// `len_bytes` bytes that stays valid for the duration of the call. `advice`
+/// must be a non-mutating hint (`MADV_COLD`/`MADV_PAGEOUT`): both only change
+/// the kernel's reclaim decision and leave the bytes readable, so concurrent
+/// reads of the range remain sound.
+#[cfg(target_os = "linux")]
+unsafe fn madvise_aligned(base_ptr: *const u8, len_bytes: usize, advice: libc::c_int) {
+    if len_bytes == 0 {
+        return;
+    }
+    let page = page_size();
+    let base_addr = base_ptr.addr();
     // Round the start up and the end down to page boundaries. Both additions
     // use `checked_add` so that an allocation sitting near the top of the
     // address space can never silently wrap into a tiny range.
@@ -91,23 +133,26 @@ fn madvise_cold(chunk: &[u64]) {
     // SAFETY: `aligned_start_addr` lies in `[base_addr, base_addr + len_bytes]`
     // by construction (rounding up the start cannot exceed `end_unaligned`,
     // which equals `base_addr + len_bytes`; the early-return above guarantees
-    // `start ≤ end`). That interval is exactly the range covered by the live
-    // `&[u64]`, so `byte_add` stays in-bounds and preserves provenance.
+    // `start ≤ end`). That interval is within the live allocation the caller
+    // promised, so `byte_add` stays in-bounds and preserves provenance.
     let aligned_ptr = unsafe { base_ptr.byte_add(aligned_start_addr - base_addr) }
         .cast::<libc::c_void>()
         .cast_mut();
     // SAFETY: pointer/length describe a fully page-aligned subrange contained
-    // within the live `&[u64]` (justified above). `MADV_COLD` is non-mutating;
-    // it only signals reclaim preference to the kernel, so concurrent reads
-    // of the slice remain sound.
+    // within the live allocation (justified above). The caller guarantees
+    // `advice` is a non-mutating reclaim hint, so concurrent reads of the range
+    // remain sound.
     unsafe {
-        libc::madvise(aligned_ptr, aligned_len, libc::MADV_COLD);
+        libc::madvise(aligned_ptr, aligned_len, advice);
     }
 }
 
 #[cfg(not(target_os = "linux"))]
 fn madvise_cold(_chunk: &[u64]) {}
 
+#[cfg(not(target_os = "linux"))]
+fn madvise_pageout(_bytes: &[u8]) {}
+
 #[cfg(target_os = "linux")]
 fn page_size() -> usize {
     // SAFETY: `sysconf` with a valid argument is safe.
@@ -250,4 +295,25 @@ mod tests {
         take_swap(h, &mut dst);
         assert_eq!(dst, vec![1, 2, 3, 4, 5]);
     }
+
+    #[mz_ore::test]
+    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `madvise` on OS `linux`
+    fn advise_pageout_leaves_bytes_readable() {
+        // `MADV_PAGEOUT` is a reclaim hint: the bytes must remain addressable
+        // and unchanged afterwards (a read re-faults the pages back in). Use a
+        // multi-page buffer so the page-aligned interior is non-empty.
+        let pattern = |i: usize| u8::try_from(i % 251).expect("0..251 fits in u8");
+        let bytes: Vec<u8> = (0..64 * 1024).map(pattern).collect();
+        advise_pageout(&bytes);
+        // Re-read after the advice; contents are preserved.
+        assert!(bytes.iter().enumerate().all(|(i, &b)| b == pattern(i)));
+    }
+
+    #[mz_ore::test]
+    fn advise_pageout_empty_and_subpage_are_noops() {
+        // Neither an empty slice nor a sub-page slice contains a whole page, so
+        // both skip the syscall entirely; they must not panic.
+        advise_pageout(&[]);
+        advise_pageout(&[1u8, 2, 3, 4]);
+    }
 }

src/timely-util/src/column_pager.rs

@@ -36,6 +36,7 @@ pub mod metrics;
 pub mod policy;
 
 use std::io::{self, Read};
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, LazyLock, RwLock};
 
 use columnar::Columnar;
@@ -257,6 +258,17 @@ impl PagingPolicy for AlwaysResidentPolicy {
 static GLOBAL_PAGER: LazyLock<RwLock<ColumnPager>> =
     LazyLock::new(|| RwLock::new(ColumnPager::disabled()));
 
+/// Process-global toggle for `MADV_PAGEOUT` on the lz4 + swap spill path.
+///
+/// When set, [`ColumnPager::page`] issues `MADV_PAGEOUT` over the compressed
+/// bytes it keeps resident in [`CompressedInner::Memory`], proactively evicting
+/// them at spill time instead of leaving them for lazy kernel reclaim. A single
+/// process-global flag (set by [`apply_tiered_config`]) mirrors the backend /
+/// codec selection: it is a process-wide operational choice, not a per-column
+/// one, and every consumer of the shared pager reads the same value. Defaults
+/// to off; the eager-reclaim syscall stays gated until proven.
+static SWAP_PAGEOUT: AtomicBool = AtomicBool::new(false);
+
 /// Install `pager` as the process-wide active pager. Subsequent
 /// [`global_pager`] calls return a clone of this value across all threads.
 ///
@@ -300,12 +312,18 @@ pub fn tiered_policy() -> &'static policy::TieredPolicy {
 /// in-flight tickets still credit the singleton, which is harmless: the
 /// budget grows above the configured total until the next enable reconciles
 /// it via `reconfigure`.
+///
+/// `swap_pageout` toggles `MADV_PAGEOUT` on the lz4 + swap spill path (see
+/// `SWAP_PAGEOUT`); it is stored unconditionally so the next `page` call
+/// observes it regardless of `enabled`.
 pub fn apply_tiered_config(
     enabled: bool,
     total_budget: usize,
     backend: Backend,
     codec: Option<Codec>,
+    swap_pageout: bool,
 ) {
+    SWAP_PAGEOUT.store(swap_pageout, Ordering::Relaxed);
     let p: &Arc<policy::TieredPolicy> = &TIERED_POLICY;
     p.reconfigure(total_budget, backend, codec);
     if enabled {
@@ -431,7 +449,21 @@ impl ColumnPager {
                     codec: Some(Codec::Lz4),
                 });
                 let inner = match backend {
-                    Backend::Swap => CompressedInner::Memory(out),
+                    Backend::Swap => {
+                        // The compressed bytes stay resident in our own address
+                        // space (we read them back in `take` via `FrameDecoder`),
+                        // so the pager's ownership-transferring `pageout` does not
+                        // fit. When `SWAP_PAGEOUT` is set, hint `MADV_PAGEOUT`
+                        // instead: it proactively swaps the pages out now, holding
+                        // RSS at the budget rather than leaving them as unmanaged
+                        // anonymous memory the kernel only reclaims lazily at the
+                        // pressure cliff. A later read re-faults them back in —
+                        // cheap, since lz4 shrank the byte volume.
+                        if SWAP_PAGEOUT.load(Ordering::Relaxed) {
+                            pager::advise_pageout(&out);
+                        }
+                        CompressedInner::Memory(out)
+                    }
                     Backend::File => {
                         // The pager deals in `Vec<u64>`, so the framed bytes
                         // must be widened. `out` is already compressed (~4x
@@ -640,6 +672,31 @@ mod tests {
         assert_eq!(collect_i64(&rt), (0i64..1024).collect::<Vec<_>>());
     }
 
+    /// With the swap-pageout flag on, the lz4 + swap path issues `MADV_PAGEOUT`
+    /// over the compressed bytes; the round-trip must still reproduce the input
+    /// (the advice is a non-destructive reclaim hint). Drives the global pager
+    /// through `apply_tiered_config` — the only path that sets the flag — with a
+    /// zero budget so every column spills. Resets the globals on the way out so
+    /// peer tests see the default disabled pager.
+    #[mz_ore::test]
+    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `madvise` on OS `linux`
+    fn round_trip_swap_lz4_pageout() {
+        apply_tiered_config(true, 0, Backend::Swap, Some(Codec::Lz4), true);
+        let cp = global_pager();
+        let mut col = sample_typed();
+        let paged = cp.page(&mut col);
+        assert!(matches!(
+            paged,
+            PagedColumn::Compressed {
+                inner: CompressedInner::Memory(_),
+                ..
+            }
+        ));
+        let rt = cp.take(paged);
+        assert_eq!(collect_i64(&rt), (0i64..1024).collect::<Vec<_>>());
+        apply_tiered_config(false, 0, Backend::Swap, None, false);
+    }
+
     #[mz_ore::test]
     #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `writev` on OS `linux`
     fn round_trip_file_uncompressed() {