feat!: bulk delete for vacuum

python/deltalake/table.py

@@ -414,17 +414,13 @@ def vacuum(
         retention_hours: Optional[int] = None,
         dry_run: bool = True,
         enforce_retention_duration: bool = True,
-        max_concurrent_requests: int = 10,
     ) -> List[str]:
         """
         Run the Vacuum command on the Delta Table: list and delete files no longer referenced by the Delta table and are older than the retention threshold.
 
         :param retention_hours: the retention threshold in hours, if none then the value from `configuration.deletedFileRetentionDuration` is used or default of 1 week otherwise.
         :param dry_run: when activated, list only the files, delete otherwise
         :param enforce_retention_duration: when disabled, accepts retention hours smaller than the value from `configuration.deletedFileRetentionDuration`.
-        :param max_concurrent_requests: the maximum number of concurrent requests to send to the backend.
-          Increasing this number may improve performance of vacuuming large tables, however it might also
-          increase the risk of hitting rate limits.
         :return: the list of files no longer referenced by the Delta Table and are older than the retention threshold.
         """
         if retention_hours:
@@ -435,7 +431,6 @@ def vacuum(
             dry_run,
             retention_hours,
             enforce_retention_duration,
-            max_concurrent_requests,
         )
 
     @property

python/src/lib.rs

@@ -244,18 +244,16 @@ impl RawDeltaTable {
 
     /// Run the Vacuum command on the Delta Table: list and delete files no longer referenced
     /// by the Delta table and are older than the retention threshold.
-    #[pyo3(signature = (dry_run, retention_hours = None, enforce_retention_duration = true, max_concurrent_requests = 10))]
+    #[pyo3(signature = (dry_run, retention_hours = None, enforce_retention_duration = true))]
     pub fn vacuum(
         &mut self,
         dry_run: bool,
         retention_hours: Option<u64>,
         enforce_retention_duration: bool,
-        max_concurrent_requests: usize,
     ) -> PyResult<Vec<String>> {
         let mut cmd = VacuumBuilder::new(self._table.object_store(), self._table.state.clone())
             .with_enforce_retention_duration(enforce_retention_duration)
-            .with_dry_run(dry_run)
-            .with_max_concurrent_requests(max_concurrent_requests);
+            .with_dry_run(dry_run);
         if let Some(retention_period) = retention_hours {
             cmd = cmd.with_retention_period(Duration::hours(retention_period as i64));
         }

rust/src/operations/vacuum.rs

@@ -28,6 +28,7 @@ use std::sync::Arc;
 use chrono::{Duration, Utc};
 use futures::future::BoxFuture;
 use futures::{StreamExt, TryStreamExt};
+use object_store::Error;
 use object_store::{path::Path, ObjectStore};
 
 use crate::errors::{DeltaResult, DeltaTableError};
@@ -80,8 +81,6 @@ pub struct VacuumBuilder {
     retention_period: Option<Duration>,
     /// Validate the retention period is not below the retention period configured in the table
     enforce_retention_duration: bool,
-    /// Maximum number of concurrent requests to make
-    max_concurrent_requests: usize,
     /// Don't delete the files. Just determine which files can be deleted
     dry_run: bool,
     /// Override the source of time
@@ -106,7 +105,6 @@ impl VacuumBuilder {
             store,
             retention_period: None,
             enforce_retention_duration: true,
-            max_concurrent_requests: 10,
             dry_run: false,
             clock: None,
         }
@@ -130,12 +128,6 @@ impl VacuumBuilder {
         self
     }
 
-    /// Set the maximum number of concurrent requests to make
-    pub fn with_max_concurrent_requests(mut self, max_concurrent_requests: usize) -> Self {
-        self.max_concurrent_requests = max_concurrent_requests;
-        self
-    }
-
     /// add a time source for testing
     #[doc(hidden)]
     pub fn with_clock(mut self, clock: Arc<dyn Clock>) -> Self {
@@ -186,10 +178,7 @@ impl VacuumBuilder {
             files_to_delete.push(obj_meta.location);
         }
 
-        Ok(VacuumPlan {
-            files_to_delete,
-            max_concurrent_requests: self.max_concurrent_requests,
-        })
+        Ok(VacuumPlan { files_to_delete })
     }
 }
 
@@ -225,8 +214,6 @@ impl std::future::IntoFuture for VacuumBuilder {
 struct VacuumPlan {
     /// What files are to be deleted
     pub files_to_delete: Vec<Path>,
-    /// How many concurrent requests to make
-    pub max_concurrent_requests: usize,
 }
 
 impl VacuumPlan {
@@ -239,13 +226,17 @@ impl VacuumPlan {
             });
         }
 
-        let files_deleted = futures::stream::iter(self.files_to_delete)
-            .map(|path| {
-                let store = store.clone();
-                async move { store.delete(&path).await.map(|_| path) }
+        let locations = futures::stream::iter(self.files_to_delete)
+            .map(Result::Ok)
+            .boxed();
+
+        let files_deleted = store
+            .delete_stream(locations)
+            .map(|res| match res {
+                Ok(path) => Ok(path.to_string()),
+                Err(Error::NotFound { path, .. }) => Ok(path),
+                Err(err) => Err(err),
             })
-            .buffer_unordered(self.max_concurrent_requests)
-            .map_ok(|path| path.to_string())
             .try_collect::<Vec<_>>()
             .await?;