@@ -223,7 +223,9 @@ async fn test_do_vacuum_temporary_files() -> anyhow::Result<()> {
223223
224224mod test_accessor {
225225 use std:: future:: Future ;
226+ use std:: sync:: Mutex ;
226227 use std:: sync:: atomic:: AtomicBool ;
228+ use std:: sync:: atomic:: AtomicUsize ;
227229 use std:: sync:: atomic:: Ordering ;
228230
229231 use opendal:: raw:: MaybeSend ;
@@ -380,6 +382,89 @@ mod test_accessor {
380382 ) )
381383 }
382384 }
385+
386+ // Accessor that records batch-delete behaviour, used to verify how
387+ // `storage_delete_batch_size` interacts with the backend's `delete_max_size`.
388+ //
389+ // - `chunk_count` counts calls to `Access::delete()`, i.e. the number of outer
390+ // chunks `remove_file_in_batch` splits into (each becomes one `delete_files`
391+ // future). This reveals whether cross-chunk parallelism is preserved.
392+ // - `flush_sizes` records the key count of each opendal flush.
393+ #[ derive( Debug ) ]
394+ pub ( crate ) struct RecordingAccessor {
395+ // `None` models a backend that advertises no batch-delete capability (like `fs`).
396+ delete_max_size : Option < usize > ,
397+ chunk_count : Arc < AtomicUsize > ,
398+ flush_sizes : Arc < Mutex < Vec < usize > > > ,
399+ }
400+
401+ impl RecordingAccessor {
402+ pub ( crate ) fn new ( delete_max_size : usize ) -> Self {
403+ Self :: with_capability ( Some ( delete_max_size) )
404+ }
405+
406+ pub ( crate ) fn with_capability ( delete_max_size : Option < usize > ) -> Self {
407+ Self {
408+ delete_max_size,
409+ chunk_count : Arc :: new ( AtomicUsize :: new ( 0 ) ) ,
410+ flush_sizes : Arc :: new ( Mutex :: new ( Vec :: new ( ) ) ) ,
411+ }
412+ }
413+
414+ pub ( crate ) fn chunk_count ( & self ) -> usize {
415+ self . chunk_count . load ( Ordering :: Acquire )
416+ }
417+
418+ pub ( crate ) fn flush_sizes ( & self ) -> Arc < Mutex < Vec < usize > > > {
419+ self . flush_sizes . clone ( )
420+ }
421+ }
422+
423+ pub struct RecordingDeleter {
424+ size : usize ,
425+ flush_sizes : Arc < Mutex < Vec < usize > > > ,
426+ }
427+
428+ impl oio:: Delete for RecordingDeleter {
429+ fn delete ( & mut self , _path : & str , _args : OpDelete ) -> opendal:: Result < ( ) > {
430+ self . size += 1 ;
431+ Ok ( ( ) )
432+ }
433+
434+ async fn flush ( & mut self ) -> opendal:: Result < usize > {
435+ let n = self . size ;
436+ if n > 0 {
437+ self . flush_sizes . lock ( ) . unwrap ( ) . push ( n) ;
438+ }
439+ self . size = 0 ;
440+ Ok ( n)
441+ }
442+ }
443+
444+ impl Access for RecordingAccessor {
445+ type Reader = ( ) ;
446+ type Writer = ( ) ;
447+ type Lister = ( ) ;
448+ type Deleter = RecordingDeleter ;
449+
450+ fn info ( & self ) -> Arc < AccessorInfo > {
451+ let info = AccessorInfo :: default ( ) ;
452+ info. set_native_capability ( opendal:: Capability {
453+ delete : true ,
454+ delete_max_size : self . delete_max_size ,
455+ ..Default :: default ( )
456+ } ) ;
457+ info. into ( )
458+ }
459+
460+ async fn delete ( & self ) -> opendal:: Result < ( RpDelete , Self :: Deleter ) > {
461+ self . chunk_count . fetch_add ( 1 , Ordering :: AcqRel ) ;
462+ Ok ( ( RpDelete :: default ( ) , RecordingDeleter {
463+ size : 0 ,
464+ flush_sizes : self . flush_sizes . clone ( ) ,
465+ } ) )
466+ }
467+ }
383468}
384469
385470#[ tokio:: test( flavor = "multi_thread" ) ]
@@ -549,6 +634,118 @@ async fn test_remove_files_in_batch_do_not_swallow_errors() -> anyhow::Result<()
549634 Ok ( ( ) )
550635}
551636
637+ #[ tokio:: test( flavor = "multi_thread" ) ]
638+ async fn test_remove_files_in_batch_respects_delete_batch_size ( ) -> anyhow:: Result < ( ) > {
639+ // The batch size must come from `storage_delete_batch_size`, not from
640+ // `max_threads` (the old `locations.len() / max_threads` behaviour). The
641+ // setting is chosen below the backend delete limit so it, and not the
642+ // backend cap, is the binding constraint.
643+ let recording = Arc :: new ( test_accessor:: RecordingAccessor :: new ( 1000 ) ) ;
644+ let flush_sizes = recording. flush_sizes ( ) ;
645+ let operator = OperatorBuilder :: new ( recording. clone ( ) ) . finish ( ) ;
646+
647+ let fixture = TestFixture :: setup ( ) . await ?;
648+ let ctx = fixture. new_query_ctx ( ) . await ?;
649+
650+ // With the old formula this many threads would shrink the batch to
651+ // 500 / 32 ~= 15; the setting must override that.
652+ ctx. get_settings ( ) . set_max_threads ( 32 ) ?;
653+ ctx. get_settings ( )
654+ . set_setting ( "storage_delete_batch_size" . to_string ( ) , "200" . to_string ( ) ) ?;
655+
656+ let file_util = Files :: create ( ctx, operator) ;
657+
658+ // 500 files with batch size 200 -> chunks of [200, 200, 100].
659+ let files: Vec < String > = ( 0 ..500 ) . map ( |i| format ! ( "f{i}" ) ) . collect ( ) ;
660+ file_util. remove_file_in_batch ( & files) . await ?;
661+
662+ let mut sizes = flush_sizes. lock ( ) . unwrap ( ) . clone ( ) ;
663+ sizes. sort_unstable ( ) ;
664+ assert_eq ! (
665+ sizes,
666+ vec![ 100 , 200 , 200 ] ,
667+ "expected batches of the configured size regardless of max_threads, got {sizes:?}"
668+ ) ;
669+ // 3 outer chunks -> parallelism preserved.
670+ assert_eq ! ( recording. chunk_count( ) , 3 ) ;
671+
672+ Ok ( ( ) )
673+ }
674+
675+ #[ tokio:: test( flavor = "multi_thread" ) ]
676+ async fn test_remove_files_in_batch_clamps_to_backend_delete_limit ( ) -> anyhow:: Result < ( ) > {
677+ // For backends whose batch-delete capability is below `storage_delete_batch_size`
678+ // (e.g. Azure=256, GCS=100), the outer chunk size must be clamped to the backend
679+ // limit. Otherwise opendal would flush backend-sized sub-batches serially inside a
680+ // single delete_files future, bypassing execute_futures_in_parallel and losing
681+ // cross-chunk concurrency.
682+ let recording = Arc :: new ( test_accessor:: RecordingAccessor :: new ( 100 ) ) ;
683+ let flush_sizes = recording. flush_sizes ( ) ;
684+ let operator = OperatorBuilder :: new ( recording. clone ( ) ) . finish ( ) ;
685+
686+ let fixture = TestFixture :: setup ( ) . await ?;
687+ let ctx = fixture. new_query_ctx ( ) . await ?;
688+
689+ ctx. get_settings ( ) . set_max_threads ( 32 ) ?;
690+ // Request 1000, but the backend only supports 100 per request.
691+ ctx. get_settings ( )
692+ . set_setting ( "storage_delete_batch_size" . to_string ( ) , "1000" . to_string ( ) ) ?;
693+
694+ let file_util = Files :: create ( ctx, operator) ;
695+
696+ // 250 files: clamped to 100 -> chunks of [100, 100, 50], i.e. 3 outer chunks
697+ // that can run in parallel, instead of one 250-file future flushing 3 serial batches.
698+ let files: Vec < String > = ( 0 ..250 ) . map ( |i| format ! ( "f{i}" ) ) . collect ( ) ;
699+ file_util. remove_file_in_batch ( & files) . await ?;
700+
701+ let mut sizes = flush_sizes. lock ( ) . unwrap ( ) . clone ( ) ;
702+ sizes. sort_unstable ( ) ;
703+ assert_eq ! (
704+ sizes,
705+ vec![ 50 , 100 , 100 ] ,
706+ "expected batches clamped to backend delete_max_size, got {sizes:?}"
707+ ) ;
708+ // 3 outer chunks -> cross-chunk parallelism preserved for small-limit backends.
709+ assert_eq ! ( recording. chunk_count( ) , 3 ) ;
710+
711+ Ok ( ( ) )
712+ }
713+
714+ #[ tokio:: test( flavor = "multi_thread" ) ]
715+ async fn test_remove_files_in_batch_no_batch_capability ( ) -> anyhow:: Result < ( ) > {
716+ // A backend advertising no batch-delete capability (`delete_max_size` unset, like `fs`)
717+ // deletes one key per request. The chunk size must fall back to 1 so files are split into
718+ // single-key tasks that run at `permit` concurrency, instead of collapsing into a single
719+ // delete_files future that opendal would then flush one key at a time serially.
720+ let recording = Arc :: new ( test_accessor:: RecordingAccessor :: with_capability ( None ) ) ;
721+ let flush_sizes = recording. flush_sizes ( ) ;
722+ let operator = OperatorBuilder :: new ( recording. clone ( ) ) . finish ( ) ;
723+
724+ let fixture = TestFixture :: setup ( ) . await ?;
725+ let ctx = fixture. new_query_ctx ( ) . await ?;
726+
727+ ctx. get_settings ( ) . set_max_threads ( 32 ) ?;
728+ ctx. get_settings ( )
729+ . set_setting ( "storage_delete_batch_size" . to_string ( ) , "1000" . to_string ( ) ) ?;
730+
731+ let file_util = Files :: create ( ctx, operator) ;
732+
733+ // 5 files on a no-batch backend -> batch size 1 -> 5 single-key chunks, not one future.
734+ let files: Vec < String > = ( 0 ..5 ) . map ( |i| format ! ( "f{i}" ) ) . collect ( ) ;
735+ file_util. remove_file_in_batch ( & files) . await ?;
736+
737+ let sizes = flush_sizes. lock ( ) . unwrap ( ) . clone ( ) ;
738+ assert_eq ! (
739+ sizes,
740+ vec![ 1 , 1 , 1 , 1 , 1 ] ,
741+ "expected single-key batches on a no-batch backend, got {sizes:?}"
742+ ) ;
743+ // 5 outer chunks -> each key is an independent task eligible for parallel execution.
744+ assert_eq ! ( recording. chunk_count( ) , 5 ) ;
745+
746+ Ok ( ( ) )
747+ }
748+
552749#[ tokio:: test( flavor = "multi_thread" ) ]
553750async fn test_vacuum_dropped_table_clean_autoincrement ( ) -> anyhow:: Result < ( ) > {
554751 // 1. Prepare local meta service
0 commit comments