diff --git a/core/src/types/operator/operator.rs b/core/src/types/operator/operator.rs
index c0f08d515f47..63557797689e 100644
--- a/core/src/types/operator/operator.rs
+++ b/core/src/types/operator/operator.rs
@@ -911,6 +911,8 @@ impl Operator {
     /// # }
     /// ```
     pub async fn remove_via(&self, input: impl Stream<Item = String> + Unpin) -> Result<()> {
+        let input = input.map(|v| normalize_path(&v));
+
         if self.info().full_capability().batch {
             let mut input = input
                 .map(|v| (v, OpDelete::default().into()))
diff --git a/core/tests/behavior/write.rs b/core/tests/behavior/write.rs
index 88585947f25c..f76765945e57 100644
--- a/core/tests/behavior/write.rs
+++ b/core/tests/behavior/write.rs
@@ -1176,7 +1176,18 @@ pub async fn test_remove_one_file(op: Operator) -> Result<()> {
     let path = uuid::Uuid::new_v4().to_string();
     let (content, _) = gen_bytes(op.info().full_capability());
 
-    op.write(&path, content).await.expect("write must succeed");
+    op.write(&path, content.clone())
+        .await
+        .expect("write must succeed");
+
+    op.remove(vec![path.clone()]).await?;
+
+    // Stat it again to check.
+    assert!(!op.is_exist(&path).await?);
+
+    op.write(&format!("/{path}"), content)
+        .await
+        .expect("write must succeed");
 
     op.remove(vec![path.clone()]).await?;