settings

Author: forsaken628

src/query/service/src/physical_plans/physical_window_partition.rs

@@ -161,6 +161,7 @@ impl IPhysicalPlan for WindowPartition {
             _ => unimplemented!(),
         };
         let window_spill_settings = MemorySettings::from_window_settings(&builder.ctx)?;
+        let enable_backpressure_spiller = settings.get_enable_backpressure_spiller()?;
 
         let processor_id = AtomicUsize::new(0);
         builder.main_pipeline.add_transform(|input, output| {
@@ -181,7 +182,7 @@ impl IPhysicalPlan for WindowPartition {
                     num_partitions,
                     window_spill_settings.clone(),
                     disk_spill.clone(),
-                    true,
+                    enable_backpressure_spiller,
                     strategy,
                 )?,
             )))

src/query/service/src/pipelines/processors/transforms/window/partition/window_partition_buffer_v2.rs

@@ -107,10 +107,10 @@ where
 {
     fn default() -> Self {
         Self {
-            state: PartitionSpillState::Empty,
-            spilled_ordinals: Vec::new(),
-            buffered_blocks: Vec::new(),
-            buffered_size: 0,
+            state: Default::default(),
+            spilled_ordinals: Default::default(),
+            buffered_blocks: Default::default(),
+            buffered_size: Default::default(),
         }
     }
 }

src/query/settings/src/settings_default.rs

@@ -228,6 +228,13 @@ impl DefaultSettings {
                     scope: SettingScope::Both,
                     range: Some(SettingRange::Numeric(0..=500)),
                 }),
+                ("enable_backpressure_spiller", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(1),
+                    desc: "Use new backpressure spiller.",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Both,
+                    range: Some(SettingRange::Numeric(0..=1)),
+                }),
                 ("max_spill_io_requests", DefaultSettingValue {
                     value: UserSettingValue::UInt64(default_max_spill_io_requests),
                     desc: "Sets the maximum number of concurrent spill I/O requests.",

src/query/settings/src/settings_getter_setter.rs

@@ -897,6 +897,10 @@ impl Settings {
         self.try_get_u64("dynamic_sample_time_budget_ms")
     }
 
+    pub fn get_enable_backpressure_spiller(&self) -> Result<bool> {
+        Ok(self.try_get_u64("enable_backpressure_spiller")? != 0)
+    }
+
     pub fn get_max_spill_io_requests(&self) -> Result<u64> {
         self.try_get_u64("max_spill_io_requests")
     }