Implement delete_objs in fs and s3 storage backends.

.github/workflows/build.yml

@@ -122,3 +122,4 @@ jobs:
        cargo test s3 --test concurrent_writes_test --features s3
        cargo test --test dynamodb_lock_test --features s3
        cargo test --test repair_s3_rename_test --features s3
+       cargo test test_s3_delete_objs --test s3_test --features s3

rust/Cargo.toml

@@ -26,6 +26,7 @@ regex = "1"
 chrono = "0"
 uuid = { version = "0.8", features = ["serde", "v4"] }
 lazy_static = "1"
+itertools = "0"
 
 # Azure
 reqwest = { version = "0", optional = true }

rust/src/delta.rs

@@ -989,19 +989,14 @@ impl DeltaTable {
             return Ok(files_to_delete);
         }
 
-        for rel_path in &files_to_delete {
-            match self
-                .storage
-                .delete_obj(&self.storage.join_path(&self.table_uri, rel_path))
-                .await
-            {
-                Ok(_) => continue,
-                Err(StorageError::NotFound) => continue,
-                Err(err) => return Err(DeltaTableError::StorageError { source: err }),
-            }
+        let paths = &files_to_delete
+            .iter()
+            .map(|rel_path| self.storage.join_path(&self.table_uri, rel_path))
+            .collect::<Vec<_>>();
+        match self.storage.delete_objs(paths).await {
+            Ok(_) => Ok(files_to_delete),
+            Err(err) => Err(DeltaTableError::StorageError { source: err }),
         }
-
-        Ok(files_to_delete)
     }
 
     /// Return table schema parsed from transaction log. Return None if table hasn't been loaded or

rust/src/storage/azure.rs

@@ -200,6 +200,10 @@ impl StorageBackend for AdlsGen2Backend {
     async fn delete_obj(&self, _path: &str) -> Result<(), StorageError> {
         unimplemented!("delete_obj not implemented for azure");
     }
+
+    async fn delete_objs(&self, _paths: &[String]) -> Result<(), StorageError> {
+        unimplemented!("delete_objs not implemented for azure");
+    }
 }
 
 #[cfg(test)]

rust/src/storage/file/mod.rs

@@ -134,6 +134,17 @@ impl StorageBackend for FileStorageBackend {
     async fn delete_obj(&self, path: &str) -> Result<(), StorageError> {
         fs::remove_file(path).await.map_err(StorageError::from)
     }
+
+    async fn delete_objs(&self, paths: &[String]) -> Result<(), StorageError> {
+        for path in paths {
+            match self.delete_obj(path).await {
+                Ok(_) => continue,
+                Err(StorageError::NotFound) => continue,
+                Err(e) => return Err(e),
+            }
+        }
+        Ok(())
+    }
 }
 
 #[cfg(test)]
@@ -182,6 +193,30 @@ mod tests {
         assert_eq!(fs::metadata(path).await.is_ok(), false)
     }
 
+    #[tokio::test]
+    async fn delete_objs() {
+        let tmp_dir = tempdir::TempDir::new("delete_test").unwrap();
+        let tmp_file_path1 = tmp_dir.path().join("tmp_file1");
+        let tmp_file_path2 = tmp_dir.path().join("tmp_file2");
+        let backend = FileStorageBackend::new(tmp_dir.path().to_str().unwrap());
+
+        // put object
+        let path1 = tmp_file_path1.to_str().unwrap();
+        let path2 = tmp_file_path2.to_str().unwrap();
+        backend.put_obj(path1, &[]).await.unwrap();
+        backend.put_obj(path2, &[]).await.unwrap();
+        assert_eq!(fs::metadata(path1).await.is_ok(), true);
+        assert_eq!(fs::metadata(path2).await.is_ok(), true);
+
+        // delete object
+        backend
+            .delete_objs(&[path1.to_string(), path2.to_string()])
+            .await
+            .unwrap();
+        assert_eq!(fs::metadata(path1).await.is_ok(), false);
+        assert_eq!(fs::metadata(path2).await.is_ok(), false)
+    }
+
     #[test]
     fn join_multiple_paths() {
         let backend = FileStorageBackend::new("./");

rust/src/storage/mod.rs

@@ -252,6 +252,14 @@ pub enum StorageError {
         #[from]
         source: rusoto_core::RusotoError<rusoto_s3::DeleteObjectError>,
     },
+    /// Error representing a failure when executing an S3 DeleteObjects request.
+    #[cfg(any(feature = "s3", feature = "s3-rustls"))]
+    #[error("Failed to delete S3 object: {source}")]
+    S3DeleteMultiple {
+        /// The underlying Rusoto S3 error.
+        #[from]
+        source: rusoto_core::RusotoError<rusoto_s3::DeleteObjectsError>,
+    },
     /// Error representing a failure when copying a S3 object
     #[cfg(any(feature = "s3", feature = "s3-rustls"))]
     #[error("Failed to copy S3 object: {source}")]
@@ -430,6 +438,9 @@ pub trait StorageBackend: Send + Sync + Debug {
 
     /// Deletes object by `path`.
     async fn delete_obj(&self, path: &str) -> Result<(), StorageError>;
+
+    /// Deletes object by `path`.
+    async fn delete_objs(&self, paths: &[String]) -> Result<(), StorageError>;
 }
 
 /// Dynamically construct a Storage backend trait object based on scheme for provided URI

rust/src/storage/s3/mod.rs

@@ -6,12 +6,13 @@ use std::{fmt, pin::Pin};
 
 use chrono::{DateTime, FixedOffset, Utc};
 use futures::Stream;
+use itertools::Itertools;
 use log::debug;
 use rusoto_core::{HttpClient, Region, RusotoError};
 use rusoto_credential::AutoRefreshingProvider;
 use rusoto_s3::{
-    CopyObjectRequest, DeleteObjectRequest, GetObjectRequest, HeadObjectRequest,
-    ListObjectsV2Request, PutObjectRequest, S3Client, S3,
+    CopyObjectRequest, Delete, DeleteObjectRequest, DeleteObjectsRequest, GetObjectRequest,
+    HeadObjectRequest, ListObjectsV2Request, ObjectIdentifier, PutObjectRequest, S3Client, S3,
 };
 use rusoto_sts::{StsAssumeRoleSessionCredentialsProvider, StsClient, WebIdentityProvider};
 use serde::{Deserialize, Serialize};
@@ -416,13 +417,47 @@ impl StorageBackend for S3StorageBackend {
         debug!("delete s3 object: {}...", path);
 
         let uri = parse_uri(path)?.into_s3object()?;
-        let put_req = DeleteObjectRequest {
+        let delete_req = DeleteObjectRequest {
             bucket: uri.bucket.to_string(),
             key: uri.key.to_string(),
             ..Default::default()
         };
 
-        self.client.delete_object(put_req).await?;
+        self.client.delete_object(delete_req).await?;
+
+        Ok(())
+    }
+
+    async fn delete_objs(&self, paths: &[String]) -> Result<(), StorageError> {
+        debug!("delete s3 objects: {:?}...", paths);
+
+        let uris = paths
+            .iter()
+            .map(|path| {
+                let s3_obj = parse_uri(path).unwrap().into_s3object().unwrap();
+                (s3_obj.bucket, s3_obj)
+            })
+            .into_group_map();
+
+        for (bucket, objs) in uris {
+            let delete = Delete {
+                objects: objs
+                    .iter()
+                    .map(|obj| ObjectIdentifier {
+                        key: obj.key.to_string(),
+                        ..Default::default()
+                    })
+                    .collect(),
+                quiet: None,
+            };
+            let delete_req = DeleteObjectsRequest {
+                bucket: bucket.to_string(),
+                delete,
+                ..Default::default()
+            };
+
+            self.client.delete_objects(delete_req).await?;
+        }
 
         Ok(())
     }

rust/tests/s3_test.rs

@@ -134,4 +134,26 @@ mod s3 {
 
         assert!(matches!(err, StorageError::NotFound));
     }
+
+    #[tokio::test]
+    #[serial]
+    async fn test_s3_delete_objs() {
+        setup();
+
+        let path1 = "s3://deltars/delete1.snappy.parquet";
+        let path2 = "s3://deltars/delete2.snappy.parquet";
+        let backend = deltalake::get_backend_for_uri(path1).unwrap();
+        backend.put_obj(path1, &[]).await.unwrap();
+        backend.put_obj(path2, &[]).await.unwrap();
+
+        backend
+            .delete_objs(&[path1.to_string(), path2.to_string()])
+            .await
+            .unwrap();
+        let err1 = backend.head_obj(path1).await.err().unwrap();
+        let err2 = backend.head_obj(path2).await.err().unwrap();
+
+        assert!(matches!(err1, StorageError::NotFound));
+        assert!(matches!(err2, StorageError::NotFound));
+    }
 }