feat(services): reduce list fallback stat for local kv backends

core/services/compfs/src/lister.rs

@@ -52,23 +52,37 @@ fn normalize(path: &Path, root: &Path) -> String {
 }
 
 fn next_entry(read_dir: &mut ReadDir, root: &Path) -> std::io::Result<Option<oio::Entry>> {
-    let Some(entry) = read_dir.next().transpose()? else {
-        return Ok(None);
-    };
-    let path = entry.path();
-    let rel_path = normalize(&path, root);
+    loop {
+        let Some(entry) = read_dir.next().transpose()? else {
+            return Ok(None);
+        };
+        let path = entry.path();
+        let rel_path = normalize(&path, root);
 
-    let file_type = entry.file_type()?;
+        let file_type = match entry.file_type() {
+            Ok(file_type) => file_type,
+            Err(err) if err.kind() == std::io::ErrorKind::NotFound => continue,
+            Err(err) => return Err(err),
+        };
 
-    let entry = if file_type.is_file() {
-        oio::Entry::new(&rel_path, Metadata::new(EntryMode::FILE))
-    } else if file_type.is_dir() {
-        oio::Entry::new(&format!("{rel_path}/"), Metadata::new(EntryMode::DIR))
-    } else {
-        oio::Entry::new(&rel_path, Metadata::new(EntryMode::Unknown))
-    };
+        let entry = if file_type.is_file() {
+            let de_metadata = match entry.metadata() {
+                Ok(de_metadata) => de_metadata,
+                Err(err) if err.kind() == std::io::ErrorKind::NotFound => continue,
+                Err(err) => return Err(err),
+            };
+            oio::Entry::new(
+                &rel_path,
+                Metadata::new(EntryMode::FILE).with_content_length(de_metadata.len()),
+            )
+        } else if file_type.is_dir() {
+            oio::Entry::new(&format!("{rel_path}/"), Metadata::new(EntryMode::DIR))
+        } else {
+            oio::Entry::new(&rel_path, Metadata::new(EntryMode::Unknown))
+        };
 
-    Ok(Some(entry))
+        return Ok(Some(entry));
+    }
 }
 
 impl oio::List for CompfsLister {

core/services/dashmap/src/lister.rs

@@ -26,12 +26,16 @@ use super::core::DashmapCore;
 pub struct DashmapLister {
     root: String,
     path: String,
-    iter: IntoIter<String>,
+    iter: IntoIter<(String, Metadata)>,
 }
 
 impl DashmapLister {
     pub fn new(core: Arc<DashmapCore>, root: String, path: String) -> Self {
-        let entries: Vec<_> = core.cache.iter().map(|item| item.key().clone()).collect();
+        let entries: Vec<_> = core
+            .cache
+            .iter()
+            .map(|item| (item.key().clone(), item.value().metadata.clone()))
+            .collect();
         let path = build_abs_path(&root, &path);
 
         Self {
@@ -44,17 +48,10 @@ impl DashmapLister {
 
 impl oio::List for DashmapLister {
     async fn next(&mut self) -> Result<Option<oio::Entry>> {
-        for key in self.iter.by_ref() {
+        for (key, metadata) in self.iter.by_ref() {
             if key.starts_with(&self.path) {
                 let path = build_rel_path(&self.root, &key);
-
-                // Determine if it's a file or directory based on trailing slash
-                let mode = if key.ends_with('/') {
-                    EntryMode::DIR
-                } else {
-                    EntryMode::FILE
-                };
-                let entry = oio::Entry::new(&path, Metadata::new(mode));
+                let entry = oio::Entry::new(&path, metadata);
                 return Ok(Some(entry));
             }
         }

core/services/mini_moka/src/lister.rs

@@ -25,41 +25,30 @@ use super::core::MiniMokaCore;
 
 pub struct MiniMokaLister {
     root: String,
-    keys: IntoIter<String>,
+    entries: IntoIter<(String, Metadata)>,
 }
 
 impl MiniMokaLister {
     pub fn new(core: Arc<MiniMokaCore>, root: String, _path: String) -> Self {
-        // Get all keys from the cache
-        let keys: Vec<String> = core
+        let entries: Vec<(String, Metadata)> = core
             .cache
             .iter()
-            .map(|entry| entry.key().to_string())
+            .map(|entry| (entry.key().to_string(), entry.value().metadata.clone()))
             .collect();
 
         Self {
             root,
-            keys: keys.into_iter(),
+            entries: entries.into_iter(),
         }
     }
 }
 
 impl oio::List for MiniMokaLister {
     async fn next(&mut self) -> Result<Option<oio::Entry>> {
-        match self.keys.next() {
-            Some(key) => {
-                // Convert absolute path to relative path
+        match self.entries.next() {
+            Some((key, metadata)) => {
                 let rel_path = build_rel_path(&self.root, &key);
 
-                // Determine if it's a file or directory based on trailing slash
-                let mode = if key.ends_with('/') {
-                    EntryMode::DIR
-                } else {
-                    EntryMode::FILE
-                };
-
-                let metadata = Metadata::new(mode);
-
                 Ok(Some(oio::Entry::new(&rel_path, metadata)))
             }
             None => Ok(None),

core/services/moka/src/lister.rs

@@ -25,39 +25,34 @@ use opendal_core::*;
 
 pub struct MokaLister {
     root: String,
-    keys: IntoIter<String>,
+    entries: IntoIter<(String, Metadata)>,
 }
 
 impl MokaLister {
     pub fn new(core: Arc<MokaCore>, root: String, _path: String) -> Self {
-        // Get all keys from the cache
-        let keys: Vec<String> = core.cache.iter().map(|kv| kv.0.to_string()).collect();
+        let entries: Vec<(String, Metadata)> = core
+            .cache
+            .iter()
+            .map(|(key, value)| (key.to_string(), value.metadata.clone()))
+            .collect();
 
         Self {
             root,
-            keys: keys.into_iter(),
+            entries: entries.into_iter(),
         }
     }
 }
 
 impl oio::List for MokaLister {
     async fn next(&mut self) -> Result<Option<oio::Entry>> {
-        match self.keys.next() {
-            Some(key) => {
-                // Convert absolute path to relative path
+        match self.entries.next() {
+            Some((key, metadata)) => {
                 let mut path = build_rel_path(&self.root, &key);
                 if path.is_empty() {
                     path = "/".to_string();
                 }
 
-                // Determine if it's a file or directory based on trailing slash
-                let mode = if key.ends_with('/') {
-                    EntryMode::DIR
-                } else {
-                    EntryMode::FILE
-                };
-
-                Ok(Some(oio::Entry::new(&path, Metadata::new(mode))))
+                Ok(Some(oio::Entry::new(&path, metadata)))
             }
             None => Ok(None),
         }

core/services/rocksdb/src/core.rs

@@ -51,17 +51,18 @@ impl RocksdbCore {
         self.db.delete(path).map_err(parse_rocksdb_error)
     }
 
-    pub fn list(&self, path: &str) -> Result<Vec<String>> {
-        let it = self.db.prefix_iterator(path).map(|r| r.map(|(k, _)| k));
+    pub fn list(&self, path: &str) -> Result<Vec<(String, u64)>> {
+        let it = self.db.prefix_iterator(path);
         let mut res = Vec::default();
 
-        for key in it {
-            let key = key.map_err(parse_rocksdb_error)?;
+        for entry in it {
+            let (key, value) = entry.map_err(parse_rocksdb_error)?;
             let key = String::from_utf8_lossy(&key);
             if !key.starts_with(path) {
                 break;
             }
-            res.push(key.to_string());
+
+            res.push((key.to_string(), value.len() as u64));
         }
 
         Ok(res)

core/services/rocksdb/src/lister.rs

@@ -25,7 +25,7 @@ use super::core::*;
 
 pub struct RocksdbLister {
     root: String,
-    iter: IntoIter<String>,
+    iter: IntoIter<(String, u64)>,
 }
 
 impl RocksdbLister {
@@ -41,7 +41,7 @@ impl RocksdbLister {
 
 impl oio::List for RocksdbLister {
     async fn next(&mut self) -> Result<Option<oio::Entry>> {
-        if let Some(key) = self.iter.next() {
+        if let Some((key, value_len)) = self.iter.next() {
             let path = build_rel_path(&self.root, &key);
 
             // Determine if it's a file or directory based on trailing slash
@@ -50,7 +50,11 @@ impl oio::List for RocksdbLister {
             } else {
                 EntryMode::FILE
             };
-            let entry = oio::Entry::new(&path, Metadata::new(mode));
+            let mut metadata = Metadata::new(mode);
+            if metadata.mode().is_file() {
+                metadata.set_content_length(value_len);
+            }
+            let entry = oio::Entry::new(&path, metadata);
             return Ok(Some(entry));
         }
 

core/services/sled/src/core.rs

@@ -51,18 +51,19 @@ impl SledCore {
         Ok(())
     }
 
-    pub fn list(&self, path: &str) -> Result<Vec<String>> {
-        let it = self.tree.scan_prefix(path).keys();
+    pub fn list(&self, path: &str) -> Result<Vec<(String, u64)>> {
+        let it = self.tree.scan_prefix(path);
         let mut res = Vec::default();
 
         for i in it {
-            let bs = i.map_err(parse_error)?.to_vec();
+            let (key, value) = i.map_err(parse_error)?;
+            let bs = key.to_vec();
             let v = String::from_utf8(bs).map_err(|err| {
                 Error::new(ErrorKind::Unexpected, "store key is not valid utf-8 string")
                     .set_source(err)
             })?;
 
-            res.push(v);
+            res.push((v, value.len() as u64));
         }
 
         Ok(res)

core/services/sled/src/lister.rs

@@ -25,7 +25,7 @@ use opendal_core::*;
 
 pub struct SledLister {
     root: String,
-    iter: IntoIter<String>,
+    iter: IntoIter<(String, u64)>,
 }
 
 impl SledLister {
@@ -41,7 +41,7 @@ impl SledLister {
 
 impl oio::List for SledLister {
     async fn next(&mut self) -> Result<Option<oio::Entry>> {
-        if let Some(key) = self.iter.next() {
+        if let Some((key, value_len)) = self.iter.next() {
             let path = build_rel_path(&self.root, &key);
 
             // Determine if it's a file or directory based on trailing slash
@@ -50,7 +50,11 @@ impl oio::List for SledLister {
             } else {
                 EntryMode::FILE
             };
-            let entry = oio::Entry::new(&path, Metadata::new(mode));
+            let mut metadata = Metadata::new(mode);
+            if metadata.mode().is_file() {
+                metadata.set_content_length(value_len);
+            }
+            let entry = oio::Entry::new(&path, metadata);
             return Ok(Some(entry));
         }