You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2023/10/30 07:58:50 UTC
(incubator-opendal) 01/03: Add size hint for rp read
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch lazy-reader
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
commit caabbe67d97081b1603b8ec3550191f53ffd109f
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Oct 30 15:06:25 2023 +0800
Add size hint for rp read
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/raw/rps.rs | 25 ++++++++++++++++++++++++-
core/src/services/azblob/backend.rs | 5 ++++-
core/src/services/azdls/backend.rs | 5 ++++-
core/src/services/azfile/backend.rs | 5 ++++-
core/src/services/cos/backend.rs | 5 ++++-
core/src/services/gcs/backend.rs | 3 ++-
core/src/services/gdrive/backend.rs | 5 ++++-
core/src/services/ghac/backend.rs | 5 ++++-
core/src/services/http/backend.rs | 5 ++++-
core/src/services/obs/backend.rs | 5 ++++-
core/src/services/onedrive/backend.rs | 6 ++++--
core/src/services/oss/backend.rs | 5 ++++-
core/src/services/s3/backend.rs | 5 ++++-
core/src/services/webdav/backend.rs | 5 ++++-
core/src/services/webhdfs/backend.rs | 5 ++++-
15 files changed, 78 insertions(+), 16 deletions(-)
diff --git a/core/src/raw/rps.rs b/core/src/raw/rps.rs
index 10b3c7159..4cba4b8cc 100644
--- a/core/src/raw/rps.rs
+++ b/core/src/raw/rps.rs
@@ -98,13 +98,36 @@ impl<T: Default> From<PresignedRequest> for Request<T> {
/// Reply for `read` operation.
#[derive(Debug, Clone, Default)]
-pub struct RpRead {}
+pub struct RpRead {
+ /// Size is the size of the reader returned by this read operation.
+ ///
+ /// - `Some(size)` means the reader has at most size bytes.
+ /// - `None` means the reader has unknown size.
+ ///
+ /// It's ok to leave size as empty, but it's recommended to set size if possible. We will use
+ /// this size as hint to do some optimization like avoid an extra stat or read.
+ size: Option<u64>,
+}
impl RpRead {
/// Create a new reply for `read`.
pub fn new() -> Self {
RpRead::default()
}
+
+ /// Got the size of the reader returned by this read operation.
+ ///
+ /// - `Some(size)` means the reader has at most size bytes.
+ /// - `None` means the reader has unknown size.
+ pub fn size(&self) -> Option<u64> {
+ self.size
+ }
+
+ /// Set the size of the reader returned by this read operation.
+ pub fn with_size(mut self, size: Option<u64>) -> Self {
+ self.size = size;
+ self
+ }
}
/// Reply for `batch` operation.
diff --git a/core/src/services/azblob/backend.rs b/core/src/services/azblob/backend.rs
index 6e8dd69ee..6e46e1b35 100644
--- a/core/src/services/azblob/backend.rs
+++ b/core/src/services/azblob/backend.rs
@@ -586,7 +586,10 @@ impl Accessor for AzblobBackend {
let status = resp.status();
match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ }
StatusCode::RANGE_NOT_SATISFIABLE => Ok((RpRead::new(), IncomingAsyncBody::empty())),
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/azdls/backend.rs b/core/src/services/azdls/backend.rs
index 435d6a574..ad03274a2 100644
--- a/core/src/services/azdls/backend.rs
+++ b/core/src/services/azdls/backend.rs
@@ -292,7 +292,10 @@ impl Accessor for AzdlsBackend {
let status = resp.status();
match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ }
StatusCode::RANGE_NOT_SATISFIABLE => Ok((RpRead::new(), IncomingAsyncBody::empty())),
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/azfile/backend.rs b/core/src/services/azfile/backend.rs
index 256243624..106ec405d 100644
--- a/core/src/services/azfile/backend.rs
+++ b/core/src/services/azfile/backend.rs
@@ -316,7 +316,10 @@ impl Accessor for AzfileBackend {
let status = resp.status();
match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ }
StatusCode::RANGE_NOT_SATISFIABLE => Ok((RpRead::new(), IncomingAsyncBody::empty())),
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/cos/backend.rs b/core/src/services/cos/backend.rs
index aef237a1a..e5a521718 100644
--- a/core/src/services/cos/backend.rs
+++ b/core/src/services/cos/backend.rs
@@ -332,7 +332,10 @@ impl Accessor for CosBackend {
let status = resp.status();
match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ }
StatusCode::RANGE_NOT_SATISFIABLE => Ok((RpRead::new(), IncomingAsyncBody::empty())),
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/gcs/backend.rs b/core/src/services/gcs/backend.rs
index 40f7f6a9b..0522a8d5c 100644
--- a/core/src/services/gcs/backend.rs
+++ b/core/src/services/gcs/backend.rs
@@ -389,7 +389,8 @@ impl Accessor for GcsBackend {
let resp = self.core.gcs_get_object(path, &args).await?;
if resp.status().is_success() {
- Ok((RpRead::new(), resp.into_body()))
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
} else if resp.status() == StatusCode::RANGE_NOT_SATISFIABLE {
Ok((RpRead::new(), IncomingAsyncBody::empty()))
} else {
diff --git a/core/src/services/gdrive/backend.rs b/core/src/services/gdrive/backend.rs
index 53ab999ea..4e1274d21 100644
--- a/core/src/services/gdrive/backend.rs
+++ b/core/src/services/gdrive/backend.rs
@@ -123,7 +123,10 @@ impl Accessor for GdriveBackend {
let status = resp.status();
match status {
- StatusCode::OK => Ok((RpRead::new(), resp.into_body())),
+ StatusCode::OK => {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ }
_ => Err(parse_error(resp).await?),
}
}
diff --git a/core/src/services/ghac/backend.rs b/core/src/services/ghac/backend.rs
index 5b9c7c27d..425cee86b 100644
--- a/core/src/services/ghac/backend.rs
+++ b/core/src/services/ghac/backend.rs
@@ -350,7 +350,10 @@ impl Accessor for GhacBackend {
let status = resp.status();
match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ }
StatusCode::RANGE_NOT_SATISFIABLE => Ok((RpRead::new(), IncomingAsyncBody::empty())),
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/http/backend.rs b/core/src/services/http/backend.rs
index 6b2388862..233237f87 100644
--- a/core/src/services/http/backend.rs
+++ b/core/src/services/http/backend.rs
@@ -237,7 +237,10 @@ impl Accessor for HttpBackend {
let status = resp.status();
match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ }
StatusCode::RANGE_NOT_SATISFIABLE => Ok((RpRead::new(), IncomingAsyncBody::empty())),
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/obs/backend.rs b/core/src/services/obs/backend.rs
index 98e51b9db..311af30a6 100644
--- a/core/src/services/obs/backend.rs
+++ b/core/src/services/obs/backend.rs
@@ -359,7 +359,10 @@ impl Accessor for ObsBackend {
let status = resp.status();
match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ }
StatusCode::RANGE_NOT_SATISFIABLE => Ok((RpRead::new(), IncomingAsyncBody::empty())),
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/onedrive/backend.rs b/core/src/services/onedrive/backend.rs
index 8b1959b9b..95e052654 100644
--- a/core/src/services/onedrive/backend.rs
+++ b/core/src/services/onedrive/backend.rs
@@ -93,8 +93,10 @@ impl Accessor for OnedriveBackend {
let status = resp.status();
match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
-
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ }
_ => Err(parse_error(resp).await?),
}
}
diff --git a/core/src/services/oss/backend.rs b/core/src/services/oss/backend.rs
index bf39396ba..93323b3a3 100644
--- a/core/src/services/oss/backend.rs
+++ b/core/src/services/oss/backend.rs
@@ -474,7 +474,10 @@ impl Accessor for OssBackend {
let status = resp.status();
match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ }
StatusCode::RANGE_NOT_SATISFIABLE => Ok((RpRead::new(), IncomingAsyncBody::empty())),
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index eeadf18d7..f2157c5dc 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -969,7 +969,10 @@ impl Accessor for S3Backend {
let status = resp.status();
match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ }
StatusCode::RANGE_NOT_SATISFIABLE => Ok((RpRead::new(), IncomingAsyncBody::empty())),
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/webdav/backend.rs b/core/src/services/webdav/backend.rs
index 1e0e3b76e..93e0a70ca 100644
--- a/core/src/services/webdav/backend.rs
+++ b/core/src/services/webdav/backend.rs
@@ -267,7 +267,10 @@ impl Accessor for WebdavBackend {
let resp = self.webdav_get(path, args).await?;
let status = resp.status();
match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ }
StatusCode::RANGE_NOT_SATISFIABLE => Ok((RpRead::new(), IncomingAsyncBody::empty())),
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/webhdfs/backend.rs b/core/src/services/webhdfs/backend.rs
index 165a9d363..6c11ebc33 100644
--- a/core/src/services/webhdfs/backend.rs
+++ b/core/src/services/webhdfs/backend.rs
@@ -465,7 +465,10 @@ impl Accessor for WebhdfsBackend {
let range = args.range();
let resp = self.webhdfs_read_file(path, range).await?;
match resp.status() {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let size = parse_content_length(resp.headers())?;
+ Ok((RpRead::new().with_size(size), resp.into_body()))
+ }
StatusCode::RANGE_NOT_SATISFIABLE => Ok((RpRead::new(), IncomingAsyncBody::empty())),
_ => Err(parse_error(resp).await?),
}