Skip to content

fix: use batch deletes in object store prefix deletion #1575

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
parmesant merged 1 commit into from
Mar 7, 2026
Merged

fix: use batch deletes in object store prefix deletion #1575

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 30 additions & 39 deletions src/storage/azure_blob.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,15 +16,7 @@
*
*/

use std::{
collections::HashSet,
path::Path,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
time::Duration,
};
use std::{collections::HashSet, path::Path, sync::Arc, time::Duration};

use async_trait::async_trait;
use bytes::Bytes;
Expand Down Expand Up @@ -271,50 +263,49 @@ impl BlobStore {
tenant_id: &Option<String>,
) -> Result<(), ObjectStorageError> {
let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
let files_scanned = Arc::new(AtomicU64::new(0));
let files_deleted = Arc::new(AtomicU64::new(0));

// List all objects under the prefix
let object_stream = self.client.list(Some(&(key.into())));
increment_object_store_calls_by_date("LIST", &Utc::now().date_naive().to_string(), tenant);

object_stream
.for_each_concurrent(None, |x| async {
files_scanned.fetch_add(1, Ordering::Relaxed);

match x {
Ok(obj) => {
files_deleted.fetch_add(1, Ordering::Relaxed);
let delete_resp = self.client.delete(&obj.location).await;
increment_object_store_calls_by_date(
"DELETE",
&Utc::now().date_naive().to_string(),
tenant,
);
if delete_resp.is_err() {
error!(
"Failed to delete object during delete stream: {:?}",
delete_resp
);
}
}
Err(err) => {
error!("Failed to fetch object during delete stream: {:?}", err);
}
};
})
.await;
// Map listed objects to their locations for delete_stream
let locations_stream = object_stream.map(|result| result.map(|obj| obj.location));

// Use delete_stream which batches deletes via the bulk delete API
let mut delete_results = self.client.delete_stream(Box::pin(locations_stream));

let mut files_deleted: u64 = 0;
let mut failed_deletes: u64 = 0;
while let Some(result) = delete_results.next().await {
match result {
Ok(_) => files_deleted += 1,
Err(err) => {
failed_deletes += 1;
error!("Failed to delete object during prefix deletion: {:?}", err);
}
}
}

let total_files = files_deleted + failed_deletes;
increment_files_scanned_in_object_store_calls_by_date(
"LIST",
files_scanned.load(Ordering::Relaxed),
total_files,
&Utc::now().date_naive().to_string(),
tenant,
);
increment_files_scanned_in_object_store_calls_by_date(
"DELETE",
files_deleted.load(Ordering::Relaxed),
files_deleted,
&Utc::now().date_naive().to_string(),
tenant,
);

if failed_deletes > 0 {
return Err(ObjectStorageError::UnhandledError(
anyhow::anyhow!("Failed to delete {failed_deletes} out of {total_files} objects under prefix '{key}'").into(),
));
}

Ok(())
}

Expand Down
71 changes: 31 additions & 40 deletions src/storage/gcs.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,15 +16,7 @@
*
*/

use std::{
collections::HashSet,
path::Path,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
time::Duration,
};
use std::{collections::HashSet, path::Path, sync::Arc, time::Duration};

use crate::{
metrics::{
Expand Down Expand Up @@ -234,51 +226,50 @@ impl Gcs {
key: &str,
tenant_id: &Option<String>,
) -> Result<(), ObjectStorageError> {
let files_scanned = Arc::new(AtomicU64::new(0));
let files_deleted = Arc::new(AtomicU64::new(0));
let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
// Track LIST operation

// List all objects under the prefix
let object_stream = self.client.list(Some(&(key.into())));
increment_object_store_calls_by_date("LIST", &Utc::now().date_naive().to_string(), tenant);
object_stream
.for_each_concurrent(None, |x| async {
files_scanned.fetch_add(1, Ordering::Relaxed);

match x {
Ok(obj) => {
files_deleted.fetch_add(1, Ordering::Relaxed);
let delete_resp = self.client.delete(&obj.location).await;
increment_object_store_calls_by_date(
"DELETE",
&Utc::now().date_naive().to_string(),
tenant,
);
if delete_resp.is_err() {
error!(
"Failed to delete object during delete stream: {:?}",
delete_resp
);
}
}
Err(err) => {
error!("Failed to fetch object during delete stream: {:?}", err);
}
};
})
.await;

// Map listed objects to their locations for delete_stream
let locations_stream = object_stream.map(|result| result.map(|obj| obj.location));

// Use delete_stream which batches deletes via the bulk delete API
let mut delete_results = self.client.delete_stream(Box::pin(locations_stream));

let mut files_deleted: u64 = 0;
let mut failed_deletes: u64 = 0;
while let Some(result) = delete_results.next().await {
match result {
Ok(_) => files_deleted += 1,
Err(err) => {
failed_deletes += 1;
error!("Failed to delete object during prefix deletion: {:?}", err);
}
}
}

let total_files = files_deleted + failed_deletes;
increment_files_scanned_in_object_store_calls_by_date(
"LIST",
files_scanned.load(Ordering::Relaxed),
total_files,
&Utc::now().date_naive().to_string(),
tenant,
);
increment_files_scanned_in_object_store_calls_by_date(
"DELETE",
files_deleted.load(Ordering::Relaxed),
files_deleted,
&Utc::now().date_naive().to_string(),
tenant,
);

if failed_deletes > 0 {
return Err(ObjectStorageError::UnhandledError(
anyhow::anyhow!("Failed to delete {failed_deletes} out of {total_files} objects under prefix '{key}'").into(),
));
}

Ok(())
}

Expand Down
71 changes: 30 additions & 41 deletions src/storage/s3.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,15 +17,7 @@
*/

use std::{
collections::HashSet,
fmt::Display,
path::Path,
str::FromStr,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
time::Duration,
collections::HashSet, fmt::Display, path::Path, str::FromStr, sync::Arc, time::Duration,
};

use async_trait::async_trait;
Expand Down Expand Up @@ -406,56 +398,53 @@ impl S3 {
tenant_id: &Option<String>,
) -> Result<(), ObjectStorageError> {
let tenant_str = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
let files_scanned = Arc::new(AtomicU64::new(0));
let files_deleted = Arc::new(AtomicU64::new(0));
// Track LIST operation

// List all objects under the prefix
let object_stream = self.client.list(Some(&(key.into())));
increment_object_store_calls_by_date(
"LIST",
&Utc::now().date_naive().to_string(),
tenant_str,
);

let tenant_str_clone = tenant_str.to_string();
object_stream
.for_each_concurrent(None, |x| async {
files_scanned.fetch_add(1, Ordering::Relaxed);

match x {
Ok(obj) => {
files_deleted.fetch_add(1, Ordering::Relaxed);
let delete_resp = self.client.delete(&obj.location).await;
increment_object_store_calls_by_date(
"DELETE",
&Utc::now().date_naive().to_string(),
&tenant_str_clone,
);
if delete_resp.is_err() {
error!(
"Failed to delete object during delete stream: {:?}",
delete_resp
);
}
}
Err(err) => {
error!("Failed to fetch object during delete stream: {:?}", err);
}
};
})
.await;
// Map listed objects to their locations for delete_stream
let locations_stream = object_stream.map(|result| result.map(|obj| obj.location));

// Use delete_stream which batches deletes via the S3 DeleteObjects API
let mut delete_results = self.client.delete_stream(Box::pin(locations_stream));

let mut files_deleted: u64 = 0;
let mut failed_deletes: u64 = 0;
while let Some(result) = delete_results.next().await {
match result {
Ok(_) => files_deleted += 1,
Err(err) => {
failed_deletes += 1;
error!("Failed to delete object during prefix deletion: {:?}", err);
}
}
}

let total_files = files_deleted + failed_deletes;
increment_files_scanned_in_object_store_calls_by_date(
"LIST",
files_scanned.load(Ordering::Relaxed),
total_files,
&Utc::now().date_naive().to_string(),
tenant_str,
);
increment_files_scanned_in_object_store_calls_by_date(
"DELETE",
files_deleted.load(Ordering::Relaxed),
files_deleted,
&Utc::now().date_naive().to_string(),
tenant_str,
);

if failed_deletes > 0 {
return Err(ObjectStorageError::UnhandledError(
anyhow::anyhow!("Failed to delete {failed_deletes} out of {total_files} objects under prefix '{key}'").into(),
));
}

Ok(())
}

Expand Down
Loading