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

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());