feat(core/types): blocking remove_all for object storage based servic…

…es (#4665)

feat(core/types): blocking remove_all on oss

Signed-off-by: TennyZhuang <zty0826@gmail.com>

core/src/types/operator/blocking_operator.rs

@@ -846,18 +846,21 @@ impl BlockingOperator {
  /// # }
  /// ```
  pub fn remove_all(&self, path: &str) -> Result<()> {
- let meta = match self.stat(path) {
- Ok(metadata) => metadata,
+ match self.stat(path) {
+ Ok(metadata) => {
+ if metadata.mode() != EntryMode::DIR {
+ self.delete(path)?;
+ // There may still be objects prefixed with the path in some backend, so we can't return here.
+ }
+ }
 
- Err(e) if e.kind() == ErrorKind::NotFound => return Ok(()),
+ // If dir not found, it may be a prefix in object store like S3,
+ // and we still need to delete objects under the prefix.
+ Err(e) if e.kind() == ErrorKind::NotFound => {}
 
  Err(e) => return Err(e),
  };
 
- if meta.mode() != EntryMode::DIR {
- return self.delete(path);
- }
-
  let obs = self.lister_with(path).recursive(true).call()?;
 
  for v in obs {

core/src/types/operator/operator.rs

@@ -1416,14 +1416,7 @@ impl Operator {
  Err(e) => return Err(e),
  };
 
- let obs = match self.lister_with(path).recursive(true).await {
- Ok(obs) => obs,
- Err(e) if e.kind() == ErrorKind::NotFound => {
- // If lister still returns NotFound, we can confirm there are no objects under the prefix in any backend.
- return Ok(());
- }
- Err(e) => return Err(e),
- };
+ let obs = self.lister_with(path).recursive(true).await?;
 
  if self.info().full_capability().batch {
  let mut obs = obs.try_chunks(self.limit());