feat: spill profile metrics

scripts/ci/deploy/config/databend-query-node-1.toml

@@ -162,3 +162,6 @@ max_bytes = 21474836480
 
 [spill]
 spill_local_disk_path = "./.databend/temp/_query_spill"
+# Cap local spill to 5GB so window spills keep ~1GB quota with default 20% ratio.
+spill_local_disk_max_bytes = 1073741824
+window_partition_spilling_disk_quota_ratio = 20

scripts/databend_test_helper/databend_test_helper/configs/databend-query-node2.toml

@@ -62,4 +62,4 @@ path = "_databend_data/cache/query2"
 max_bytes = 1073741824  # 1GB
 
 [spill]
-spill_local_disk_path = "_databend_data/spill/query2"
+spill_local_disk_path = "_databend_data/spill/query2"

scripts/databend_test_helper/databend_test_helper/configs/databend-query-node3.toml

@@ -62,4 +62,4 @@ path = "_databend_data/cache/query3"
 max_bytes = 1073741824  # 1GB
 
 [spill]
-spill_local_disk_path = "_databend_data/spill/query3"
+spill_local_disk_path = "_databend_data/spill/query3"

scripts/test-bend-tests/configs/query/query-2.toml

@@ -63,4 +63,4 @@ path = "_databend_data/cache/query2"
 max_bytes = 1073741824  # 1GB
 
 [spill]
-spill_local_disk_path = "_databend_data/spill/query2"
+spill_local_disk_path = "_databend_data/spill/query2"

scripts/test-bend-tests/configs/query/query-3.toml

@@ -63,4 +63,4 @@ path = "_databend_data/cache/query3"
 max_bytes = 1073741824  # 1GB
 
 [spill]
-spill_local_disk_path = "_databend_data/spill/query3"
+spill_local_disk_path = "_databend_data/spill/query3"

scripts/test-bend-tests/configs/query/query-4.toml

@@ -63,4 +63,4 @@ path = "_databend_data/cache/query4"
 max_bytes = 1073741824  # 1GB
 
 [spill]
-spill_local_disk_path = "_databend_data/spill/query4"
+spill_local_disk_path = "_databend_data/spill/query4"

scripts/test-bend-tests/configs/query/query-5.toml

@@ -63,4 +63,4 @@ path = "_databend_data/cache/query5"
 max_bytes = 1073741824  # 1GB
 
 [spill]
-spill_local_disk_path = "_databend_data/spill/query5"
+spill_local_disk_path = "_databend_data/spill/query5"

src/query/config/src/config.rs

@@ -798,7 +798,7 @@ pub struct FsStorageConfig {
 
 impl FsStorageConfig {
     fn default_reserved_space_percentage() -> Option<OrderedFloat<f64>> {
-        None // Use None as default, will use system default (30.0) if not specified
+        None // Use None as default, will use system default (10.0) if not specified
     }
 }
 
@@ -3585,13 +3585,32 @@ pub struct SpillConfig {
     #[clap(long, value_name = "VALUE", default_value = "")]
     pub spill_local_disk_path: String,
 
-    #[clap(long, value_name = "VALUE", default_value = "30")]
+    #[clap(long, value_name = "VALUE", default_value = "10")]
     /// Percentage of reserve disk space that won't be used for spill to local disk.
     pub spill_local_disk_reserved_space_percentage: OrderedFloat<f64>,
 
     #[clap(long, value_name = "VALUE", default_value = "18446744073709551615")]
     /// Allow space in bytes to spill to local disk.
     pub spill_local_disk_max_bytes: u64,
+
+    /// Maximum percentage of the global local spill quota that a single sort
+    /// operator may use for one query.
+    ///
+    /// Value range: 0-100. Effective only when local spill is enabled (there is
+    /// a valid local spill path and non-zero `spill_local_disk_max_bytes`).
+    #[clap(long, value_name = "PERCENT", default_value = "60")]
+    pub sort_spilling_disk_quota_ratio: u64,
+
+    /// Maximum percentage of the global local spill quota that window
+    /// partitioners may use for one query.
+    #[clap(long, value_name = "PERCENT", default_value = "60")]
+    pub window_partition_spilling_disk_quota_ratio: u64,
+
+    /// Maximum percentage of the global local spill quota that HTTP
+    /// result-set spilling may use for one query.
+    /// TODO: keep 0 to avoid deleting local result-set spill dir before HTTP pagination finishes.
+    #[clap(long, value_name = "PERCENT", default_value = "0")]
+    pub result_set_spilling_disk_quota_ratio: u64,
 }
 
 impl SpillConfig {
@@ -3630,8 +3649,12 @@ impl Default for SpillConfig {
         Self {
             storage: None,
             spill_local_disk_path: String::new(),
-            spill_local_disk_reserved_space_percentage: OrderedFloat(30.0),
+            spill_local_disk_reserved_space_percentage: OrderedFloat(10.0),
             spill_local_disk_max_bytes: u64::MAX,
+            sort_spilling_disk_quota_ratio: 60,
+            window_partition_spilling_disk_quota_ratio: 60,
+            // TODO: keep 0 to avoid deleting local result-set spill dir before HTTP pagination finishes.
+            result_set_spilling_disk_quota_ratio: 0,
         }
     }
 }
@@ -3706,7 +3729,7 @@ mod cache_config_converters {
                 catalogs.insert(CATALOG_HIVE.to_string(), catalog);
             }
 
-            let spill = convert_local_spill_config(spill, &cache.disk_cache_config)?;
+            let spill = convert_local_spill_config(spill)?;
 
             Ok(InnerConfig {
                 query: query.try_into()?,
@@ -3803,10 +3826,7 @@ mod cache_config_converters {
         }
     }
 
-    fn convert_local_spill_config(
-        spill: SpillConfig,
-        cache: &DiskCacheConfig,
-    ) -> Result<inner::SpillConfig> {
+    fn convert_local_spill_config(spill: SpillConfig) -> Result<inner::SpillConfig> {
         // Determine configuration based on auto-detected spill type
         let spill_type = spill.get_spill_type();
         let (local_writeable_root, path, reserved_disk_ratio, global_bytes_limit, storage_params) =
@@ -3818,7 +3838,7 @@ mod cache_config_converters {
                         let reserved_ratio = storage
                             .fs
                             .reserved_space_percentage
-                            .unwrap_or(OrderedFloat(30.0))
+                            .unwrap_or(OrderedFloat(10.0))
                             / 100.0;
                         let max_bytes = storage.fs.max_bytes.unwrap_or(u64::MAX);
 
@@ -3864,16 +3884,11 @@ mod cache_config_converters {
                     )
                 }
                 _ => {
-                    // Default behavior for "default" type and any unrecognized types
-                    // Default behavior with backward compatibility
-                    let local_writeable_root = if cache.path != DiskCacheConfig::default().path
-                        && spill.spill_local_disk_path.is_empty()
-                    {
-                        Some(cache.path.clone())
-                    } else {
-                        None
-                    };
-
+                    // Default behavior for "default" type and any unrecognized types:
+                    // do NOT implicitly reuse the data cache disk. Local spill is
+                    // enabled only when explicitly configured via either
+                    //   - [spill.storage] with type = "fs", or
+                    //   - legacy spill_local_disk_path.
                     let storage_params = spill
                         .storage
                         .map(|storage| {
@@ -3883,7 +3898,7 @@ mod cache_config_converters {
                         .transpose()?;
 
                     (
-                        local_writeable_root,
+                        None,
                         spill.spill_local_disk_path,
                         spill.spill_local_disk_reserved_space_percentage / 100.0,
                         spill.spill_local_disk_max_bytes,
@@ -3898,6 +3913,10 @@ mod cache_config_converters {
             reserved_disk_ratio,
             global_bytes_limit,
             storage_params,
+            sort_spilling_disk_quota_ratio: spill.sort_spilling_disk_quota_ratio,
+            window_partition_spilling_disk_quota_ratio: spill
+                .window_partition_spilling_disk_quota_ratio,
+            result_set_spilling_disk_quota_ratio: spill.result_set_spilling_disk_quota_ratio,
         })
     }
 
@@ -3916,6 +3935,10 @@ mod cache_config_converters {
                 spill_local_disk_path: value.path,
                 spill_local_disk_reserved_space_percentage: value.reserved_disk_ratio * 100.0,
                 spill_local_disk_max_bytes: value.global_bytes_limit,
+                sort_spilling_disk_quota_ratio: value.sort_spilling_disk_quota_ratio,
+                window_partition_spilling_disk_quota_ratio: value
+                    .window_partition_spilling_disk_quota_ratio,
+                result_set_spilling_disk_quota_ratio: value.result_set_spilling_disk_quota_ratio,
             }
         }
     }

src/query/config/src/inner.rs

@@ -787,6 +787,22 @@ pub struct SpillConfig {
     pub global_bytes_limit: u64,
 
     pub storage_params: Option<StorageParams>,
+
+    /// Maximum percentage of the global local spill quota that a single
+    /// sort operator may use for one query.
+    ///
+    /// Value range: 0-100. Effective only when local spill is enabled
+    /// (i.e. there is a valid local spill path and non-zero global
+    /// bytes limit).
+    pub sort_spilling_disk_quota_ratio: u64,
+
+    /// Maximum percentage of the global local spill quota that window
+    /// partitioners may use for one query.
+    pub window_partition_spilling_disk_quota_ratio: u64,
+
+    /// Maximum percentage of the global local spill quota that HTTP
+    /// result-set spilling may use for one query.
+    pub result_set_spilling_disk_quota_ratio: u64,
 }
 
 impl SpillConfig {
@@ -807,13 +823,56 @@ impl SpillConfig {
         None
     }
 
+    /// Helper to compute a per-query local spill quota (in bytes) from a
+    /// percentage of the global local spill limit.
+    ///
+    /// - If local spill is disabled (no local path or zero global
+    ///   limit), returns 0.
+    /// - `ratio` is clamped into [0, 100].
+    pub fn quota_bytes_from_ratio(&self, ratio: u64) -> usize {
+        // Only effective when local spill is enabled.
+        if self.local_path().is_none() {
+            return 0;
+        }
+
+        let ratio = std::cmp::min(ratio, 100);
+        if ratio == 0 {
+            return 0;
+        }
+
+        let bytes = self.global_bytes_limit.saturating_mul(ratio) / 100;
+
+        // TempDirManager works with `usize` limits.
+        std::cmp::min(bytes, usize::MAX as u64) as usize
+    }
+
+    /// Per-query quota for sort operators.
+    pub fn sort_spill_bytes_limit(&self) -> usize {
+        self.quota_bytes_from_ratio(self.sort_spilling_disk_quota_ratio)
+    }
+
+    /// Per-query quota for window partitioners.
+    pub fn window_partition_spill_bytes_limit(&self) -> usize {
+        self.quota_bytes_from_ratio(self.window_partition_spilling_disk_quota_ratio)
+    }
+
+    /// Per-query quota for HTTP result-set spilling.
+    pub fn result_set_spill_bytes_limit(&self) -> usize {
+        self.quota_bytes_from_ratio(self.result_set_spilling_disk_quota_ratio)
+    }
+
     pub fn new_for_test(path: String, reserved_disk_ratio: f64, global_bytes_limit: u64) -> Self {
         Self {
             local_writeable_root: None,
             path,
             reserved_disk_ratio: OrderedFloat(reserved_disk_ratio),
             global_bytes_limit,
             storage_params: None,
+            // Use the same defaults as the external config.
+            sort_spilling_disk_quota_ratio: 60,
+            window_partition_spilling_disk_quota_ratio: 60,
+            // TODO: keep 0 to avoid deleting local result-set spill dir before HTTP pagination finishes.
+            result_set_spilling_disk_quota_ratio: 0,
         }
     }
 }
@@ -823,9 +882,13 @@ impl Default for SpillConfig {
         Self {
             local_writeable_root: None,
             path: "".to_string(),
-            reserved_disk_ratio: OrderedFloat(0.3),
+            reserved_disk_ratio: OrderedFloat(0.1),
             global_bytes_limit: u64::MAX,
             storage_params: None,
+            sort_spilling_disk_quota_ratio: 60,
+            window_partition_spilling_disk_quota_ratio: 60,
+            // TODO: keep 0 to avoid deleting local result-set spill dir before HTTP pagination finishes.
+            result_set_spilling_disk_quota_ratio: 0,
         }
     }
 }

src/query/config/src/mask.rs

@@ -215,13 +215,20 @@ impl SpillConfig {
             ref spill_local_disk_path,
             spill_local_disk_reserved_space_percentage,
             spill_local_disk_max_bytes,
+            sort_spilling_disk_quota_ratio,
+            window_partition_spilling_disk_quota_ratio,
+            result_set_spilling_disk_quota_ratio,
         } = *self;
 
         Self {
             storage: storage.as_ref().map(|storage| storage.mask_display()),
             spill_local_disk_path: spill_local_disk_path.clone(),
             spill_local_disk_reserved_space_percentage,
             spill_local_disk_max_bytes,
+            sort_spilling_disk_quota_ratio,
+            window_partition_spilling_disk_quota_ratio,
+            // TODO: keep 0 to avoid deleting local result-set spill dir before HTTP pagination finishes.
+            result_set_spilling_disk_quota_ratio,
         }
     }
 }
@@ -375,6 +382,10 @@ mod tests {
             spill_local_disk_path: "".to_string(),
             spill_local_disk_reserved_space_percentage: 30.0.into(),
             spill_local_disk_max_bytes: 10,
+            sort_spilling_disk_quota_ratio: 60,
+            window_partition_spilling_disk_quota_ratio: 30,
+            // TODO: keep 0 to avoid deleting local result-set spill dir before HTTP pagination finishes.
+            result_set_spilling_disk_quota_ratio: 0,
             storage: Some(StorageConfig {
                 typ: "s3".to_string(),
                 s3: S3StorageConfig {

src/query/service/src/physical_plans/physical_recluster.rs

@@ -21,6 +21,7 @@ use databend_common_catalog::plan::DataSourcePlan;
 use databend_common_catalog::plan::ReclusterTask;
 use databend_common_catalog::table::Table;
 use databend_common_catalog::table_context::TableContext;
+use databend_common_config::GlobalConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::DataSchemaRef;
@@ -319,9 +320,12 @@ impl IPhysicalPlan for HilbertPartition {
         )?;
 
         let settings = builder.settings.clone();
-        let disk_bytes_limit = settings.get_window_partition_spilling_to_disk_bytes_limit()?;
         let temp_dir_manager = TempDirManager::instance();
 
+        let disk_bytes_limit = GlobalConfig::instance()
+            .spill
+            .window_partition_spill_bytes_limit();
+
         let enable_dio = settings.get_enable_dio()?;
         let disk_spill = temp_dir_manager
             .get_disk_spill_dir(disk_bytes_limit, &builder.ctx.get_id())

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

@@ -18,6 +18,7 @@ use std::sync::atomic::AtomicUsize;
 
 use databend_common_catalog::plan::DataSourcePlan;
 use databend_common_catalog::table_context::TableContext;
+use databend_common_config::GlobalConfig;
 use databend_common_exception::Result;
 use databend_common_expression::SortColumnDescription;
 use databend_common_pipeline::core::ProcessorPtr;
@@ -148,7 +149,9 @@ impl IPhysicalPlan for WindowPartition {
         }
 
         let temp_dir_manager = TempDirManager::instance();
-        let disk_bytes_limit = settings.get_window_partition_spilling_to_disk_bytes_limit()?;
+        let disk_bytes_limit = GlobalConfig::instance()
+            .spill
+            .window_partition_spill_bytes_limit();
         let enable_dio = settings.get_enable_dio()?;
         let disk_spill = temp_dir_manager
             .get_disk_spill_dir(disk_bytes_limit, &builder.ctx.get_id())

src/query/service/src/pipelines/builders/builder_sort.rs

@@ -14,6 +14,7 @@
 
 use std::sync::Arc;
 
+use databend_common_config::GlobalConfig;
 use databend_common_exception::Result;
 use databend_common_expression::DataSchemaRef;
 use databend_common_expression::LimitType;
@@ -132,7 +133,7 @@ impl SortPipelineBuilder {
 
         let spiller = {
             let temp_dir_manager = TempDirManager::instance();
-            let disk_bytes_limit = settings.get_sort_spilling_to_disk_bytes_limit()?;
+            let disk_bytes_limit = GlobalConfig::instance().spill.sort_spill_bytes_limit();
             let enable_dio = settings.get_enable_dio()?;
             let disk_spill = temp_dir_manager
                 .get_disk_spill_dir(disk_bytes_limit, &self.ctx.get_id())