refactor(query): simplify SpillsBufferPool with owned runtime and configurable settings (#19781)

* refactor(query): simplify SpillsBufferPool with owned runtime and configurable settings

- SpillsBufferPool now owns a dedicated Runtime instead of borrowing GlobalIORuntime
- Remove SpillTarget from public buffer pool APIs; callers no longer derive it
- Add spill_buffer_pool_memory and spill_buffer_pool_workers to SpillConfig
- Track buffer pool blocking time via atomic counters for observability
- Simplify BufferWriter by removing SpillTarget and redundant comments
- Clean up tests to use multi_thread tokio flavor

* z

* z

* fix(query): prevent hang in SpillsBufferPool by spawning Fetch and CreateWriter ops

Background workers were directly awaiting Fetch and CreateWriter ops in
their event loop, blocking the worker thread for the duration of the I/O.
With only 2 workers, concurrent Fetch/CreateWriter ops could starve
reader_task_loop tasks (spawned via tokio::spawn onto the same runtime),
causing recv_blocking() in SpillsDataReader::read() to hang indefinitely.

Fix: spawn Fetch and CreateWriter as independent tasks, consistent with
WriterTask and ReaderTask, so workers remain free to dequeue new ops.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* z

* z

* z

* z

* z

* z

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: zhang2014

src/query/config/src/config.rs

@@ -3481,6 +3481,14 @@ pub struct SpillConfig {
     /// 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,
+
+    /// Total memory for the spill buffer pool in bytes.
+    #[clap(long, value_name = "VALUE", default_value = "209715200")]
+    pub spill_buffer_pool_memory: u64,
+
+    /// Number of worker tasks in the spill buffer pool.
+    #[clap(long, value_name = "VALUE", default_value = "2")]
+    pub spill_buffer_pool_workers: usize,
 }
 
 impl Default for SpillConfig {
@@ -3594,6 +3602,8 @@ mod config_converters {
             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,
+            buffer_pool_memory: spill.spill_buffer_pool_memory,
+            buffer_pool_workers: spill.spill_buffer_pool_workers,
         })
     }
 
@@ -3616,6 +3626,8 @@ mod config_converters {
                 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,
+                spill_buffer_pool_memory: value.buffer_pool_memory,
+                spill_buffer_pool_workers: value.buffer_pool_workers,
             }
         }
     }

src/query/config/src/inner.rs

@@ -433,6 +433,12 @@ pub struct SpillConfig {
     /// 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,
+
+    /// Total memory for the spill buffer pool in bytes.
+    pub buffer_pool_memory: u64,
+
+    /// Number of worker tasks in the spill buffer pool.
+    pub buffer_pool_workers: usize,
 }
 
 impl SpillConfig {
@@ -503,6 +509,8 @@ impl SpillConfig {
             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,
+            buffer_pool_memory: 200 * 1024 * 1024,
+            buffer_pool_workers: 2,
         }
     }
 }
@@ -519,6 +527,8 @@ impl Default for SpillConfig {
             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,
+            buffer_pool_memory: 200 * 1024 * 1024,
+            buffer_pool_workers: 2,
         }
     }
 }

src/query/config/src/mask.rs

@@ -218,6 +218,8 @@ impl SpillConfig {
             sort_spilling_disk_quota_ratio,
             window_partition_spilling_disk_quota_ratio,
             result_set_spilling_disk_quota_ratio,
+            spill_buffer_pool_memory,
+            spill_buffer_pool_workers,
         } = *self;
 
         Self {
@@ -227,8 +229,9 @@ impl SpillConfig {
             storage: storage.as_ref().map(|storage| storage.mask_display()),
             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,
+            spill_buffer_pool_memory,
+            spill_buffer_pool_workers,
         }
     }
 }
@@ -384,8 +387,9 @@ mod tests {
             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,
+            spill_buffer_pool_memory: 209715200,
+            spill_buffer_pool_workers: 2,
             storage: Some(StorageConfig {
                 typ: "s3".to_string(),
                 s3: S3StorageConfig {

src/query/service/src/global_services.rs

@@ -115,7 +115,7 @@ impl GlobalServices {
         // 4. cluster discovery init.
         ClusterDiscovery::init(config, version).await?;
 
-        SpillsBufferPool::init();
+        SpillsBufferPool::init(&config.spill)?;
         // TODO(xuanwo):
         //
         // This part is a bit complex because catalog are used widely in different

src/query/service/src/pipelines/processors/transforms/aggregator/new_aggregate/new_aggregate_spiller.rs

@@ -54,11 +54,10 @@ struct PayloadWriter {
 impl PayloadWriter {
     fn try_create(prefix: &str) -> Result<Self> {
         let data_operator = DataOperator::instance();
-        let target = SpillTarget::from_storage_params(data_operator.spill_params());
         let operator = data_operator.spill_operator();
         let buffer_pool = SpillsBufferPool::instance();
         let file_path = format!("{}/{}", prefix, GlobalUniq::unique());
-        let spills_data_writer = buffer_pool.writer(operator, file_path.clone(), target)?;
+        let spills_data_writer = buffer_pool.writer(operator, file_path.clone())?;
 
         Ok(PayloadWriter {
             path: file_path,
@@ -414,10 +413,11 @@ fn restore_payload(
         data_schema.clone(),
         vec![row_group.clone()],
         target,
+        read_setting,
     )?;
 
     let instant = Instant::now();
-    let data_block = reader.read(read_setting)?;
+    let data_block = reader.read()?;
     let elapsed = instant.elapsed();
 
     let read_size = reader.read_bytes();

src/query/service/src/pipelines/processors/transforms/new_hash_join/grace/grace_join.rs

@@ -278,9 +278,10 @@ impl<T: GraceMemoryJoin> GraceHashJoin<T> {
                 self.desc.build_schema.clone(),
                 data.row_groups,
                 target,
+                self.read_settings,
             )?;
 
-            while let Some(data_block) = reader.read(self.read_settings)? {
+            while let Some(data_block) = reader.read()? {
                 self.memory_hash_join.add_block(Some(data_block))?;
             }
         }
@@ -417,12 +418,11 @@ pub struct GraceJoinPartition {
 impl GraceJoinPartition {
     pub fn create(prefix: &str) -> Result<GraceJoinPartition> {
         let data_operator = DataOperator::instance();
-        let target = SpillTarget::from_storage_params(data_operator.spill_params());
 
         let operator = data_operator.spill_operator();
         let buffer_pool = SpillsBufferPool::instance();
         let file_path = format!("{}/{}", prefix, GlobalUniq::unique());
-        let spills_data_writer = buffer_pool.writer(operator, file_path.clone(), target)?;
+        let spills_data_writer = buffer_pool.writer(operator, file_path.clone())?;
 
         Ok(GraceJoinPartition {
             path: file_path,
@@ -500,6 +500,7 @@ impl<'a, T: GraceMemoryJoin> RestoreProbeStream<'a, T> {
                         self.join.desc.probe_schema.clone(),
                         data.row_groups,
                         target,
+                        self.join.read_settings,
                     )?;
                     self.spills_reader = Some(reader);
                     break;
@@ -511,7 +512,7 @@ impl<'a, T: GraceMemoryJoin> RestoreProbeStream<'a, T> {
             }
 
             if let Some(mut spills_reader) = self.spills_reader.take() {
-                if let Some(v) = spills_reader.read(self.join.read_settings)? {
+                if let Some(v) = spills_reader.read()? {
                     self.spills_reader = Some(spills_reader);
                     return Ok(Some(v));
                 }