[incubator-opendal] branch main updated: feat: add max_batch_operations for AccessorInfo (#1615)

[xu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git


The following commit(s) were added to refs/heads/main by this push:
     new 0cff2537 feat: add max_batch_operations for AccessorInfo (#1615)
0cff2537 is described below

commit 0cff253730e0e3e8998eca0b900c49ccdc6a5203
Author: ClSlaid <ca...@bupt.edu.cn>
AuthorDate: Thu Mar 16 13:59:36 2023 +0800

    feat: add max_batch_operations for AccessorInfo (#1615)
    
    * faet: add batch limit for AccessorInfo
    
    Signed-off-by: ClSlaid <ca...@bupt.edu.cn>
    
    * refactor: typo improve
    
    1. naming changes
    2. change visibility of max_batch_operations
    
    Signed-off-by: ClSlaid <ca...@bupt.edu.cn>
    
    ---------
    
    Signed-off-by: ClSlaid <ca...@bupt.edu.cn>
---
 src/raw/accessor.rs                     | 19 +++++++++++++++++++
 src/services/oss/backend.rs             |  1 +
 src/services/s3/backend.rs              |  1 +
 src/types/operator/blocking_operator.rs | 10 ++++++----
 4 files changed, 27 insertions(+), 4 deletions(-)

diff --git a/src/raw/accessor.rs b/src/raw/accessor.rs
index c374e1c5..e3564ebd 100644
--- a/src/raw/accessor.rs
+++ b/src/raw/accessor.rs
@@ -337,6 +337,7 @@ impl Accessor for () {
             scheme: Scheme::Custom("dummy"),
             root: "".to_string(),
             name: "dummy".to_string(),
+            max_batch_operations: None,
             capabilities: None.into(),
             hints: None.into(),
         }
@@ -431,6 +432,9 @@ pub struct AccessorInfo {
     scheme: Scheme,
     root: String,
     name: String,
+    /// limit of batch operation
+    /// only meaningful when accessor supports batch operation
+    max_batch_operations: Option<usize>,
     capabilities: FlagSet<AccessorCapability>,
     hints: FlagSet<AccessorHint>,
 }
@@ -476,6 +480,21 @@ impl AccessorInfo {
         self
     }
 
+    /// backend's number limitation of operations in a single batch.
+    ///
+    /// # Note
+    /// - Got Some(x): limitation is x
+    /// - Got None: no limitation
+    pub(crate) fn max_batch_operations(&self) -> Option<usize> {
+        self.max_batch_operations
+    }
+
+    /// Set batch size limit for backend.
+    pub(crate) fn set_max_batch_operations(&mut self, limit: usize) -> &mut Self {
+        self.max_batch_operations = Some(limit);
+        self
+    }
+
     /// Get backend's capabilities.
     pub fn capabilities(&self) -> FlagSet<AccessorCapability> {
         self.capabilities
diff --git a/src/services/oss/backend.rs b/src/services/oss/backend.rs
index 8164aed7..efc433e8 100644
--- a/src/services/oss/backend.rs
+++ b/src/services/oss/backend.rs
@@ -414,6 +414,7 @@ impl Accessor for OssBackend {
         am.set_scheme(Scheme::Oss)
             .set_root(&self.root)
             .set_name(&self.bucket)
+            .set_max_batch_operations(1000)
             .set_capabilities(Read | Write | List | Scan | Presign | Batch)
             .set_hints(ReadStreamable);
 
diff --git a/src/services/s3/backend.rs b/src/services/s3/backend.rs
index 6a63492f..2cdc2b74 100644
--- a/src/services/s3/backend.rs
+++ b/src/services/s3/backend.rs
@@ -1121,6 +1121,7 @@ impl Accessor for S3Backend {
         am.set_scheme(Scheme::S3)
             .set_root(&self.root)
             .set_name(&self.bucket)
+            .set_max_batch_operations(1000)
             .set_capabilities(Read | Write | List | Scan | Presign | Batch)
             .set_hints(ReadStreamable);
 
diff --git a/src/types/operator/blocking_operator.rs b/src/types/operator/blocking_operator.rs
index 1374752f..0238c314 100644
--- a/src/types/operator/blocking_operator.rs
+++ b/src/types/operator/blocking_operator.rs
@@ -65,11 +65,13 @@ impl BlockingOperator {
         &self.accessor
     }
 
+    /// create a new blocking operator from inner accessor.
+    ///
+    /// # Note
+    /// default batch limit is 1000.
     pub(crate) fn from_inner(accessor: FusedAccessor) -> Self {
-        Self {
-            accessor,
-            limit: 1000,
-        }
+        let limit = accessor.info().max_batch_operations().unwrap_or(1000);
+        Self { accessor, limit }
     }
 
     /// Get current operator's limit