This is an automated email from the ASF dual-hosted git repository. meteorgan pushed a commit to branch delete_with_recursive in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 0bd348ef4d132fcb3df8a40785764d6a1ccf02d9 Author: meteorgan <[email protected]> AuthorDate: Thu Mar 13 23:02:01 2025 +0800 feat(core): support delete with recursive and versions --- core/src/raw/ops.rs | 37 ++++++++++ core/src/types/delete/blocking_deleter.rs | 4 +- core/src/types/error.rs | 2 +- core/src/types/operator/blocking_operator.rs | 21 ++++-- core/src/types/operator/operator.rs | 20 ++++-- core/src/types/operator/operator_functions.rs | 12 ++++ core/src/types/operator/operator_futures.rs | 10 +++ core/tests/behavior/async_delete.rs | 99 +++++++++++++++++++++++---- core/tests/behavior/async_list.rs | 30 -------- core/tests/behavior/blocking_delete.rs | 24 +++---- 10 files changed, 190 insertions(+), 69 deletions(-) diff --git a/core/src/raw/ops.rs b/core/src/raw/ops.rs index b76dbaff1..1b91eb73a 100644 --- a/core/src/raw/ops.rs +++ b/core/src/raw/ops.rs @@ -44,6 +44,21 @@ impl OpCreateDir { #[derive(Debug, Clone, Default, Eq, Hash, PartialEq)] pub struct OpDelete { version: Option<String>, + /// The `recursive` is used to control whether the delete operation is recursive. + /// + /// if `false`, delete operation will only delete the given path. + /// if `true`, delete operation will delete all entries that starts with given path. + /// + /// Default to `false`. + recursive: bool, + /// The `versions` is used to control whether the object versions should be deleted. + /// + /// if `false`, delete operation will only delete the latest version. + /// if `true`, delete operation will delete all the object versions if object versioning is supported + /// by the underlying service. + /// + /// Default to `false`. + versions: bool, } impl OpDelete { @@ -64,6 +79,28 @@ impl OpDelete { pub fn version(&self) -> Option<&str> { self.version.as_deref() } + + /// Change the recursive flag of this delete operation. + pub fn with_recursive(mut self, recursive: bool) -> Self { + self.recursive = recursive; + self + } + + /// Get the recursive flag of this delete operation. + pub fn recursive(&self) -> bool { + self.recursive + } + + /// Change the versions flag of this delete operation. + pub fn with_versions(mut self, versions: bool) -> Self { + self.versions = versions; + self + } + + /// Get the versions flag of this delete operation. + pub fn versions(&self) -> bool { + self.versions + } } /// Args for `delete` operation. diff --git a/core/src/types/delete/blocking_deleter.rs b/core/src/types/delete/blocking_deleter.rs index 53a5f2619..a0b5336a0 100644 --- a/core/src/types/delete/blocking_deleter.rs +++ b/core/src/types/delete/blocking_deleter.rs @@ -62,7 +62,7 @@ impl BlockingDeleter { /// /// Also see: /// - /// - [`BlockingDeleter::delete_try_iter`]: delete an fallible iterator of paths. + /// - [`BlockingDeleter::delete_try_iter`]: delete a fallible iterator of paths. pub fn delete_iter<I, D>(&mut self, iter: I) -> Result<()> where I: IntoIterator<Item = D>, @@ -75,7 +75,7 @@ impl BlockingDeleter { Ok(()) } - /// Delete an fallible iterator of paths. + /// Delete a fallible iterator of paths. /// /// Also see: /// diff --git a/core/src/types/error.rs b/core/src/types/error.rs index 01c489bfc..260ff86c6 100644 --- a/core/src/types/error.rs +++ b/core/src/types/error.rs @@ -80,7 +80,7 @@ pub enum ErrorKind { /// 2. reading a file with If-None-Match header but the file's ETag is match (will get a 304 Not Modified). /// /// As OpenDAL cannot handle the `condition not match` error, it will always return this error to users. - /// So users could to handle this error by themselves. + /// So users could handle this error by themselves. ConditionNotMatch, /// The range of the content is not satisfied. /// diff --git a/core/src/types/operator/blocking_operator.rs b/core/src/types/operator/blocking_operator.rs index b7d4e24b2..d321199db 100644 --- a/core/src/types/operator/blocking_operator.rs +++ b/core/src/types/operator/blocking_operator.rs @@ -16,7 +16,6 @@ // under the License. use super::operator_functions::*; -use crate::raw::oio::BlockingDelete; use crate::raw::*; use crate::*; @@ -758,9 +757,21 @@ impl BlockingOperator { path, OpDelete::new(), |inner, path, args| { - let (_, mut deleter) = inner.blocking_delete()?; - deleter.delete(&path, args)?; - deleter.flush()?; + let mut deleter = BlockingDeleter::create(inner.clone())?; + if args.recursive() { + let op = OpList::new() + .with_recursive(true) + .with_versions(args.versions()); + let lister = BlockingLister::create(inner, path.as_str(), op)?; + deleter.delete_try_iter(lister)?; + } else if args.versions() { + let op = OpList::new().with_versions(true); + let lister = BlockingLister::create(inner, path.as_str(), op)?; + deleter.delete_try_iter(lister)?; + } else { + deleter.delete((path, args))?; + } + deleter.close()?; Ok(()) }, @@ -771,7 +782,7 @@ impl BlockingOperator { /// /// Also see: /// - /// - [`BlockingOperator::delete_try_iter`]: delete an fallible iterator of paths. + /// - [`BlockingOperator::delete_try_iter`]: delete a fallible iterator of paths. pub fn delete_iter<I, D>(&self, iter: I) -> Result<()> where I: IntoIterator<Item = D>, diff --git a/core/src/types/operator/operator.rs b/core/src/types/operator/operator.rs index c79a9f6fa..6fab4ddeb 100644 --- a/core/src/types/operator/operator.rs +++ b/core/src/types/operator/operator.rs @@ -24,7 +24,6 @@ use futures::TryStreamExt; use super::BlockingOperator; use crate::operator_futures::*; -use crate::raw::oio::DeleteDyn; use crate::raw::*; use crate::types::delete::Deleter; use crate::*; @@ -1104,9 +1103,22 @@ impl Operator { path, OpDelete::default(), |inner, path, args| async move { - let (_, mut deleter) = inner.delete_dyn().await?; - deleter.delete_dyn(&path, args)?; - deleter.flush_dyn().await?; + let mut deleter = Deleter::create(inner.clone()).await?; + if args.recursive() { + let op = OpList::new() + .with_recursive(true) + .with_versions(args.versions()); + let lister = Lister::create(inner.clone(), &path, op).await?; + deleter.delete_try_stream(lister).await?; + } else if args.versions() { + let lister = + Lister::create(inner.clone(), &path, OpList::new().with_versions(true)) + .await?; + deleter.delete_try_stream(lister).await?; + } else { + deleter.delete((path, args)).await?; + } + deleter.close().await?; Ok(()) }, ) diff --git a/core/src/types/operator/operator_functions.rs b/core/src/types/operator/operator_functions.rs index e21be9a2e..0e754c72e 100644 --- a/core/src/types/operator/operator_functions.rs +++ b/core/src/types/operator/operator_functions.rs @@ -244,6 +244,18 @@ impl FunctionDelete { self } + /// Set the versions flag for this operation. + pub fn versions(mut self, v: bool) -> Self { + self.0 = self.0.map_args(|args| args.with_versions(v)); + self + } + + /// Set the recursive flag for this operation. + pub fn recursive(mut self, v: bool) -> Self { + self.0 = self.0.map_args(|args| args.with_recursive(v)); + self + } + /// Call the function to consume all the input and generate a /// result. pub fn call(self) -> Result<()> { diff --git a/core/src/types/operator/operator_futures.rs b/core/src/types/operator/operator_futures.rs index c70481720..3d93e9808 100644 --- a/core/src/types/operator/operator_futures.rs +++ b/core/src/types/operator/operator_futures.rs @@ -1554,6 +1554,16 @@ impl<F: Future<Output = Result<()>>> FutureDelete<F> { pub fn version(self, v: &str) -> Self { self.map(|args| args.with_version(v)) } + + /// Change the recursive flag of this delete operation. + pub fn recursive(self, v: bool) -> Self { + self.map(|args| args.with_recursive(v)) + } + + /// Change the versions flag of this delete operation. + pub fn versions(self, v: bool) -> Self { + self.map(|args| args.with_versions(v)) + } } /// Future that generated by [`Operator::deleter_with`]. diff --git a/core/tests/behavior/async_delete.rs b/core/tests/behavior/async_delete.rs index c88ecd66e..2a61cb4b1 100644 --- a/core/tests/behavior/async_delete.rs +++ b/core/tests/behavior/async_delete.rs @@ -17,7 +17,7 @@ use crate::*; use anyhow::Result; -use futures::TryStreamExt; +use futures::{StreamExt, TryStreamExt}; use log::warn; use opendal::raw::{Access, OpDelete}; @@ -32,17 +32,17 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) { test_delete_with_special_chars, test_delete_not_existing, test_delete_stream, - test_remove_one_file, + test_delete_one_file, test_delete_with_version, test_delete_with_not_existing_version, test_batch_delete, - test_batch_delete_with_version + test_batch_delete_with_version, + test_delete_all_basic, + test_delete_one_object_with_versions, + test_delete_all_objects_with_recursive_and_versions )); - if cap.list_with_recursive { - tests.extend(async_trials!(op, test_remove_all_basic)); - if !cap.create_dir { - tests.extend(async_trials!(op, test_remove_all_with_prefix_exists)); - } + if !cap.create_dir { + tests.extend(async_trials!(op, test_delete_all_with_prefix_exists)); } } } @@ -107,7 +107,7 @@ pub async fn test_delete_not_existing(op: Operator) -> Result<()> { } /// Remove one file -pub async fn test_remove_one_file(op: Operator) -> Result<()> { +pub async fn test_delete_one_file(op: Operator) -> Result<()> { let (path, content, _) = TEST_FIXTURE.new_file(op.clone()); op.write(&path, content.clone()) @@ -180,7 +180,7 @@ async fn test_blocking_remove_all_with_objects( op.write(&path, content).await.expect("write must succeed"); } - op.remove_all(&parent).await?; + op.delete_with(&parent).recursive(true).await?; let found = op .lister_with(&format!("{parent}/")) @@ -197,14 +197,14 @@ async fn test_blocking_remove_all_with_objects( Ok(()) } -/// Remove all under a prefix -pub async fn test_remove_all_basic(op: Operator) -> Result<()> { +/// Delete all under a prefix +pub async fn test_delete_all_basic(op: Operator) -> Result<()> { let parent = uuid::Uuid::new_v4().to_string(); test_blocking_remove_all_with_objects(op, parent, ["a/b", "a/c", "a/d/e"]).await } -/// Remove all under a prefix, while the prefix itself is also an object -pub async fn test_remove_all_with_prefix_exists(op: Operator) -> Result<()> { +/// Delete all under a prefix, while the prefix itself is also an object +pub async fn test_delete_all_with_prefix_exists(op: Operator) -> Result<()> { let parent = uuid::Uuid::new_v4().to_string(); let (content, _) = gen_bytes(op.info().full_capability()); op.write(&parent, content) @@ -350,3 +350,74 @@ pub async fn test_batch_delete_with_version(op: Operator) -> Result<()> { Ok(()) } + +pub async fn test_delete_all_objects_with_recursive_and_versions(op: Operator) -> Result<()> { + if !op.info().full_capability().delete_with_version { + return Ok(()); + } + + let parent = uuid::Uuid::new_v4().to_string(); + let paths = ["a/b", "a/d", "a/d/e", "b/c", "c/"]; + for path in paths { + let path = format!("{parent}/{path}"); + if path.ends_with('/') { + op.create_dir(path.as_str()).await?; + } else { + // generate two versions. + op.write(&path, "delete_with_version") + .await + .expect("write must succeed"); + op.write(&path, "delete_with_version_2") + .await + .expect("write must succeed"); + } + } + + let data = op + .lister_with(parent.as_str()) + .recursive(true) + .versions(true) + .await? + .collect::<Vec<_>>() + .await; + assert!(data.len() > paths.len()); + + op.delete_with(parent.as_str()) + .recursive(true) + .versions(true) + .await?; + + let found = op + .lister_with(parent.as_str()) + .recursive(true) + .versions(true) + .await? + .next() + .await + .is_some(); + assert!(found, "all objects should be removed"); + + Ok(()) +} + +pub async fn test_delete_one_object_with_versions(op: Operator) -> Result<()> { + if !op.info().full_capability().delete_with_version { + return Ok(()); + } + + let path = uuid::Uuid::new_v4().to_string(); + for _ in 0..3 { + op.write(path.as_str(), "delete_with_version") + .await + .expect("write must succeed"); + } + let data = op.list_with(path.as_str()).versions(true).await?; + assert_eq!(data.len(), 3); + + op.delete_with(path.as_str()).versions(true).await?; + + let found = op.list_with(path.as_str()).versions(true).await?.len(); + assert_eq!(found, 0, "all objects should be removed"); + + Ok(()) +} diff --git a/core/tests/behavior/async_list.rs b/core/tests/behavior/async_list.rs index 0667a914a..607be4cbe 100644 --- a/core/tests/behavior/async_list.rs +++ b/core/tests/behavior/async_list.rs @@ -47,7 +47,6 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) { test_list_dir_with_recursive_no_trailing_slash, test_list_file_with_recursive, test_list_root_with_recursive, - test_remove_all, test_list_files_with_versions, test_list_with_versions_and_limit, test_list_with_versions_and_start_after, @@ -541,35 +540,6 @@ pub async fn test_list_file_with_recursive(op: Operator) -> Result<()> { Ok(()) } -// Remove all should remove all in this path. -pub async fn test_remove_all(op: Operator) -> Result<()> { - let parent = uuid::Uuid::new_v4().to_string(); - - let expected = [ - "x/", "x/y", "x/x/", "x/x/y", "x/x/x/", "x/x/x/y", "x/x/x/x/", - ]; - for path in expected.iter() { - if path.ends_with('/') { - op.create_dir(&format!("{parent}/{path}")).await?; - } else { - op.write(&format!("{parent}/{path}"), "test_scan").await?; - } - } - - op.remove_all(&format!("{parent}/x/")).await?; - - for path in expected.iter() { - if path.ends_with('/') { - continue; - } - assert!( - !op.exists(&format!("{parent}/{path}")).await?, - "{parent}/{path} should be removed" - ) - } - Ok(()) -} - /// Stat normal file and dir should return metadata pub async fn test_list_only(op: Operator) -> Result<()> { let mut entries = HashMap::new(); diff --git a/core/tests/behavior/blocking_delete.rs b/core/tests/behavior/blocking_delete.rs index 7b66e520d..2633bb743 100644 --- a/core/tests/behavior/blocking_delete.rs +++ b/core/tests/behavior/blocking_delete.rs @@ -27,16 +27,14 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) { tests.extend(blocking_trials!( op, test_blocking_delete_file, - test_blocking_remove_one_file + test_blocking_delete_one_file, + test_blocking_delete_all_basic )); - if cap.list_with_recursive { - tests.extend(blocking_trials!(op, test_blocking_remove_all_basic)); - if !cap.create_dir { - tests.extend(blocking_trials!( - op, - test_blocking_remove_all_with_prefix_exists - )); - } + if !cap.create_dir { + tests.extend(blocking_trials!( + op, + test_blocking_delete_all_with_prefix_exists + )); } } } @@ -58,7 +56,7 @@ pub fn test_blocking_delete_file(op: BlockingOperator) -> Result<()> { } /// Remove one file -pub fn test_blocking_remove_one_file(op: BlockingOperator) -> Result<()> { +pub fn test_blocking_delete_one_file(op: BlockingOperator) -> Result<()> { let path = uuid::Uuid::new_v4().to_string(); let (content, _) = gen_bytes(op.info().full_capability()); @@ -83,7 +81,7 @@ fn test_blocking_remove_all_with_objects( op.write(&path, content).expect("write must succeed"); } - op.remove_all(&parent)?; + op.delete_with(&parent).recursive(true).call()?; let found = op .lister_with(&format!("{parent}/")) @@ -99,13 +97,13 @@ fn test_blocking_remove_all_with_objects( } /// Remove all under a prefix -pub fn test_blocking_remove_all_basic(op: BlockingOperator) -> Result<()> { +pub fn test_blocking_delete_all_basic(op: BlockingOperator) -> Result<()> { let parent = uuid::Uuid::new_v4().to_string(); test_blocking_remove_all_with_objects(op, parent, ["a/b", "a/c", "a/d/e"]) } /// Remove all under a prefix, while the prefix itself is also an object -pub fn test_blocking_remove_all_with_prefix_exists(op: BlockingOperator) -> Result<()> { +pub fn test_blocking_delete_all_with_prefix_exists(op: BlockingOperator) -> Result<()> { let parent = uuid::Uuid::new_v4().to_string(); let (content, _) = gen_bytes(op.info().full_capability()); op.write(&parent, content).expect("write must succeed");
