feat(config): improve spill configuration with structured (#18593)

* feat(config): improve spill configuration with structured

* improve the spill config

* add spiller config behavior unit test

* fix configs_table_basic.txt test

Author: BohuTANG

src/query/config/src/config.rs

@@ -783,6 +783,22 @@ pub struct FsStorageConfig {
         default_value = "_data"
     )]
     pub data_path: String,
+
+    /// Percentage of disk space to reserve (0.0-100.0) - for spill usage
+    #[clap(skip)]
+    #[serde(default = "FsStorageConfig::default_reserved_space_percentage")]
+    pub reserved_space_percentage: Option<OrderedFloat<f64>>,
+
+    /// Maximum bytes allowed for spill (0 = unlimited) - for spill usage  
+    #[clap(skip)]
+    #[serde(default)]
+    pub max_bytes: Option<u64>,
+}
+
+impl FsStorageConfig {
+    fn default_reserved_space_percentage() -> Option<OrderedFloat<f64>> {
+        None // Use None as default, will use system default (30.0) if not specified
+    }
 }
 
 impl Default for FsStorageConfig {
@@ -795,6 +811,8 @@ impl From<InnerStorageFsConfig> for FsStorageConfig {
     fn from(inner: InnerStorageFsConfig) -> Self {
         Self {
             data_path: inner.root,
+            reserved_space_percentage: None,
+            max_bytes: None,
         }
     }
 }
@@ -3533,6 +3551,17 @@ impl Default for DiskCacheConfig {
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Args)]
 #[serde(default)]
 pub struct SpillConfig {
+    /// Unified spill storage configuration
+    /// Auto-detects type based on storage.type field:
+    /// - "fs": Local filesystem spill
+    /// - "s3": S3 remote spill  
+    /// - "azblob": Azure blob remote spill
+    /// - etc.
+    ///   If not configured, uses main data storage with _spill prefix (default)
+    #[clap(skip)]
+    pub storage: Option<StorageConfig>,
+
+    // Legacy fields for backward compatibility
     /// Path of spill to local disk. disable if it's empty.
     #[clap(long, value_name = "VALUE", default_value = "")]
     pub spill_local_disk_path: String,
@@ -3544,15 +3573,47 @@ pub struct SpillConfig {
     #[clap(long, value_name = "VALUE", default_value = "18446744073709551615")]
     /// Allow space in bytes to spill to local disk.
     pub spill_local_disk_max_bytes: u64,
+}
 
-    // TODO: We need to fix StorageConfig so that it supports command line injections.
-    #[clap(skip)]
-    pub storage: Option<StorageConfig>,
+impl SpillConfig {
+    /// Get the spill type based on configuration
+    /// Auto-detects from storage configuration
+    pub fn get_spill_type(&self) -> String {
+        // Check new unified storage configuration first (higher priority)
+        if let Some(ref storage) = self.storage {
+            match storage.typ.as_str() {
+                "fs" => "fs".to_string(),
+                "s3" => "s3".to_string(),
+                "azblob" => "azblob".to_string(),
+                "gcs" => "gcs".to_string(),
+                "oss" => "oss".to_string(),
+                "obs" => "obs".to_string(),
+                "cos" => "cos".to_string(),
+                "webhdfs" => "webhdfs".to_string(),
+                _ => "remote".to_string(), // fallback for other storage types
+            }
+        } else if !self.spill_local_disk_path.is_empty() {
+            // Fall back to legacy configuration
+            "fs".to_string() // legacy local disk maps to fs
+        } else {
+            "default".to_string()
+        }
+    }
+
+    /// Check if this is using legacy local disk configuration
+    pub fn is_legacy_local_disk(&self) -> bool {
+        !self.spill_local_disk_path.is_empty()
+    }
 }
 
 impl Default for SpillConfig {
     fn default() -> Self {
-        inner::SpillConfig::default().into()
+        Self {
+            storage: None,
+            spill_local_disk_path: String::new(),
+            spill_local_disk_reserved_space_percentage: OrderedFloat(30.0),
+            spill_local_disk_max_bytes: u64::MAX,
+        }
     }
 }
 
@@ -3727,28 +3788,96 @@ mod cache_config_converters {
         spill: SpillConfig,
         cache: &DiskCacheConfig,
     ) -> Result<inner::SpillConfig> {
-        // Trick for cloud, perhaps we should introduce a new configuration for the local writeable root.
-        let local_writeable_root = if cache.path != DiskCacheConfig::default().path
-            && spill.spill_local_disk_path.is_empty()
-        {
-            Some(cache.path.clone())
-        } else {
-            None
-        };
-
-        let storage_params = spill
-            .storage
-            .map(|storage| {
-                let storage: InnerStorageConfig = storage.try_into()?;
-                Ok::<_, ErrorCode>(storage.params)
-            })
-            .transpose()?;
+        // 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) =
+            match spill_type.as_str() {
+                "fs" => {
+                    if let Some(ref storage) = spill.storage {
+                        // Use new filesystem storage configuration (higher priority)
+                        let fs_path = storage.fs.data_path.clone();
+                        let reserved_ratio = storage
+                            .fs
+                            .reserved_space_percentage
+                            .unwrap_or(OrderedFloat(30.0))
+                            / 100.0;
+                        let max_bytes = storage.fs.max_bytes.unwrap_or(u64::MAX);
+
+                        // Validate fs configuration
+                        if fs_path.is_empty() {
+                            return Err(ErrorCode::InvalidConfig(
+                                "FS storage configured but data_path is empty. Either specify a path or remove the fs storage configuration to use default behavior."
+                            ));
+                        } else {
+                            (None, fs_path, reserved_ratio, max_bytes, None)
+                        }
+                    } else if spill.is_legacy_local_disk() {
+                        // Fall back to legacy local disk configuration
+                        (
+                            None,
+                            spill.spill_local_disk_path,
+                            spill.spill_local_disk_reserved_space_percentage / 100.0,
+                            spill.spill_local_disk_max_bytes,
+                            None,
+                        )
+                    } else {
+                        return Err(ErrorCode::InvalidConfig(
+                            "FS storage configuration not found",
+                        ));
+                    }
+                }
+                "s3" | "azblob" | "gcs" | "oss" | "obs" | "cos" | "webhdfs" | "remote" => {
+                    // Use remote storage configuration for all remote storage types
+                    let storage_params = spill
+                        .storage
+                        .map(|storage| {
+                            let storage: InnerStorageConfig = storage.try_into()?;
+                            Ok::<_, ErrorCode>(storage.params)
+                        })
+                        .transpose()?;
+
+                    (
+                        None,
+                        String::new(),
+                        OrderedFloat(0.3),
+                        u64::MAX,
+                        storage_params,
+                    )
+                }
+                _ => {
+                    // 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
+                    };
+
+                    let storage_params = spill
+                        .storage
+                        .map(|storage| {
+                            let storage: InnerStorageConfig = storage.try_into()?;
+                            Ok::<_, ErrorCode>(storage.params)
+                        })
+                        .transpose()?;
+
+                    (
+                        local_writeable_root,
+                        spill.spill_local_disk_path,
+                        spill.spill_local_disk_reserved_space_percentage / 100.0,
+                        spill.spill_local_disk_max_bytes,
+                        storage_params,
+                    )
+                }
+            };
 
         Ok(inner::SpillConfig {
             local_writeable_root,
-            path: spill.spill_local_disk_path,
-            reserved_disk_ratio: spill.spill_local_disk_reserved_space_percentage / 100.0,
-            global_bytes_limit: spill.spill_local_disk_max_bytes,
+            path,
+            reserved_disk_ratio,
+            global_bytes_limit,
             storage_params,
         })
     }
@@ -3764,10 +3893,10 @@ mod cache_config_converters {
             });
 
             Self {
+                storage,
                 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,
-                storage,
             }
         }
     }

src/query/config/src/mask.rs

@@ -211,17 +211,17 @@ impl StorageConfig {
 impl SpillConfig {
     fn mask_display(&self) -> Self {
         let Self {
+            ref storage,
             ref spill_local_disk_path,
             spill_local_disk_reserved_space_percentage,
             spill_local_disk_max_bytes,
-            ref storage,
         } = *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,
-            storage: storage.as_ref().map(|storage| storage.mask_display()),
         }
     }
 }