feat(services/oss): Add start-after support for oss list (#3410)

core/src/services/oss/backend.rs

@@ -424,6 +424,8 @@ impl Accessor for OssBackend {
                 copy: true,
 
                 list: true,
+                list_with_limit: true,
+                list_with_start_after: true,
                 list_with_delimiter_slash: true,
                 list_without_delimiter: true,
 
@@ -544,7 +546,13 @@ impl Accessor for OssBackend {
     async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Pager)> {
         Ok((
             RpList::default(),
-            OssPager::new(self.core.clone(), path, args.delimiter(), args.limit()),
+            OssPager::new(
+                self.core.clone(),
+                path,
+                args.delimiter(),
+                args.limit(),
+                args.start_after(),
+            ),
         ))
     }
 

core/src/services/oss/core.rs

@@ -15,10 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::fmt::Debug;
-use std::fmt::Formatter;
-use std::time::Duration;
-
 use bytes::Bytes;
 use http::header::CACHE_CONTROL;
 use http::header::CONTENT_DISPOSITION;
@@ -36,6 +32,10 @@ use reqsign::AliyunLoader;
 use reqsign::AliyunOssSigner;
 use serde::Deserialize;
 use serde::Serialize;
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::fmt::Write;
+use std::time::Duration;
 
 use crate::raw::*;
 use crate::*;
@@ -327,19 +327,41 @@ impl OssCore {
         token: Option<&str>,
         delimiter: &str,
         limit: Option<usize>,
+        start_after: Option<String>,
     ) -> Result<Request<AsyncBody>> {
         let p = build_abs_path(&self.root, path);
 
         let endpoint = self.get_endpoint(false);
-        let url = format!(
-            "{}/?list-type=2&delimiter={delimiter}&prefix={}{}{}",
-            endpoint,
-            percent_encode_path(&p),
-            limit.map(|t| format!("&max-keys={t}")).unwrap_or_default(),
-            token
-                .map(|t| format!("&continuation-token={}", percent_encode_path(t)))
-                .unwrap_or_default(),
-        );
+        let mut url = format!("{}/?list-type=2", endpoint);
+
+        write!(url, "&delimiter={delimiter}").expect("write into string must succeed");
+        // prefix
+        if !p.is_empty() {
+            write!(url, "&prefix={}", percent_encode_path(&p))
+                .expect("write into string must succeed");
+        }
+
+        // max-key
+        if let Some(limit) = limit {
+            write!(url, "&max-keys={limit}").expect("write into string must succeed");
+        }
+
+        // continuation_token
+        if let Some(continuation_token) = token {
+            write!(
+                url,
+                "&continuation-token={}",
+                percent_encode_path(continuation_token)
+            )
+            .expect("write into string must succeed");
+        }
+
+        // start-after
+        if let Some(start_after) = start_after {
+            let start_after = build_abs_path(&self.root, &start_after);
+            write!(url, "&start-after={}", percent_encode_path(&start_after))
+                .expect("write into string must succeed");
+        }
 
         let req = Request::get(&url)
             .body(AsyncBody::Empty)
@@ -427,8 +449,9 @@ impl OssCore {
         token: Option<&str>,
         delimiter: &str,
         limit: Option<usize>,
+        start_after: Option<String>,
     ) -> Result<Response<IncomingAsyncBody>> {
-        let mut req = self.oss_list_object_request(path, token, delimiter, limit)?;
+        let mut req = self.oss_list_object_request(path, token, delimiter, limit, start_after)?;
 
         self.sign(&mut req).await?;
         self.send(req).await

core/src/services/oss/pager.rs

@@ -38,19 +38,28 @@ pub struct OssPager {
     path: String,
     delimiter: String,
     limit: Option<usize>,
+    /// Filter results to objects whose names are lexicographically
+    /// **equal to or after** startOffset
+    start_after: Option<String>,
 
     token: Option<String>,
     done: bool,
 }
 
 impl OssPager {
-    pub fn new(core: Arc<OssCore>, path: &str, delimiter: &str, limit: Option<usize>) -> Self {
+    pub fn new(
+        core: Arc<OssCore>,
+        path: &str,
+        delimiter: &str,
+        limit: Option<usize>,
+        start_after: Option<&str>,
+    ) -> Self {
         Self {
             core,
             path: path.to_string(),
             delimiter: delimiter.to_string(),
             limit,
-
+            start_after: start_after.map(String::from),
             token: None,
 
             done: false,
@@ -72,6 +81,7 @@ impl oio::Page for OssPager {
                 self.token.as_deref(),
                 &self.delimiter,
                 self.limit,
+                self.start_after.clone(),
             )
             .await?;
 
@@ -101,6 +111,12 @@ impl oio::Page for OssPager {
             if object.key.ends_with('/') {
                 continue;
             }
+
+            // exclude the inclusive start_after itself
+            let path = &build_rel_path(&self.core.root, &object.key);
+            if self.start_after.as_ref() == Some(path) {
+                continue;
+            }
             let mut meta = Metadata::new(EntryMode::FILE);
 
             meta.set_etag(&object.etag);