diff --git a/core/src/raw/oio/list/prefix_list.rs b/core/src/raw/oio/list/prefix_list.rs
index f176e9bb9f1c..166dbcaecebf 100644
--- a/core/src/raw/oio/list/prefix_list.rs
+++ b/core/src/raw/oio/list/prefix_list.rs
@@ -53,6 +53,15 @@ impl<L> PrefixLister<L> {
     }
 }
 
+#[inline]
+fn starts_with_not_eq(entry: &oio::Entry, prefix: &str) -> bool {
+    match entry.path().strip_prefix(prefix) {
+        None => false,
+        Some("") => false,
+        Some(_) => true,
+    }
+}
+
 impl<L> oio::List for PrefixLister<L>
 where
     L: oio::List,
@@ -60,7 +69,7 @@ where
     fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<oio::Entry>>> {
         loop {
             match ready!(self.lister.poll_next(cx)) {
-                Ok(Some(e)) if !e.path().starts_with(&self.prefix) => continue,
+                Ok(Some(e)) if !starts_with_not_eq(&e, &self.prefix) => continue,
                 v => return Poll::Ready(v),
             }
         }
@@ -74,7 +83,7 @@ where
     fn next(&mut self) -> Result<Option<oio::Entry>> {
         loop {
             match self.lister.next() {
-                Ok(Some(e)) if !e.path().starts_with(&self.prefix) => continue,
+                Ok(Some(e)) if !starts_with_not_eq(&e, &self.prefix) => continue,
                 v => return v,
             }
         }
diff --git a/core/tests/behavior/async_list.rs b/core/tests/behavior/async_list.rs
index 9358ec4f0492..2d22183e5234 100644
--- a/core/tests/behavior/async_list.rs
+++ b/core/tests/behavior/async_list.rs
@@ -44,7 +44,9 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
             test_list_nested_dir,
             test_list_dir_with_file_path,
             test_list_with_start_after,
-            test_list_with_recursive,
+            test_list_dir_with_recursive,
+            test_list_dir_with_recursive_no_trailing_slash,
+            test_list_file_with_recursive,
             test_list_root_with_recursive,
             test_remove_all
         ))
@@ -476,31 +478,30 @@ pub async fn test_list_root_with_recursive(op: Operator) -> Result<()> {
         .map(|v| v.path().to_string())
         .collect::<HashSet<_>>();
 
-    assert!(!actual.contains("/"), "empty root should return itself");
-    assert!(!actual.contains(""), "empty root should return empty");
+    assert!(!actual.contains("/"), "empty root should not return itself");
+    assert!(!actual.contains(""), "empty root should not return empty");
     Ok(())
 }
 
 // Walk top down should output as expected
-pub async fn test_list_with_recursive(op: Operator) -> Result<()> {
+pub async fn test_list_dir_with_recursive(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/",
+    let paths = [
+        "x/", "x/x/", "x/x/x/", "x/x/x/x/", "x/x/x/y", "x/x/y", "x/y", "x/yy",
     ];
-    for path in expected.iter() {
+    for path in paths.iter() {
         if path.ends_with('/') {
             op.create_dir(&format!("{parent}/{path}")).await?;
         } else {
             op.write(&format!("{parent}/{path}"), "test_scan").await?;
         }
     }
-
     let w = op
         .lister_with(&format!("{parent}/x/"))
         .recursive(true)
         .await?;
-    let actual = w
+    let mut actual = w
         .try_collect::<Vec<_>>()
         .await?
         .into_iter()
@@ -510,13 +511,82 @@ pub async fn test_list_with_recursive(op: Operator) -> Result<()> {
                 .unwrap()
                 .to_string()
         })
-        .collect::<HashSet<_>>();
+        .collect::<Vec<_>>();
+    actual.sort();
+
+    let expected = vec![
+        "x/x/", "x/x/x/", "x/x/x/x/", "x/x/x/y", "x/x/y", "x/y", "x/yy",
+    ];
+    assert_eq!(actual, expected);
+    Ok(())
+}
+
+// same as test_list_dir_with_recursive except listing 'x' instead of 'x/'
+pub async fn test_list_dir_with_recursive_no_trailing_slash(op: Operator) -> Result<()> {
+    let parent = uuid::Uuid::new_v4().to_string();
+
+    let paths = [
+        "x/", "x/x/", "x/x/x/", "x/x/x/x/", "x/x/x/y", "x/x/y", "x/y", "x/yy",
+    ];
+    for path in paths.iter() {
+        if path.ends_with('/') {
+            op.create_dir(&format!("{parent}/{path}")).await?;
+        } else {
+            op.write(&format!("{parent}/{path}"), "test_scan").await?;
+        }
+    }
+    let w = op
+        .lister_with(&format!("{parent}/x"))
+        .recursive(true)
+        .await?;
+    let mut actual = w
+        .try_collect::<Vec<_>>()
+        .await?
+        .into_iter()
+        .map(|v| {
+            v.path()
+                .strip_prefix(&format!("{parent}/"))
+                .unwrap()
+                .to_string()
+        })
+        .collect::<Vec<_>>();
+    actual.sort();
 
-    debug!("walk top down: {:?}", actual);
+    let expected = paths.to_vec();
+    assert_eq!(actual, expected);
+    Ok(())
+}
 
-    assert!(actual.contains("x/y"));
-    assert!(actual.contains("x/x/y"));
-    assert!(actual.contains("x/x/x/y"));
+pub async fn test_list_file_with_recursive(op: Operator) -> Result<()> {
+    let parent = uuid::Uuid::new_v4().to_string();
+
+    let paths = ["y", "yy"];
+    for path in paths.iter() {
+        if path.ends_with('/') {
+            op.create_dir(&format!("{parent}/{path}")).await?;
+        } else {
+            op.write(&format!("{parent}/{path}"), "test_scan").await?;
+        }
+    }
+    let w = op
+        .lister_with(&format!("{parent}/y"))
+        .recursive(true)
+        .await?;
+    let mut actual = w
+        .try_collect::<Vec<_>>()
+        .await?
+        .into_iter()
+        .map(|v| {
+            v.path()
+                .strip_prefix(&format!("{parent}/"))
+                .unwrap()
+                .to_string()
+        })
+        .collect::<Vec<_>>();
+    actual.sort();
+
+    let expected = vec!["yy"];
+    assert_eq!(actual, expected);
     Ok(())
 }
 
diff --git a/core/tests/behavior/blocking_list.rs b/core/tests/behavior/blocking_list.rs
index 9208929287a7..c6bbd25f60b9 100644
--- a/core/tests/behavior/blocking_list.rs
+++ b/core/tests/behavior/blocking_list.rs
@@ -16,7 +16,6 @@
 // under the License.
 
 use std::collections::HashMap;
-use std::collections::HashSet;
 
 use anyhow::Result;
 use log::debug;
@@ -33,7 +32,9 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
             test_blocking_list_dir_with_metakey,
             test_blocking_list_dir_with_metakey_complete,
             test_blocking_list_non_exist_dir,
-            test_blocking_scan,
+            test_blocking_list_dir_with_recursive,
+            test_blocking_list_dir_with_recursive_no_trailing_slash,
+            test_blocking_list_file_with_recursive,
             test_blocking_remove_all
         ))
     }
@@ -172,13 +173,14 @@ pub fn test_blocking_list_non_exist_dir(op: BlockingOperator) -> Result<()> {
     Ok(())
 }
 
-// Walk top down should output as expected
-pub fn test_blocking_scan(op: BlockingOperator) -> Result<()> {
+// Remove all should remove all in this path.
+pub fn test_blocking_remove_all(op: BlockingOperator) -> 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}"))?;
@@ -187,11 +189,40 @@ pub fn test_blocking_scan(op: BlockingOperator) -> Result<()> {
         }
     }
 
+    op.remove_all(&format!("{parent}/x/"))?;
+
+    for path in expected.iter() {
+        if path.ends_with('/') {
+            continue;
+        }
+        assert!(
+            !op.is_exist(&format!("{parent}/{path}"))?,
+            "{parent}/{path} should be removed"
+        )
+    }
+
+    Ok(())
+}
+
+// Walk top down should output as expected
+pub fn test_blocking_list_dir_with_recursive(op: BlockingOperator) -> Result<()> {
+    let parent = uuid::Uuid::new_v4().to_string();
+
+    let paths = [
+        "x/", "x/x/", "x/x/x/", "x/x/x/x/", "x/x/x/y", "x/x/y", "x/y", "x/yy",
+    ];
+    for path in paths.iter() {
+        if path.ends_with('/') {
+            op.create_dir(&format!("{parent}/{path}"))?;
+        } else {
+            op.write(&format!("{parent}/{path}"), "test_scan")?;
+        }
+    }
     let w = op
         .lister_with(&format!("{parent}/x/"))
         .recursive(true)
         .call()?;
-    let actual = w
+    let mut actual = w
         .collect::<Vec<_>>()
         .into_iter()
         .map(|v| {
@@ -201,43 +232,81 @@ pub fn test_blocking_scan(op: BlockingOperator) -> Result<()> {
                 .unwrap()
                 .to_string()
         })
-        .collect::<HashSet<_>>();
-
-    debug!("walk top down: {:?}", actual);
-
-    assert!(actual.contains("x/y"));
-    assert!(actual.contains("x/x/y"));
-    assert!(actual.contains("x/x/x/y"));
+        .collect::<Vec<_>>();
+    actual.sort();
+    let expected = vec![
+        "x/x/", "x/x/x/", "x/x/x/x/", "x/x/x/y", "x/x/y", "x/y", "x/yy",
+    ];
+    assert_eq!(actual, expected);
     Ok(())
 }
 
-// Remove all should remove all in this path.
-pub fn test_blocking_remove_all(op: BlockingOperator) -> Result<()> {
+// Walk top down should output as expected
+// same as test_list_dir_with_recursive except listing 'x' instead of 'x/'
+pub fn test_blocking_list_dir_with_recursive_no_trailing_slash(op: BlockingOperator) -> 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/",
+    let paths = [
+        "x/", "x/x/", "x/x/x/", "x/x/x/x/", "x/x/x/y", "x/x/y", "x/y", "x/yy",
     ];
-
-    for path in expected.iter() {
+    for path in paths.iter() {
         if path.ends_with('/') {
             op.create_dir(&format!("{parent}/{path}"))?;
         } else {
             op.write(&format!("{parent}/{path}"), "test_scan")?;
         }
     }
+    let w = op
+        .lister_with(&format!("{parent}/x"))
+        .recursive(true)
+        .call()?;
+    let mut actual = w
+        .collect::<Vec<_>>()
+        .into_iter()
+        .map(|v| {
+            v.unwrap()
+                .path()
+                .strip_prefix(&format!("{parent}/"))
+                .unwrap()
+                .to_string()
+        })
+        .collect::<Vec<_>>();
+    actual.sort();
+    let expected = paths.to_vec();
+    assert_eq!(actual, expected);
+    Ok(())
+}
 
-    op.remove_all(&format!("{parent}/x/"))?;
+// Walk top down should output as expected
+// same as test_list_dir_with_recursive except listing 'x' instead of 'x/'
+pub fn test_blocking_list_file_with_recursive(op: BlockingOperator) -> Result<()> {
+    let parent = uuid::Uuid::new_v4().to_string();
 
-    for path in expected.iter() {
+    let paths = ["y", "yy"];
+    for path in paths.iter() {
         if path.ends_with('/') {
-            continue;
+            op.create_dir(&format!("{parent}/{path}"))?;
+        } else {
+            op.write(&format!("{parent}/{path}"), "test_scan")?;
         }
-        assert!(
-            !op.is_exist(&format!("{parent}/{path}"))?,
-            "{parent}/{path} should be removed"
-        )
     }
-
+    let w = op
+        .lister_with(&format!("{parent}/y"))
+        .recursive(true)
+        .call()?;
+    let mut actual = w
+        .collect::<Vec<_>>()
+        .into_iter()
+        .map(|v| {
+            v.unwrap()
+                .path()
+                .strip_prefix(&format!("{parent}/"))
+                .unwrap()
+                .to_string()
+        })
+        .collect::<Vec<_>>();
+    actual.sort();
+    let expected = vec!["yy"];
+    assert_eq!(actual, expected);
     Ok(())
 }