fix(storage): deduplicate file paths in remove_file_in_batch to prevent opendal hang (#19939)

When the file list passed to remove_file_in_batch contains duplicate paths
(e.g. blocks with and without the vacuum2 'h' object-key prefix, or certain
object storage implementations returning duplicates during listing
pagination), opendal's Operator::delete_iter enters an infinite loop in
Deleter::close(): the internal BatchDeleter uses a HashSet (deduplicating
to N unique entries) while the outer Deleter tracks cur_size by call count.
After flush returns N < cur_size, the buffer is empty but cur_size never
reaches zero, spinning forever.

Fix: deduplicate locations inside remove_file_in_batch before chunking and
deletion. Log a warning with duplicate count and sample paths when
duplicates are detected, to aid future investigation.

Author: dantengsky

src/query/ee/src/storages/fuse/operations/vacuum_table_v2.rs

@@ -220,6 +220,7 @@ pub async fn do_vacuum2(
         .chain(blocks_to_gc.into_iter())
         .chain(stats_to_gc.into_iter())
         .collect();
+
     let op = Files::create(ctx.clone(), fuse_table.get_operator());
 
     // order is important

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

@@ -20,6 +20,7 @@ use databend_enterprise_query::test_kits::context::EESetup;
 use databend_query::sessions::QueryContext;
 use databend_query::sessions::TableContextTableAccess;
 use databend_query::test_kits::TestFixture;
+use databend_storages_common_io::dedup_file_locations;
 
 // TODO investigate this
 // NOTE: SHOULD specify flavor = "multi_thread", otherwise query execution might be hanged
@@ -116,3 +117,47 @@ async fn test_vacuum2_all() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+/// Verifies that dedup_file_locations correctly removes duplicates and reports samples.
+#[test]
+fn test_dedup_file_locations() {
+    // Simulate the vacuum2 scenario: bloom index paths generated from blocks
+    // with and without the 'h' prefix map to the same location.
+    let mut locations = vec![
+        "548052/604310/_i_b_v2/019bdabd0292702a96f51f3b3ea64335_v4.parquet".to_string(),
+        "548052/604310/_i_b_v2/019bdabd02927061af2ed18c2562b79e_v4.parquet".to_string(),
+        "548052/604310/_i_b_v2/019c0df29e8a7016a6a1be466e094ec3_v4.parquet".to_string(),
+        // Duplicates (same paths appearing again)
+        "548052/604310/_i_b_v2/019bdabd0292702a96f51f3b3ea64335_v4.parquet".to_string(),
+        "548052/604310/_i_b_v2/019bdabd02927061af2ed18c2562b79e_v4.parquet".to_string(),
+    ];
+
+    let (duplicates, samples) = dedup_file_locations(&mut locations);
+
+    assert_eq!(duplicates, 2);
+    assert_eq!(locations.len(), 3);
+    assert_eq!(samples.len(), 2);
+    assert_eq!(
+        samples[0],
+        "548052/604310/_i_b_v2/019bdabd0292702a96f51f3b3ea64335_v4.parquet"
+    );
+    assert_eq!(
+        samples[1],
+        "548052/604310/_i_b_v2/019bdabd02927061af2ed18c2562b79e_v4.parquet"
+    );
+}
+
+#[test]
+fn test_dedup_file_locations_no_duplicates() {
+    let mut locations = vec![
+        "a/b/file1.parquet".to_string(),
+        "a/b/file2.parquet".to_string(),
+        "a/b/file3.parquet".to_string(),
+    ];
+
+    let (duplicates, samples) = dedup_file_locations(&mut locations);
+
+    assert_eq!(duplicates, 0);
+    assert_eq!(locations.len(), 3);
+    assert!(samples.is_empty());
+}

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

@@ -21,6 +21,21 @@ use databend_common_exception::Result;
 use log::info;
 use opendal::Operator;
 
+/// Deduplicate a list of file locations in place.
+/// Returns (number_of_duplicates_removed, up_to_5_sample_duplicate_paths).
+pub fn dedup_file_locations(locations: &mut Vec<String>) -> (usize, Vec<String>) {
+    let len_before = locations.len();
+    locations.sort_unstable();
+    let dup_samples: Vec<String> = locations
+        .windows(2)
+        .filter(|w| w[0] == w[1])
+        .map(|w| w[0].clone())
+        .take(5)
+        .collect();
+    locations.dedup();
+    (len_before - locations.len(), dup_samples)
+}
+
 // File related operations.
 pub struct Files {
     ctx: Arc<dyn TableContext>,
@@ -40,12 +55,30 @@ impl Files {
         &self,
         file_locations: impl IntoIterator<Item = impl AsRef<str>>,
     ) -> Result<()> {
-        let locations = Vec::from_iter(file_locations.into_iter().map(|v| v.as_ref().to_string()));
+        let mut locations: Vec<String> = file_locations
+            .into_iter()
+            .map(|v| v.as_ref().trim_start_matches('/').to_string())
+            .filter(|v| !v.is_empty())
+            .collect();
 
         if locations.is_empty() {
             return Ok(());
         }
 
+        // Deduplicate: opendal's Deleter uses a HashSet internally but tracks size by
+        // insertion count. Duplicates cause cur_size to diverge from the buffer, making
+        // Deleter::close() loop forever.
+        let (duplicates, dup_samples) = dedup_file_locations(&mut locations);
+        if duplicates > 0 {
+            info!(
+                "remove_file_in_batch: deduplicated {} entries ({} -> {}), duplicate samples: {:?}",
+                duplicates,
+                duplicates + locations.len(),
+                locations.len(),
+                dup_samples
+            );
+        }
+
         // adjusts batch_size according to the `max_threads` settings,
         // limits its min/max value to 1 and 1000.
         let threads_nums = self.ctx.get_settings().get_max_threads()? as usize;
@@ -90,12 +123,6 @@ impl Files {
     #[async_backtrace::framed]
     async fn delete_files(op: Operator, locations: Vec<String>) -> Result<()> {
         let start = Instant::now();
-        // temporary fix for https://github.com/datafuselabs/databend/issues/13804
-        let locations = locations
-            .into_iter()
-            .map(|loc| loc.trim_start_matches('/').to_owned())
-            .filter(|loc| !loc.is_empty())
-            .collect::<Vec<_>>();
         info!("deleting files {:?}", &locations);
         let num_of_files = locations.len();
 

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

@@ -20,6 +20,7 @@ mod read_settings;
 
 pub use buffer_reader::BufferReader;
 pub use files::Files;
+pub use files::dedup_file_locations;
 pub use merge_io_reader::MergeIOReader;
 pub use merge_io_result::MergeIOReadResult;
 pub use merge_io_result::OwnerMemory;