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 2024/05/08 07:44:43 UTC
(opendal) branch main updated: feat: add checksums to MultiPartComplete (#4580)
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/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 0321e9dcb2 feat: add checksums to MultiPartComplete (#4580)
0321e9dcb2 is described below
commit 0321e9dcb2c6da2dada891557ee634395de85647
Author: JWackerbauer <ja...@gmail.com>
AuthorDate: Wed May 8 09:44:37 2024 +0200
feat: add checksums to MultiPartComplete (#4580)
* add checksums to MultiPartComplete
* Fix formatting
* Attempt to fix binding checks
* Fix warning
* Fix docs
* Fix formatting
* Fix serialization
* Fix formatting
---------
Co-authored-by: Jan Wackerbauer <ja...@compax.at>
---
core/src/raw/oio/write/multipart_write.rs | 4 +++
core/src/services/b2/writer.rs | 1 +
core/src/services/cos/writer.rs | 6 ++++-
core/src/services/obs/writer.rs | 6 ++++-
core/src/services/oss/writer.rs | 6 ++++-
core/src/services/s3/core.rs | 41 ++++++++++++++++++++-----------
core/src/services/s3/writer.rs | 35 ++++++++++++++++++++------
core/src/services/upyun/writer.rs | 1 +
core/src/services/vercel_blob/writer.rs | 1 +
9 files changed, 77 insertions(+), 24 deletions(-)
diff --git a/core/src/raw/oio/write/multipart_write.rs b/core/src/raw/oio/write/multipart_write.rs
index 64c993fa7d..0582553246 100644
--- a/core/src/raw/oio/write/multipart_write.rs
+++ b/core/src/raw/oio/write/multipart_write.rs
@@ -108,12 +108,15 @@ pub trait MultipartWrite: Send + Sync + Unpin + 'static {
///
/// - `part_number` is the index of the part, starting from 0.
/// - `etag` is the `ETag` of the part.
+/// - `checksum` is the optional checksum of the part.
#[derive(Clone)]
pub struct MultipartPart {
/// The number of the part, starting from 0.
pub part_number: usize,
/// The etag of the part.
pub etag: String,
+ /// The checksum of the part.
+ pub checksum: Option<String>,
}
/// WritePartResult is the result returned by [`WritePartFuture`].
@@ -381,6 +384,7 @@ mod tests {
Ok(MultipartPart {
part_number,
etag: "etag".to_string(),
+ checksum: None,
})
}
diff --git a/core/src/services/b2/writer.rs b/core/src/services/b2/writer.rs
index ef39b85e3c..f1741fd013 100644
--- a/core/src/services/b2/writer.rs
+++ b/core/src/services/b2/writer.rs
@@ -106,6 +106,7 @@ impl oio::MultipartWrite for B2Writer {
Ok(oio::MultipartPart {
etag: result.content_sha1,
part_number,
+ checksum: None,
})
}
_ => Err(parse_error(resp).await?),
diff --git a/core/src/services/cos/writer.rs b/core/src/services/cos/writer.rs
index c40ba783e6..2cc053f24e 100644
--- a/core/src/services/cos/writer.rs
+++ b/core/src/services/cos/writer.rs
@@ -111,7 +111,11 @@ impl oio::MultipartWrite for CosWriter {
})?
.to_string();
- Ok(oio::MultipartPart { part_number, etag })
+ Ok(oio::MultipartPart {
+ part_number,
+ etag,
+ checksum: None,
+ })
}
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/obs/writer.rs b/core/src/services/obs/writer.rs
index 50355d0d43..e07309aca5 100644
--- a/core/src/services/obs/writer.rs
+++ b/core/src/services/obs/writer.rs
@@ -112,7 +112,11 @@ impl oio::MultipartWrite for ObsWriter {
})?
.to_string();
- Ok(MultipartPart { part_number, etag })
+ Ok(MultipartPart {
+ part_number,
+ etag,
+ checksum: None,
+ })
}
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/oss/writer.rs b/core/src/services/oss/writer.rs
index a9a6cc2b31..224db8e4c0 100644
--- a/core/src/services/oss/writer.rs
+++ b/core/src/services/oss/writer.rs
@@ -117,7 +117,11 @@ impl oio::MultipartWrite for OssWriter {
})?
.to_string();
- Ok(oio::MultipartPart { part_number, etag })
+ Ok(oio::MultipartPart {
+ part_number,
+ etag,
+ checksum: None,
+ })
}
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/s3/core.rs b/core/src/services/s3/core.rs
index 2b1737009a..be4f7c34f0 100644
--- a/core/src/services/s3/core.rs
+++ b/core/src/services/s3/core.rs
@@ -250,21 +250,23 @@ impl S3Core {
req
}
-
+ pub fn calculate_checksum(&self, body: &Buffer) -> Option<String> {
+ match self.checksum_algorithm {
+ None => None,
+ Some(ChecksumAlgorithm::Crc32c) => {
+ let mut crc = 0u32;
+ body.clone()
+ .for_each(|b| crc = crc32c::crc32c_append(crc, &b));
+ Some(BASE64_STANDARD.encode(crc.to_be_bytes()))
+ }
+ }
+ }
pub fn insert_checksum_header(
&self,
mut req: http::request::Builder,
- body: &Buffer,
+ checksum: &str,
) -> http::request::Builder {
if let Some(checksum_algorithm) = self.checksum_algorithm.as_ref() {
- let checksum = match checksum_algorithm {
- ChecksumAlgorithm::Crc32c => {
- let mut crc = 0u32;
- body.clone()
- .for_each(|b| crc = crc32c::crc32c_append(crc, &b));
- BASE64_STANDARD.encode(crc.to_be_bytes())
- }
- };
req = req.header(checksum_algorithm.to_header_name(), checksum);
}
req
@@ -441,8 +443,11 @@ impl S3Core {
// Set SSE headers.
req = self.insert_sse_headers(req, true);
- // Set Checksum header.
- req = self.insert_checksum_header(req, &body);
+ // Calculate Checksum.
+ if let Some(checksum) = self.calculate_checksum(&body) {
+ // Set Checksum header.
+ req = self.insert_checksum_header(req, &checksum);
+ }
// Set body
let req = req.body(body).map_err(new_request_build_error)?;
@@ -626,6 +631,7 @@ impl S3Core {
part_number: usize,
size: u64,
body: Buffer,
+ checksum: Option<String>,
) -> Result<Request<Buffer>> {
let p = build_abs_path(&self.root, path);
@@ -644,8 +650,10 @@ impl S3Core {
// Set SSE headers.
req = self.insert_sse_headers(req, true);
- // Set Checksum header.
- req = self.insert_checksum_header(req, &body);
+ if let Some(checksum) = checksum {
+ // Set Checksum header.
+ req = self.insert_checksum_header(req, &checksum);
+ }
// Set body
let req = req.body(body).map_err(new_request_build_error)?;
@@ -792,6 +800,8 @@ pub struct CompleteMultipartUploadRequestPart {
/// ref: <https://github.com/tafia/quick-xml/issues/362>
#[serde(rename = "ETag")]
pub etag: String,
+ #[serde(rename = "ChecksumCRC32C", skip_serializing_if = "Option::is_none")]
+ pub checksum_crc32c: Option<String>,
}
/// Request of DeleteObjects.
@@ -921,14 +931,17 @@ mod tests {
CompleteMultipartUploadRequestPart {
part_number: 1,
etag: "\"a54357aff0632cce46d942af68356b38\"".to_string(),
+ ..Default::default()
},
CompleteMultipartUploadRequestPart {
part_number: 2,
etag: "\"0c78aef83f66abc1fa1e8477f296d394\"".to_string(),
+ ..Default::default()
},
CompleteMultipartUploadRequestPart {
part_number: 3,
etag: "\"acbd18db4cc2f85cedef654fccc4a4d8\"".to_string(),
+ ..Default::default()
},
],
};
diff --git a/core/src/services/s3/writer.rs b/core/src/services/s3/writer.rs
index e894175a94..3866404442 100644
--- a/core/src/services/s3/writer.rs
+++ b/core/src/services/s3/writer.rs
@@ -93,9 +93,16 @@ impl oio::MultipartWrite for S3Writer {
// AWS S3 requires part number must between [1..=10000]
let part_number = part_number + 1;
- let mut req =
- self.core
- .s3_upload_part_request(&self.path, upload_id, part_number, size, body)?;
+ let checksum = self.core.calculate_checksum(&body);
+
+ let mut req = self.core.s3_upload_part_request(
+ &self.path,
+ upload_id,
+ part_number,
+ size,
+ body,
+ checksum.clone(),
+ )?;
self.core.sign(&mut req).await?;
@@ -114,7 +121,11 @@ impl oio::MultipartWrite for S3Writer {
})?
.to_string();
- Ok(oio::MultipartPart { part_number, etag })
+ Ok(oio::MultipartPart {
+ part_number,
+ etag,
+ checksum,
+ })
}
_ => Err(parse_error(resp).await?),
}
@@ -123,9 +134,19 @@ impl oio::MultipartWrite for S3Writer {
async fn complete_part(&self, upload_id: &str, parts: &[oio::MultipartPart]) -> Result<()> {
let parts = parts
.iter()
- .map(|p| CompleteMultipartUploadRequestPart {
- part_number: p.part_number,
- etag: p.etag.clone(),
+ .map(|p| match &self.core.checksum_algorithm {
+ None => CompleteMultipartUploadRequestPart {
+ part_number: p.part_number,
+ etag: p.etag.clone(),
+ ..Default::default()
+ },
+ Some(checksum_algorithm) => match checksum_algorithm {
+ ChecksumAlgorithm::Crc32c => CompleteMultipartUploadRequestPart {
+ part_number: p.part_number,
+ etag: p.etag.clone(),
+ checksum_crc32c: p.checksum.clone(),
+ },
+ },
})
.collect();
diff --git a/core/src/services/upyun/writer.rs b/core/src/services/upyun/writer.rs
index 77df8dc2fb..71b4d8e115 100644
--- a/core/src/services/upyun/writer.rs
+++ b/core/src/services/upyun/writer.rs
@@ -98,6 +98,7 @@ impl oio::MultipartWrite for UpyunWriter {
StatusCode::NO_CONTENT | StatusCode::CREATED => Ok(oio::MultipartPart {
part_number,
etag: "".to_string(),
+ checksum: None,
}),
_ => Err(parse_error(resp).await?),
}
diff --git a/core/src/services/vercel_blob/writer.rs b/core/src/services/vercel_blob/writer.rs
index 242698ff6f..a335001a3b 100644
--- a/core/src/services/vercel_blob/writer.rs
+++ b/core/src/services/vercel_blob/writer.rs
@@ -106,6 +106,7 @@ impl oio::MultipartWrite for VercelBlobWriter {
Ok(oio::MultipartPart {
part_number,
etag: resp.etag,
+ checksum: None,
})
}
_ => Err(parse_error(resp).await?),