feat(query): add spill backoff with sleep for low-memory queries under global pressure

When global memory pressure triggers spill but the current query's memory
usage is low, sleep with exponential backoff before spilling to give other
queries a chance to release memory. Rename low_query_threshold to
low_query_memory_bytes for clarity and add observability logging.

Author: dqhl76

src/query/pipeline/transforms/src/processors/memory_settings.rs

@@ -20,6 +20,62 @@ use bytesize::ByteSize;
 use databend_common_base::runtime::GLOBAL_MEM_STAT;
 use databend_common_base::runtime::MemStat;
 use databend_common_base::runtime::ThreadTracker;
+use log::info;
+
+const GLOBAL_PRESSURE_SLEEP_BACKOFF_INIT_MS: u64 = 200;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SpillDecision {
+    NoSpill,
+    SpillNow,
+    Sleep(u64),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct SpillBackoffSettings {
+    pub max_sleep_ms: u64,
+    pub low_query_memory_bytes: u64,
+}
+
+impl SpillBackoffSettings {
+    fn should_backoff(&self, query_usage: usize) -> bool {
+        query_usage <= self.low_query_memory_bytes as usize
+    }
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct SpillBackoffState {
+    consumed_sleep_ms: u64,
+    attempts: u32,
+}
+
+impl SpillBackoffState {
+    fn reset(&mut self) {
+        self.consumed_sleep_ms = 0;
+        self.attempts = 0;
+    }
+
+    fn next_sleep_ms(&mut self, max_sleep_ms: u64) -> Option<u64> {
+        if max_sleep_ms == 0 {
+            return None;
+        }
+
+        let remaining_budget = max_sleep_ms.saturating_sub(self.consumed_sleep_ms);
+        if remaining_budget == 0 {
+            return None;
+        }
+
+        let mut delay_ms = GLOBAL_PRESSURE_SLEEP_BACKOFF_INIT_MS;
+        for _ in 0..self.attempts.min(16) {
+            delay_ms = delay_ms.saturating_mul(2);
+        }
+
+        let actual_sleep_ms = delay_ms.min(remaining_budget);
+        self.consumed_sleep_ms = self.consumed_sleep_ms.saturating_add(actual_sleep_ms);
+        self.attempts = self.attempts.saturating_add(1);
+        Some(actual_sleep_ms)
+    }
+}
 
 #[derive(Clone)]
 #[non_exhaustive]
@@ -35,6 +91,8 @@ pub struct MemorySettings {
     pub enable_query_level_spill: bool,
     pub max_query_memory_usage: usize,
     pub query_memory_tracking: Option<Arc<MemStat>>,
+
+    pub spill_backoff: Option<SpillBackoffSettings>,
 }
 
 impl Debug for MemorySettings {
@@ -69,6 +127,7 @@ impl Debug for MemorySettings {
         }
 
         f.field("enable_query_level_spill", &self.enable_query_level_spill)
+            .field("spill_backoff", &self.spill_backoff)
             .field("spill_unit_size", &self.spill_unit_size)
             .finish()
     }
@@ -82,6 +141,8 @@ pub struct MemorySettingsBuilder {
     max_query_memory_usage: Option<usize>,
     query_memory_tracking: Option<Arc<MemStat>>,
 
+    spill_backoff: Option<SpillBackoffSettings>,
+
     spill_unit_size: Option<usize>,
 }
 
@@ -111,6 +172,11 @@ impl MemorySettingsBuilder {
         self
     }
 
+    pub fn with_spill_backoff(mut self, spill_backoff: Option<SpillBackoffSettings>) -> Self {
+        self.spill_backoff = spill_backoff;
+        self
+    }
+
     pub fn build(self) -> MemorySettings {
         MemorySettings {
             enable_group_spill: self.enable_group_spill,
@@ -123,6 +189,8 @@ impl MemorySettingsBuilder {
             max_query_memory_usage: self.max_query_memory_usage.unwrap_or(usize::MAX),
             query_memory_tracking: self.query_memory_tracking,
 
+            spill_backoff: self.spill_backoff,
+
             spill_unit_size: self.spill_unit_size.unwrap_or(0),
         }
     }
@@ -138,6 +206,8 @@ impl MemorySettings {
             max_query_memory_usage: None,
             query_memory_tracking: None,
 
+            spill_backoff: None,
+
             spill_unit_size: None,
         }
     }
@@ -164,6 +234,55 @@ impl MemorySettings {
         }
     }
 
+    pub fn check_spill_with_backoff(&self, backoff_state: &mut SpillBackoffState) -> SpillDecision {
+        let decision = (|| {
+            if !self.check_spill() {
+                return SpillDecision::NoSpill;
+            }
+
+            let Some(spill_backoff) = self.spill_backoff else {
+                return SpillDecision::SpillNow;
+            };
+
+            let Some(query_usage) = self.current_query_usage() else {
+                return SpillDecision::SpillNow;
+            };
+
+            if !spill_backoff.should_backoff(query_usage) {
+                return SpillDecision::SpillNow;
+            }
+
+            match backoff_state.next_sleep_ms(spill_backoff.max_sleep_ms) {
+                Some(sleep_ms) => {
+                    info!(
+                        "Spill backoff: sleeping {}ms (query_usage={}, threshold={}, consumed={}ms/{}ms, attempt={})",
+                        sleep_ms,
+                        ByteSize(query_usage as u64),
+                        ByteSize(spill_backoff.low_query_memory_bytes),
+                        backoff_state.consumed_sleep_ms,
+                        spill_backoff.max_sleep_ms,
+                        backoff_state.attempts,
+                    );
+                    SpillDecision::Sleep(sleep_ms)
+                }
+                None => {
+                    info!(
+                        "Spill backoff: budget exhausted, proceeding to spill (query_usage={}, consumed={}ms/{}ms)",
+                        ByteSize(query_usage as u64),
+                        backoff_state.consumed_sleep_ms,
+                        spill_backoff.max_sleep_ms,
+                    );
+                    SpillDecision::SpillNow
+                }
+            }
+        })();
+
+        if !matches!(decision, SpillDecision::Sleep(_)) {
+            backoff_state.reset();
+        }
+        decision
+    }
+
     fn check_global(&self) -> Option<isize> {
         self.enable_global_level_spill.then(|| {
             let usage = self.global_memory_tracking.get_memory_usage();
@@ -200,7 +319,7 @@ impl MemorySettings {
             return None;
         }
 
-        let usage = self.query_memory_tracking.as_ref()?.get_memory_usage();
+        let usage = self.current_query_usage()?;
 
         Some(if usage >= self.max_query_memory_usage {
             -((usage - self.max_query_memory_usage) as isize)
@@ -219,6 +338,12 @@ impl MemorySettings {
         .flatten()
         .reduce(|a, b| a.min(b))
     }
+
+    fn current_query_usage(&self) -> Option<usize> {
+        self.query_memory_tracking
+            .as_ref()
+            .map(|tracking| tracking.get_memory_usage())
+    }
 }
 
 #[cfg(test)]
@@ -239,6 +364,7 @@ mod tests {
                 max_query_memory_usage: 0,
                 query_memory_tracking: None,
                 enable_query_level_spill: false,
+                spill_backoff: None,
                 spill_unit_size: 4096,
             }
         }
@@ -370,4 +496,99 @@ mod tests {
         };
         assert!(!settings.check_spill());
     }
+
+    #[test]
+    fn backoff_sleeps_when_spill_triggered_and_query_is_low_memory() {
+        let query_mem = create_mem_stat(40);
+        let global_mem = create_static_mem_stat(100);
+        let settings = MemorySettings {
+            enable_global_level_spill: true,
+            global_memory_tracking: global_mem,
+            max_memory_usage: 100,
+            query_memory_tracking: Some(query_mem),
+            spill_backoff: Some(SpillBackoffSettings {
+                max_sleep_ms: 500,
+                low_query_memory_bytes: 50,
+            }),
+            ..Default::default()
+        };
+
+        let mut backoff_state = SpillBackoffState::default();
+
+        assert_eq!(
+            settings.check_spill_with_backoff(&mut backoff_state),
+            SpillDecision::Sleep(200)
+        );
+        assert_eq!(
+            settings.check_spill_with_backoff(&mut backoff_state),
+            SpillDecision::Sleep(300)
+        );
+        assert_eq!(
+            settings.check_spill_with_backoff(&mut backoff_state),
+            SpillDecision::SpillNow
+        );
+    }
+
+    #[test]
+    fn backoff_spills_immediately_when_query_is_not_low_memory() {
+        let query_mem = create_mem_stat(60);
+        let global_mem = create_static_mem_stat(100);
+        let settings = MemorySettings {
+            enable_global_level_spill: true,
+            global_memory_tracking: global_mem,
+            max_memory_usage: 100,
+            query_memory_tracking: Some(query_mem),
+            spill_backoff: Some(SpillBackoffSettings {
+                max_sleep_ms: 500,
+                low_query_memory_bytes: 50,
+            }),
+            ..Default::default()
+        };
+
+        let mut backoff_state = SpillBackoffState::default();
+
+        assert_eq!(
+            settings.check_spill_with_backoff(&mut backoff_state),
+            SpillDecision::SpillNow
+        );
+    }
+
+    #[test]
+    fn backoff_resets_after_pressure_is_relieved() {
+        let query_mem = create_mem_stat(40);
+        let pressured_global_mem = create_static_mem_stat(100);
+        let relaxed_global_mem = create_static_mem_stat(80);
+
+        let pressured_settings = MemorySettings {
+            enable_global_level_spill: true,
+            global_memory_tracking: pressured_global_mem,
+            max_memory_usage: 100,
+            query_memory_tracking: Some(query_mem.clone()),
+            spill_backoff: Some(SpillBackoffSettings {
+                max_sleep_ms: 1000,
+                low_query_memory_bytes: 50,
+            }),
+            ..Default::default()
+        };
+
+        let relaxed_settings = MemorySettings {
+            global_memory_tracking: relaxed_global_mem,
+            ..pressured_settings.clone()
+        };
+
+        let mut backoff_state = SpillBackoffState::default();
+
+        assert_eq!(
+            pressured_settings.check_spill_with_backoff(&mut backoff_state),
+            SpillDecision::Sleep(200)
+        );
+        assert_eq!(
+            relaxed_settings.check_spill_with_backoff(&mut backoff_state),
+            SpillDecision::NoSpill
+        );
+        assert_eq!(
+            pressured_settings.check_spill_with_backoff(&mut backoff_state),
+            SpillDecision::Sleep(200)
+        );
+    }
 }

src/query/pipeline/transforms/src/processors/mod.rs

@@ -17,4 +17,7 @@ pub mod traits;
 mod transforms;
 
 pub use memory_settings::MemorySettings;
+pub use memory_settings::SpillBackoffSettings;
+pub use memory_settings::SpillBackoffState;
+pub use memory_settings::SpillDecision;
 pub use transforms::*;

src/query/service/src/pipelines/memory_settings.rs

@@ -15,6 +15,7 @@
 use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::Result;
 use databend_common_pipeline_transforms::MemorySettings;
+use databend_common_pipeline_transforms::SpillBackoffSettings;
 use databend_common_settings::OutofMemoryBehavior;
 
 use crate::sessions::QueryContext;
@@ -154,6 +155,17 @@ impl MemorySettingsExt for MemorySettings {
             );
         }
 
+        let max_sleep_ms = settings.get_spill_global_backoff_max_sleep_ms()?;
+        let low_query_ratio = settings.get_spill_global_backoff_low_query_ratio()? as f64 / 100.0;
+        let low_query_memory_bytes = (max_memory_usage as f64 * low_query_ratio) as u64;
+
+        if max_sleep_ms > 0 && max_memory_usage != 0 {
+            builder = builder.with_spill_backoff(Some(SpillBackoffSettings {
+                max_sleep_ms,
+                low_query_memory_bytes,
+            }));
+        }
+
         Ok(builder.build())
     }
 }
@@ -163,6 +175,7 @@ mod tests {
     use databend_common_catalog::table_context::TableContext;
     use databend_common_exception::Result;
     use databend_common_pipeline_transforms::MemorySettings;
+    use databend_common_pipeline_transforms::SpillBackoffSettings;
 
     use crate::pipelines::memory_settings::MemorySettingsExt;
     use crate::test_kits::TestFixture;
@@ -384,4 +397,40 @@ mod tests {
         assert_eq!(memory_settings.spill_unit_size, 3 * 1024 * 1024);
         Ok(())
     }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_aggregate_spill_backoff_enabled_from_settings() -> Result<()> {
+        let fixture = TestFixture::setup().await?;
+        let ctx = fixture.new_query_ctx().await?;
+        let settings = ctx.get_settings();
+
+        settings.set_setting("max_memory_usage".into(), "1000".into())?;
+        settings.set_setting("spill_global_backoff_max_sleep_ms".into(), "100".into())?;
+        settings.set_setting("spill_global_backoff_low_query_ratio".into(), "25".into())?;
+
+        let memory_settings = MemorySettings::from_aggregate_settings(&ctx)?;
+
+        assert_eq!(
+            memory_settings.spill_backoff,
+            Some(SpillBackoffSettings {
+                max_sleep_ms: 100,
+                low_query_memory_bytes: 250,
+            })
+        );
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_aggregate_spill_backoff_disabled_when_sleep_is_zero() -> Result<()> {
+        let fixture = TestFixture::setup().await?;
+        let ctx = fixture.new_query_ctx().await?;
+        let settings = ctx.get_settings();
+
+        settings.set_setting("spill_global_backoff_max_sleep_ms".into(), "0".into())?;
+
+        let memory_settings = MemorySettings::from_aggregate_settings(&ctx)?;
+
+        assert_eq!(memory_settings.spill_backoff, None);
+        Ok(())
+    }
 }