fix(storage): decouple batch-delete size from max_threads (#20219)

* fix(storage): size delete batches for the capped concurrency

remove_file_in_batch sized each batch as locations.len() / max_threads
clamped to 1000. This was introduced in #15005 to split a deletion into
roughly max_threads chunks so they could saturate the delete concurrency
pool, which back then was also max_threads: one chunk per slot, one wave.

#16770 later lowered that concurrency to permit = (max_threads/2).clamp(1,3)
to mitigate S3 rate-limit errors, but left the chunking formula untouched.
With only ~3 slots, splitting into ~max_threads chunks no longer buys
parallelism, the extra chunks just queue. On a background warehouse
(max_threads~16) a 1000-file chunk became 16 requests of ~62 keys run 3 at
a time (~6 waves) instead of one 1000-key request.

A batch delete costs one request regardless of key count (S3 allows up to
1000), and measurements show per-request latency is dominated by fixed
overhead: ~1.0s for 62 keys vs ~1.0-1.4s for 1000. So smaller batches only
multiply request count and wall time. fuse_vacuum2 on a table with ~22k
block chunks hit the 96h execution limit at ~57% progress.

With concurrency fixed at <=3, the fastest choice is the largest batch
(fewest requests, fewest waves). Introduce a runtime-tunable setting
storage_delete_batch_size (default 1000, the S3 DeleteObjects limit) and use
it directly instead of dividing by max_threads. The permit cap from #16770
is left unchanged, so this reduces request count without raising
request-rate pressure.

Add a regression test asserting the batch size is independent of max_threads.

* fix(storage): clamp delete batch to backend delete_max_size

Backends with a batch-delete capability below storage_delete_batch_size
(Azure=256, GCS=100) would otherwise take the single-future branch for
chunks up to the requested size, letting opendal's Deleter flush
backend-sized sub-batches serially inside one delete_files future. That
bypasses execute_futures_in_parallel and loses the permit concurrency
those backends still need.

Cap the outer chunk size at the operator's delete_max_size capability so
each chunk maps to one backend request and cross-chunk parallelism is
preserved. S3 (limit 1000) is unaffected.

Add a regression test covering a small-limit backend.

Author: dantengsky

src/query/ee/tests/it/storages/fuse/operations/vacuum.rs

@@ -223,7 +223,9 @@ async fn test_do_vacuum_temporary_files() -> anyhow::Result<()> {
 
 mod test_accessor {
     use std::future::Future;
+    use std::sync::Mutex;
     use std::sync::atomic::AtomicBool;
+    use std::sync::atomic::AtomicUsize;
     use std::sync::atomic::Ordering;
 
     use opendal::raw::MaybeSend;
@@ -380,6 +382,89 @@ mod test_accessor {
             ))
         }
     }
+
+    // Accessor that records batch-delete behaviour, used to verify how
+    // `storage_delete_batch_size` interacts with the backend's `delete_max_size`.
+    //
+    // - `chunk_count` counts calls to `Access::delete()`, i.e. the number of outer
+    //   chunks `remove_file_in_batch` splits into (each becomes one `delete_files`
+    //   future). This reveals whether cross-chunk parallelism is preserved.
+    // - `flush_sizes` records the key count of each opendal flush.
+    #[derive(Debug)]
+    pub(crate) struct RecordingAccessor {
+        // `None` models a backend that advertises no batch-delete capability (like `fs`).
+        delete_max_size: Option<usize>,
+        chunk_count: Arc<AtomicUsize>,
+        flush_sizes: Arc<Mutex<Vec<usize>>>,
+    }
+
+    impl RecordingAccessor {
+        pub(crate) fn new(delete_max_size: usize) -> Self {
+            Self::with_capability(Some(delete_max_size))
+        }
+
+        pub(crate) fn with_capability(delete_max_size: Option<usize>) -> Self {
+            Self {
+                delete_max_size,
+                chunk_count: Arc::new(AtomicUsize::new(0)),
+                flush_sizes: Arc::new(Mutex::new(Vec::new())),
+            }
+        }
+
+        pub(crate) fn chunk_count(&self) -> usize {
+            self.chunk_count.load(Ordering::Acquire)
+        }
+
+        pub(crate) fn flush_sizes(&self) -> Arc<Mutex<Vec<usize>>> {
+            self.flush_sizes.clone()
+        }
+    }
+
+    pub struct RecordingDeleter {
+        size: usize,
+        flush_sizes: Arc<Mutex<Vec<usize>>>,
+    }
+
+    impl oio::Delete for RecordingDeleter {
+        fn delete(&mut self, _path: &str, _args: OpDelete) -> opendal::Result<()> {
+            self.size += 1;
+            Ok(())
+        }
+
+        async fn flush(&mut self) -> opendal::Result<usize> {
+            let n = self.size;
+            if n > 0 {
+                self.flush_sizes.lock().unwrap().push(n);
+            }
+            self.size = 0;
+            Ok(n)
+        }
+    }
+
+    impl Access for RecordingAccessor {
+        type Reader = ();
+        type Writer = ();
+        type Lister = ();
+        type Deleter = RecordingDeleter;
+
+        fn info(&self) -> Arc<AccessorInfo> {
+            let info = AccessorInfo::default();
+            info.set_native_capability(opendal::Capability {
+                delete: true,
+                delete_max_size: self.delete_max_size,
+                ..Default::default()
+            });
+            info.into()
+        }
+
+        async fn delete(&self) -> opendal::Result<(RpDelete, Self::Deleter)> {
+            self.chunk_count.fetch_add(1, Ordering::AcqRel);
+            Ok((RpDelete::default(), RecordingDeleter {
+                size: 0,
+                flush_sizes: self.flush_sizes.clone(),
+            }))
+        }
+    }
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -549,6 +634,118 @@ async fn test_remove_files_in_batch_do_not_swallow_errors() -> anyhow::Result<()
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_remove_files_in_batch_respects_delete_batch_size() -> anyhow::Result<()> {
+    // The batch size must come from `storage_delete_batch_size`, not from
+    // `max_threads` (the old `locations.len() / max_threads` behaviour). The
+    // setting is chosen below the backend delete limit so it, and not the
+    // backend cap, is the binding constraint.
+    let recording = Arc::new(test_accessor::RecordingAccessor::new(1000));
+    let flush_sizes = recording.flush_sizes();
+    let operator = OperatorBuilder::new(recording.clone()).finish();
+
+    let fixture = TestFixture::setup().await?;
+    let ctx = fixture.new_query_ctx().await?;
+
+    // With the old formula this many threads would shrink the batch to
+    // 500 / 32 ~= 15; the setting must override that.
+    ctx.get_settings().set_max_threads(32)?;
+    ctx.get_settings()
+        .set_setting("storage_delete_batch_size".to_string(), "200".to_string())?;
+
+    let file_util = Files::create(ctx, operator);
+
+    // 500 files with batch size 200 -> chunks of [200, 200, 100].
+    let files: Vec<String> = (0..500).map(|i| format!("f{i}")).collect();
+    file_util.remove_file_in_batch(&files).await?;
+
+    let mut sizes = flush_sizes.lock().unwrap().clone();
+    sizes.sort_unstable();
+    assert_eq!(
+        sizes,
+        vec![100, 200, 200],
+        "expected batches of the configured size regardless of max_threads, got {sizes:?}"
+    );
+    // 3 outer chunks -> parallelism preserved.
+    assert_eq!(recording.chunk_count(), 3);
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_remove_files_in_batch_clamps_to_backend_delete_limit() -> anyhow::Result<()> {
+    // For backends whose batch-delete capability is below `storage_delete_batch_size`
+    // (e.g. Azure=256, GCS=100), the outer chunk size must be clamped to the backend
+    // limit. Otherwise opendal would flush backend-sized sub-batches serially inside a
+    // single delete_files future, bypassing execute_futures_in_parallel and losing
+    // cross-chunk concurrency.
+    let recording = Arc::new(test_accessor::RecordingAccessor::new(100));
+    let flush_sizes = recording.flush_sizes();
+    let operator = OperatorBuilder::new(recording.clone()).finish();
+
+    let fixture = TestFixture::setup().await?;
+    let ctx = fixture.new_query_ctx().await?;
+
+    ctx.get_settings().set_max_threads(32)?;
+    // Request 1000, but the backend only supports 100 per request.
+    ctx.get_settings()
+        .set_setting("storage_delete_batch_size".to_string(), "1000".to_string())?;
+
+    let file_util = Files::create(ctx, operator);
+
+    // 250 files: clamped to 100 -> chunks of [100, 100, 50], i.e. 3 outer chunks
+    // that can run in parallel, instead of one 250-file future flushing 3 serial batches.
+    let files: Vec<String> = (0..250).map(|i| format!("f{i}")).collect();
+    file_util.remove_file_in_batch(&files).await?;
+
+    let mut sizes = flush_sizes.lock().unwrap().clone();
+    sizes.sort_unstable();
+    assert_eq!(
+        sizes,
+        vec![50, 100, 100],
+        "expected batches clamped to backend delete_max_size, got {sizes:?}"
+    );
+    // 3 outer chunks -> cross-chunk parallelism preserved for small-limit backends.
+    assert_eq!(recording.chunk_count(), 3);
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_remove_files_in_batch_no_batch_capability() -> anyhow::Result<()> {
+    // A backend advertising no batch-delete capability (`delete_max_size` unset, like `fs`)
+    // deletes one key per request. The chunk size must fall back to 1 so files are split into
+    // single-key tasks that run at `permit` concurrency, instead of collapsing into a single
+    // delete_files future that opendal would then flush one key at a time serially.
+    let recording = Arc::new(test_accessor::RecordingAccessor::with_capability(None));
+    let flush_sizes = recording.flush_sizes();
+    let operator = OperatorBuilder::new(recording.clone()).finish();
+
+    let fixture = TestFixture::setup().await?;
+    let ctx = fixture.new_query_ctx().await?;
+
+    ctx.get_settings().set_max_threads(32)?;
+    ctx.get_settings()
+        .set_setting("storage_delete_batch_size".to_string(), "1000".to_string())?;
+
+    let file_util = Files::create(ctx, operator);
+
+    // 5 files on a no-batch backend -> batch size 1 -> 5 single-key chunks, not one future.
+    let files: Vec<String> = (0..5).map(|i| format!("f{i}")).collect();
+    file_util.remove_file_in_batch(&files).await?;
+
+    let sizes = flush_sizes.lock().unwrap().clone();
+    assert_eq!(
+        sizes,
+        vec![1, 1, 1, 1, 1],
+        "expected single-key batches on a no-batch backend, got {sizes:?}"
+    );
+    // 5 outer chunks -> each key is an independent task eligible for parallel execution.
+    assert_eq!(recording.chunk_count(), 5);
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn test_vacuum_dropped_table_clean_autoincrement() -> anyhow::Result<()> {
     // 1. Prepare local meta service

src/query/settings/src/settings_default.rs

@@ -188,6 +188,13 @@ impl DefaultSettings {
                     scope: SettingScope::Both,
                     range: Some(SettingRange::Numeric(1..=3)),
                 }),
+                ("storage_delete_batch_size", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(1000),
+                    desc: "Sets the number of object keys deleted per batch delete request (e.g. S3 DeleteObjects). Larger values reduce request count; defaults to the S3 batch limit of 1000.",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Both,
+                    range: Some(SettingRange::Numeric(1..=1000)),
+                }),
                 ("max_memory_usage", DefaultSettingValue {
                     value: UserSettingValue::UInt64(max_memory_usage),
                     desc: "Sets the maximum memory usage in bytes for processing a single query.",

src/query/settings/src/settings_getter_setter.rs

@@ -233,6 +233,10 @@ impl Settings {
         self.try_get_u64("max_vacuum_threads")
     }
 
+    pub fn get_storage_delete_batch_size(&self) -> Result<u64> {
+        self.try_get_u64("storage_delete_batch_size")
+    }
+
     // Get storage_fetch_part_num.
     pub fn get_storage_fetch_part_num(&self) -> Result<u64> {
         match self.try_get_u64("storage_fetch_part_num")? {

src/query/storages/common/io/src/files.rs

@@ -79,10 +79,33 @@ impl Files {
             );
         }
 
-        // adjusts batch_size according to the `max_threads` settings,
-        // limits its min/max value to 1 and 1000.
+        // Number of object keys deleted per outer chunk. Kept independent of `max_threads`:
+        // a single batch-delete request (e.g. S3 DeleteObjects) costs one request regardless
+        // of key count, so a larger chunk means fewer requests, which is strictly better for
+        // both throughput and request-rate limits. Runtime-tunable via `storage_delete_batch_size`.
+        //
+        // The chunk size is capped by the backend's batch-delete capability
+        // (`delete_max_size`, e.g. S3 1000, Azure 256, GCS 100). Chunking above that limit
+        // would let opendal's `Deleter` flush backend-sized sub-batches *serially* inside a
+        // single `delete_files` future, bypassing `execute_futures_in_parallel` and losing
+        // the `permit` concurrency those backends still need. Capping at the backend limit
+        // keeps one request per chunk and preserves cross-chunk parallelism.
+        //
+        // Backends that advertise no batch-delete capability (`delete_max_size` unset, e.g.
+        // the `fs` service) delete one key per request; opendal's `Deleter` treats the missing
+        // capability as `max_size = 1`. Mirror that with `unwrap_or(1)` so such backends are
+        // chunked into single-key tasks that run at `permit` concurrency, rather than collapsing
+        // into one future that deletes serially.
         let threads_nums = self.ctx.get_settings().get_max_threads()? as usize;
-        let batch_size = (locations.len() / threads_nums).clamp(1, 1000);
+        let backend_delete_limit = self
+            .operator
+            .info()
+            .full_capability()
+            .delete_max_size
+            .unwrap_or(1)
+            .max(1);
+        let batch_size = (self.ctx.get_settings().get_storage_delete_batch_size()? as usize)
+            .clamp(1, backend_delete_limit);
 
         info!(
             "remove file in batch, batch_size: {}, number of chunks {}",