refactor: List must support list without recursive

bindings/c/include/opendal.h

@@ -635,10 +635,6 @@ typedef struct opendal_capability {
    * If backend supports list with start after.
    */
   bool list_with_start_after;
-  /**
-   * If backend support list with using slash as delimiter.
-   */
-  bool list_without_recursive;
   /**
    * If backend supports list without delimiter.
    */

bindings/c/src/operator_info.rs

@@ -116,8 +116,6 @@ pub struct opendal_capability {
     pub list_with_limit: bool,
     /// If backend supports list with start after.
     pub list_with_start_after: bool,
-    /// If backend support list with using slash as delimiter.
-    pub list_without_recursive: bool,
     /// If backend supports list without delimiter.
     pub list_with_recursive: bool,
 
@@ -266,7 +264,6 @@ impl From<core::Capability> for opendal_capability {
             list_with_limit: value.list_with_limit,
             list_with_start_after: value.list_with_start_after,
             list_with_recursive: value.list_with_recursive,
-            list_without_recursive: value.list_without_recursive,
             presign: value.presign,
             presign_read: value.presign_read,
             presign_stat: value.presign_stat,

bindings/java/src/lib.rs

@@ -153,7 +153,7 @@ fn make_operator_info<'a>(env: &mut JNIEnv<'a>, info: OperatorInfo) -> Result<JO
 fn make_capability<'a>(env: &mut JNIEnv<'a>, cap: Capability) -> Result<JObject<'a>> {
     let capability = env.new_object(
         "org/apache/opendal/Capability",
-        "(ZZZZZZZZZZZZZZZZZZJJJZZZZZZZZZZZZZZZJZ)V",
+        "(ZZZZZZZZZZZZZZZZZZJJJZZZZZZZZZZZZZZJZ)V",
         &[
             JValue::Bool(cap.stat as jboolean),
             JValue::Bool(cap.stat_with_if_match as jboolean),
@@ -183,7 +183,6 @@ fn make_capability<'a>(env: &mut JNIEnv<'a>, cap: Capability) -> Result<JObject<
             JValue::Bool(cap.list as jboolean),
             JValue::Bool(cap.list_with_limit as jboolean),
             JValue::Bool(cap.list_with_start_after as jboolean),
-            JValue::Bool(cap.list_without_recursive as jboolean),
             JValue::Bool(cap.list_with_recursive as jboolean),
             JValue::Bool(cap.presign as jboolean),
             JValue::Bool(cap.presign_read as jboolean),

bindings/nodejs/src/capability.rs

@@ -230,12 +230,6 @@ impl Capability {
         self.0.list_with_recursive
     }
 
-    /// If backend supports list without recursive.
-    #[napi(getter)]
-    pub fn list_without_recursive(&self) -> bool {
-        self.0.list_without_recursive
-    }
-
     /// If operator supports presign.
     #[napi(getter)]
     pub fn presign(&self) -> bool {

bindings/python/src/capability.rs

@@ -96,8 +96,6 @@ pub struct Capability {
     pub list_with_limit: bool,
     /// If backend supports list with start after.
     pub list_with_start_after: bool,
-    /// If backend support list with using slash as delimiter.
-    pub list_without_recursive: bool,
     /// If backend supports list without delimiter.
     pub list_with_recursive: bool,
 
@@ -155,7 +153,6 @@ impl Capability {
             list: capability.list,
             list_with_limit: capability.list_with_limit,
             list_with_start_after: capability.list_with_start_after,
-            list_without_recursive: capability.list_without_recursive,
             list_with_recursive: capability.list_with_recursive,
             presign: capability.presign,
             presign_read: capability.presign_read,

core/src/layers/complete.rs

@@ -29,7 +29,6 @@ use bytes::Bytes;
 
 use crate::raw::oio::FileReader;
 use crate::raw::oio::FlatLister;
-use crate::raw::oio::HierarchyLister;
 use crate::raw::oio::LazyReader;
 use crate::raw::oio::RangeReader;
 use crate::raw::oio::StreamableReader;
@@ -366,33 +365,18 @@ impl<A: Accessor> CompleteAccessor<A> {
 
         let recursive = args.recursive();
 
-        match (
-            recursive,
-            cap.list_with_recursive,
-            cap.list_without_recursive,
-        ) {
-            // - If service can both list_with_recursive and list_without_recursive
-            // - If recursive is true while services can list_with_recursive
-            // - If recursive is false while services can list_without_recursive
-            (_, true, true) | (true, true, _) | (false, _, true) => {
+        match (recursive, cap.list_with_recursive) {
+            // - If service can list_with_recursive
+            // - If recursive is false
+            (_, true) | (false, _) => {
                 let (rp, p) = self.inner.list(path, args).await?;
                 Ok((rp, CompleteLister::AlreadyComplete(p)))
             }
-            // If services can't list_with_recursive nor list_without_recursive.
-            //
-            // It should be a service level bug.
-            (_, false, false) => Err(self.new_unsupported_error(Operation::List)),
             // If recursive is true but service can't list_with_recursive
-            (true, false, true) => {
+            (true, false) => {
                 let p = FlatLister::new(self.inner.clone(), path);
                 Ok((RpList::default(), CompleteLister::NeedFlat(p)))
             }
-            // If recursive is false but service can't list_without_recursive
-            (false, true, false) => {
-                let (_, p) = self.inner.list(path, args.with_recursive(true)).await?;
-                let p = HierarchyLister::new(p, path);
-                Ok((RpList::default(), CompleteLister::NeedHierarchy(p)))
-            }
         }
     }
 
@@ -408,34 +392,18 @@ impl<A: Accessor> CompleteAccessor<A> {
 
         let recursive = args.recursive();
 
-        match (
-            recursive,
-            cap.list_with_recursive,
-            cap.list_without_recursive,
-        ) {
-            // - If service can both list_with_recursive and list_without_recursive
-            // - If recursive is true while services can list_with_recursive
-            // - If recursive is false while services can list_without_recursive
-            (_, true, true) | (true, true, _) | (false, _, true) => {
+        match (recursive, cap.list_with_recursive) {
+            // - If service can both list_with_recursive
+            // - If recursive is false
+            (_, true) | (false, _) => {
                 let (rp, p) = self.inner.blocking_list(path, args)?;
                 Ok((rp, CompleteLister::AlreadyComplete(p)))
             }
-            // If services can't list_with_recursive nor list_without_recursive.
-            //
-            // It should be a service level bug.
-            (_, false, false) => Err(self.new_unsupported_error(Operation::List)),
             // If recursive is true but service can't list_with_recursive
-            (true, false, true) => {
+            (true, false) => {
                 let p = FlatLister::new(self.inner.clone(), path);
                 Ok((RpList::default(), CompleteLister::NeedFlat(p)))
             }
-            // If recursive is false but service can't list_without_recursive
-            (false, true, false) => {
-                let (_, p) = self.inner.blocking_list(path, args.with_recursive(true))?;
-                let p: HierarchyLister<<A as Accessor>::BlockingLister> =
-                    HierarchyLister::new(p, path);
-                Ok((RpList::default(), CompleteLister::NeedHierarchy(p)))
-            }
         }
     }
 }
@@ -738,7 +706,6 @@ where
 pub enum CompleteLister<A: Accessor, P> {
     AlreadyComplete(P),
     NeedFlat(FlatLister<Arc<A>, P>),
-    NeedHierarchy(HierarchyLister<P>),
 }
 
 #[async_trait]
@@ -753,7 +720,6 @@ where
         match self {
             AlreadyComplete(p) => p.poll_next(cx),
             NeedFlat(p) => p.poll_next(cx),
-            NeedHierarchy(p) => p.poll_next(cx),
         }
     }
 }
@@ -769,7 +735,6 @@ where
         match self {
             AlreadyComplete(p) => p.next(),
             NeedFlat(p) => p.next(),
-            NeedHierarchy(p) => p.next(),
         }
     }
 }

core/src/layers/immutable_index.rs

@@ -156,7 +156,6 @@ impl<A: Accessor> LayeredAccessor for ImmutableIndexAccessor<A> {
         let cap = meta.full_capability_mut();
         cap.list = true;
         cap.list_with_recursive = true;
-        cap.list_without_recursive = true;
 
         meta
     }

core/src/layers/retry.rs

@@ -1173,7 +1173,6 @@ mod tests {
             am.set_native_capability(Capability {
                 read: true,
                 list: true,
-                list_without_recursive: true,
                 list_with_recursive: true,
                 batch: true,
                 ..Default::default()

core/src/raw/adapters/kv/backend.rs

@@ -28,6 +28,7 @@ use futures::future::BoxFuture;
 use futures::FutureExt;
 
 use super::Adapter;
+use crate::raw::oio::HierarchyLister;
 use crate::raw::*;
 use crate::*;
 
@@ -69,8 +70,8 @@ impl<S: Adapter> Accessor for Backend<S> {
     type BlockingReader = oio::Cursor;
     type Writer = KvWriter<S>;
     type BlockingWriter = KvWriter<S>;
-    type Lister = KvLister;
-    type BlockingLister = KvLister;
+    type Lister = HierarchyLister<KvLister>;
+    type BlockingLister = HierarchyLister<KvLister>;
 
     fn info(&self) -> AccessorInfo {
         let mut am: AccessorInfo = self.kv.metadata().into();
@@ -89,14 +90,6 @@ impl<S: Adapter> Accessor for Backend<S> {
             cap.delete = true;
         }
 
-        if cap.read && cap.write {
-            cap.copy = true;
-        }
-
-        if cap.read && cap.write && cap.delete {
-            cap.rename = true;
-        }
-
         if cap.list {
             cap.list_with_recursive = true;
         }
@@ -193,6 +186,7 @@ impl<S: Adapter> Accessor for Backend<S> {
         let p = build_abs_path(&self.root, path);
         let res = self.kv.scan(&p).await?;
         let lister = KvLister::new(&self.root, res);
+        let lister = HierarchyLister::new(lister, path);
 
         Ok((RpList::default(), lister))
     }
@@ -201,71 +195,10 @@ impl<S: Adapter> Accessor for Backend<S> {
         let p = build_abs_path(&self.root, path);
         let res = self.kv.blocking_scan(&p)?;
         let lister = KvLister::new(&self.root, res);
+        let lister = HierarchyLister::new(lister, path);
 
         Ok((RpList::default(), lister))
     }
-
-    async fn rename(&self, from: &str, to: &str, _: OpRename) -> Result<RpRename> {
-        let from = build_abs_path(&self.root, from);
-        let to = build_abs_path(&self.root, to);
-
-        let bs = match self.kv.get(&from).await? {
-            // TODO: we can reuse the metadata in value to build content range.
-            Some(bs) => bs,
-            None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
-        };
-
-        self.kv.set(&to, &bs).await?;
-
-        self.kv.delete(&from).await?;
-        Ok(RpRename::default())
-    }
-
-    fn blocking_rename(&self, from: &str, to: &str, _: OpRename) -> Result<RpRename> {
-        let from = build_abs_path(&self.root, from);
-        let to = build_abs_path(&self.root, to);
-
-        let bs = match self.kv.blocking_get(&from)? {
-            // TODO: we can reuse the metadata in value to build content range.
-            Some(bs) => bs,
-            None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
-        };
-
-        self.kv.blocking_set(&to, &bs)?;
-
-        self.kv.blocking_delete(&from)?;
-        Ok(RpRename::default())
-    }
-
-    async fn copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
-        let from = build_abs_path(&self.root, from);
-        let to = build_abs_path(&self.root, to);
-
-        let bs = match self.kv.get(&from).await? {
-            // TODO: we can reuse the metadata in value to build content range.
-            Some(bs) => bs,
-            None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
-        };
-
-        self.kv.set(&to, &bs).await?;
-
-        Ok(RpCopy::default())
-    }
-
-    fn blocking_copy(&self, from: &str, to: &str, _: OpCopy) -> Result<RpCopy> {
-        let from = build_abs_path(&self.root, from);
-        let to = build_abs_path(&self.root, to);
-
-        let bs = match self.kv.blocking_get(&from)? {
-            // TODO: we can reuse the metadata in value to build content range.
-            Some(bs) => bs,
-            None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
-        };
-
-        self.kv.blocking_set(&to, &bs)?;
-
-        Ok(RpCopy::default())
-    }
 }
 
 impl<S> Backend<S>