You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by al...@apache.org on 2023/10/25 16:59:43 UTC
[arrow-rs] branch master updated: Add MultiPartStore (#4961) (#4608) (#4971)
This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/master by this push:
new e78d1409c2 Add MultiPartStore (#4961) (#4608) (#4971)
e78d1409c2 is described below
commit e78d1409c265ae5c216fc62e51a0f20aa55f6415
Author: Raphael Taylor-Davies <17...@users.noreply.github.com>
AuthorDate: Wed Oct 25 17:59:37 2023 +0100
Add MultiPartStore (#4961) (#4608) (#4971)
* Add MultiPartStore (#4961) (#4608)
* Parse CompleteMultipartUploadResult (#4965)
* More docs
* Add integration test
* Fix azure
* More docs
* Don't gate multipart behind feature flag
---
object_store/src/aws/client.rs | 51 +++++++++++++++++++++++++---
object_store/src/aws/mod.rs | 49 ++++++++++++++++++++-------
object_store/src/azure/client.rs | 49 +++++++++++++++++++++++++--
object_store/src/azure/mod.rs | 73 +++++++++++++++++++---------------------
object_store/src/gcp/client.rs | 34 ++++++++++++++++---
object_store/src/gcp/mod.rs | 52 +++++++++++++++++++---------
object_store/src/lib.rs | 42 +++++++++++++++++++++--
object_store/src/multipart.rs | 61 ++++++++++++++++++++++++++++++---
object_store/src/signer.rs | 3 +-
9 files changed, 329 insertions(+), 85 deletions(-)
diff --git a/object_store/src/aws/client.rs b/object_store/src/aws/client.rs
index 00d6ee446f..4e98f259f8 100644
--- a/object_store/src/aws/client.rs
+++ b/object_store/src/aws/client.rs
@@ -27,7 +27,9 @@ use crate::client::retry::RetryExt;
use crate::client::GetOptionsExt;
use crate::multipart::PartId;
use crate::path::DELIMITER;
-use crate::{ClientOptions, GetOptions, ListResult, MultipartId, Path, Result, RetryConfig};
+use crate::{
+ ClientOptions, GetOptions, ListResult, MultipartId, Path, PutResult, Result, RetryConfig,
+};
use async_trait::async_trait;
use base64::prelude::BASE64_STANDARD;
use base64::Engine;
@@ -115,6 +117,9 @@ pub(crate) enum Error {
#[snafu(display("Error performing complete multipart request: {}", source))]
CompleteMultipartRequest { source: crate::client::retry::Error },
+ #[snafu(display("Error getting complete multipart response body: {}", source))]
+ CompleteMultipartResponseBody { source: reqwest::Error },
+
#[snafu(display("Got invalid list response: {}", source))]
InvalidListResponse { source: quick_xml::de::DeError },
@@ -162,6 +167,13 @@ struct MultipartPart {
part_number: usize,
}
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase", rename = "CompleteMultipartUploadResult")]
+struct CompleteMultipartResult {
+ #[serde(rename = "ETag")]
+ e_tag: String,
+}
+
#[derive(Deserialize)]
#[serde(rename_all = "PascalCase", rename = "DeleteResult")]
struct BatchDeleteResponse {
@@ -506,12 +518,32 @@ impl S3Client {
Ok(response.upload_id)
}
+ pub async fn put_part(
+ &self,
+ path: &Path,
+ upload_id: &MultipartId,
+ part_idx: usize,
+ data: Bytes,
+ ) -> Result<PartId> {
+ let part = (part_idx + 1).to_string();
+
+ let content_id = self
+ .put_request(
+ path,
+ data,
+ &[("partNumber", &part), ("uploadId", upload_id)],
+ )
+ .await?;
+
+ Ok(PartId { content_id })
+ }
+
pub async fn complete_multipart(
&self,
location: &Path,
upload_id: &str,
parts: Vec<PartId>,
- ) -> Result<()> {
+ ) -> Result<PutResult> {
let parts = parts
.into_iter()
.enumerate()
@@ -527,7 +559,8 @@ impl S3Client {
let credential = self.get_credential().await?;
let url = self.config.path_url(location);
- self.client
+ let response = self
+ .client
.request(Method::POST, url)
.query(&[("uploadId", upload_id)])
.body(body)
@@ -542,7 +575,17 @@ impl S3Client {
.await
.context(CompleteMultipartRequestSnafu)?;
- Ok(())
+ let data = response
+ .bytes()
+ .await
+ .context(CompleteMultipartResponseBodySnafu)?;
+
+ let response: CompleteMultipartResult =
+ quick_xml::de::from_reader(data.reader()).context(InvalidMultipartResponseSnafu)?;
+
+ Ok(PutResult {
+ e_tag: Some(response.e_tag),
+ })
}
}
diff --git a/object_store/src/aws/mod.rs b/object_store/src/aws/mod.rs
index 25894a1c34..57254c7cf4 100644
--- a/object_store/src/aws/mod.rs
+++ b/object_store/src/aws/mod.rs
@@ -44,7 +44,7 @@ use crate::aws::client::S3Client;
use crate::client::get::GetClientExt;
use crate::client::list::ListClientExt;
use crate::client::CredentialProvider;
-use crate::multipart::{PartId, PutPart, WriteMultiPart};
+use crate::multipart::{MultiPartStore, PartId, PutPart, WriteMultiPart};
use crate::signer::Signer;
use crate::{
GetOptions, GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, Path, PutResult,
@@ -246,18 +246,9 @@ struct S3MultiPartUpload {
#[async_trait]
impl PutPart for S3MultiPartUpload {
async fn put_part(&self, buf: Vec<u8>, part_idx: usize) -> Result<PartId> {
- let part = (part_idx + 1).to_string();
-
- let content_id = self
- .client
- .put_request(
- &self.location,
- buf.into(),
- &[("partNumber", &part), ("uploadId", &self.upload_id)],
- )
- .await?;
-
- Ok(PartId { content_id })
+ self.client
+ .put_part(&self.location, &self.upload_id, part_idx, buf.into())
+ .await
}
async fn complete(&self, completed_parts: Vec<PartId>) -> Result<()> {
@@ -268,6 +259,36 @@ impl PutPart for S3MultiPartUpload {
}
}
+#[async_trait]
+impl MultiPartStore for AmazonS3 {
+ async fn create_multipart(&self, path: &Path) -> Result<MultipartId> {
+ self.client.create_multipart(path).await
+ }
+
+ async fn put_part(
+ &self,
+ path: &Path,
+ id: &MultipartId,
+ part_idx: usize,
+ data: Bytes,
+ ) -> Result<PartId> {
+ self.client.put_part(path, id, part_idx, data).await
+ }
+
+ async fn complete_multipart(
+ &self,
+ path: &Path,
+ id: &MultipartId,
+ parts: Vec<PartId>,
+ ) -> Result<PutResult> {
+ self.client.complete_multipart(path, id, parts).await
+ }
+
+ async fn abort_multipart(&self, path: &Path, id: &MultipartId) -> Result<()> {
+ self.client.delete_request(path, &[("uploadId", id)]).await
+ }
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -293,6 +314,8 @@ mod tests {
list_with_delimiter(&integration).await;
rename_and_copy(&integration).await;
stream_get(&integration).await;
+ multipart(&integration, &integration).await;
+
if test_not_exists {
copy_if_not_exists(&integration).await;
}
diff --git a/object_store/src/azure/client.rs b/object_store/src/azure/client.rs
index cd3df8c7b8..9f47b9a815 100644
--- a/object_store/src/azure/client.rs
+++ b/object_store/src/azure/client.rs
@@ -19,13 +19,16 @@ use super::credential::AzureCredential;
use crate::azure::credential::*;
use crate::azure::{AzureCredentialProvider, STORE};
use crate::client::get::GetClient;
-use crate::client::header::HeaderConfig;
+use crate::client::header::{get_etag, HeaderConfig};
use crate::client::list::ListClient;
use crate::client::retry::RetryExt;
use crate::client::GetOptionsExt;
+use crate::multipart::PartId;
use crate::path::DELIMITER;
use crate::util::deserialize_rfc1123;
-use crate::{ClientOptions, GetOptions, ListResult, ObjectMeta, Path, Result, RetryConfig};
+use crate::{
+ ClientOptions, GetOptions, ListResult, ObjectMeta, Path, PutResult, Result, RetryConfig,
+};
use async_trait::async_trait;
use base64::prelude::BASE64_STANDARD;
use base64::Engine;
@@ -84,6 +87,11 @@ pub(crate) enum Error {
Authorization {
source: crate::azure::credential::Error,
},
+
+ #[snafu(display("Unable to extract metadata from headers: {}", source))]
+ Metadata {
+ source: crate::client::header::Error,
+ },
}
impl From<Error> for crate::Error {
@@ -190,6 +198,43 @@ impl AzureClient {
Ok(response)
}
+ /// PUT a block <https://learn.microsoft.com/en-us/rest/api/storageservices/put-block>
+ pub async fn put_block(&self, path: &Path, part_idx: usize, data: Bytes) -> Result<PartId> {
+ let content_id = format!("{part_idx:20}");
+ let block_id: BlockId = content_id.clone().into();
+
+ self.put_request(
+ path,
+ Some(data),
+ true,
+ &[
+ ("comp", "block"),
+ ("blockid", &BASE64_STANDARD.encode(block_id)),
+ ],
+ )
+ .await?;
+
+ Ok(PartId { content_id })
+ }
+
+ /// PUT a block list <https://learn.microsoft.com/en-us/rest/api/storageservices/put-block-list>
+ pub async fn put_block_list(&self, path: &Path, parts: Vec<PartId>) -> Result<PutResult> {
+ let blocks = parts
+ .into_iter()
+ .map(|part| BlockId::from(part.content_id))
+ .collect();
+
+ let block_list = BlockList { blocks };
+ let block_xml = block_list.to_xml();
+
+ let response = self
+ .put_request(path, Some(block_xml.into()), true, &[("comp", "blocklist")])
+ .await?;
+
+ let e_tag = get_etag(response.headers()).context(MetadataSnafu)?;
+ Ok(PutResult { e_tag: Some(e_tag) })
+ }
+
/// Make an Azure Delete request <https://docs.microsoft.com/en-us/rest/api/storageservices/delete-blob>
pub async fn delete_request<T: Serialize + ?Sized + Sync>(
&self,
diff --git a/object_store/src/azure/mod.rs b/object_store/src/azure/mod.rs
index 5f768756a6..779ac2f71f 100644
--- a/object_store/src/azure/mod.rs
+++ b/object_store/src/azure/mod.rs
@@ -26,15 +26,12 @@
//! [ObjectStore::abort_multipart] is a no-op, since Azure Blob Store doesn't provide
//! a way to drop old blocks. Instead unused blocks are automatically cleaned up
//! after 7 days.
-use self::client::{BlockId, BlockList};
use crate::{
multipart::{PartId, PutPart, WriteMultiPart},
path::Path,
GetOptions, GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, PutResult, Result,
};
use async_trait::async_trait;
-use base64::prelude::BASE64_STANDARD;
-use base64::Engine;
use bytes::Bytes;
use futures::stream::BoxStream;
use std::fmt::Debug;
@@ -53,6 +50,7 @@ mod credential;
/// [`CredentialProvider`] for [`MicrosoftAzure`]
pub type AzureCredentialProvider = Arc<dyn CredentialProvider<Credential = AzureCredential>>;
use crate::client::header::get_etag;
+use crate::multipart::MultiPartStore;
pub use builder::{AzureConfigKey, MicrosoftAzureBuilder};
pub use credential::AzureCredential;
@@ -151,43 +149,44 @@ struct AzureMultiPartUpload {
#[async_trait]
impl PutPart for AzureMultiPartUpload {
- async fn put_part(&self, buf: Vec<u8>, part_idx: usize) -> Result<PartId> {
- let content_id = format!("{part_idx:20}");
- let block_id: BlockId = content_id.clone().into();
-
- self.client
- .put_request(
- &self.location,
- Some(buf.into()),
- true,
- &[
- ("comp", "block"),
- ("blockid", &BASE64_STANDARD.encode(block_id)),
- ],
- )
- .await?;
+ async fn put_part(&self, buf: Vec<u8>, idx: usize) -> Result<PartId> {
+ self.client.put_block(&self.location, idx, buf.into()).await
+ }
- Ok(PartId { content_id })
+ async fn complete(&self, parts: Vec<PartId>) -> Result<()> {
+ self.client.put_block_list(&self.location, parts).await?;
+ Ok(())
}
+}
- async fn complete(&self, completed_parts: Vec<PartId>) -> Result<()> {
- let blocks = completed_parts
- .into_iter()
- .map(|part| BlockId::from(part.content_id))
- .collect();
+#[async_trait]
+impl MultiPartStore for MicrosoftAzure {
+ async fn create_multipart(&self, _: &Path) -> Result<MultipartId> {
+ Ok(String::new())
+ }
- let block_list = BlockList { blocks };
- let block_xml = block_list.to_xml();
+ async fn put_part(
+ &self,
+ path: &Path,
+ _: &MultipartId,
+ part_idx: usize,
+ data: Bytes,
+ ) -> Result<PartId> {
+ self.client.put_block(path, part_idx, data).await
+ }
- self.client
- .put_request(
- &self.location,
- Some(block_xml.into()),
- true,
- &[("comp", "blocklist")],
- )
- .await?;
+ async fn complete_multipart(
+ &self,
+ path: &Path,
+ _: &MultipartId,
+ parts: Vec<PartId>,
+ ) -> Result<PutResult> {
+ self.client.put_block_list(path, parts).await
+ }
+ async fn abort_multipart(&self, _: &Path, _: &MultipartId) -> Result<()> {
+ // There is no way to drop blocks that have been uploaded. Instead, they simply
+ // expire in 7 days.
Ok(())
}
}
@@ -195,10 +194,7 @@ impl PutPart for AzureMultiPartUpload {
#[cfg(test)]
mod tests {
use super::*;
- use crate::tests::{
- copy_if_not_exists, get_opts, list_uses_directories_correctly, list_with_delimiter,
- put_get_delete_list_opts, rename_and_copy, stream_get,
- };
+ use crate::tests::*;
#[tokio::test]
async fn azure_blob_test() {
@@ -212,6 +208,7 @@ mod tests {
rename_and_copy(&integration).await;
copy_if_not_exists(&integration).await;
stream_get(&integration).await;
+ multipart(&integration, &integration).await;
}
#[test]
diff --git a/object_store/src/gcp/client.rs b/object_store/src/gcp/client.rs
index 558a6f8d2a..8c44f90164 100644
--- a/object_store/src/gcp/client.rs
+++ b/object_store/src/gcp/client.rs
@@ -24,7 +24,7 @@ use crate::client::GetOptionsExt;
use crate::gcp::{GcpCredential, GcpCredentialProvider, STORE};
use crate::multipart::PartId;
use crate::path::{Path, DELIMITER};
-use crate::{ClientOptions, GetOptions, ListResult, MultipartId, Result, RetryConfig};
+use crate::{ClientOptions, GetOptions, ListResult, MultipartId, PutResult, Result, RetryConfig};
use async_trait::async_trait;
use bytes::{Buf, Bytes};
use percent_encoding::{percent_encode, utf8_percent_encode, NON_ALPHANUMERIC};
@@ -184,6 +184,30 @@ impl GoogleCloudStorageClient {
Ok(get_etag(response.headers()).context(MetadataSnafu)?)
}
+ /// Perform a put part request <https://cloud.google.com/storage/docs/xml-api/put-object-multipart>
+ ///
+ /// Returns the new [`PartId`]
+ pub async fn put_part(
+ &self,
+ path: &Path,
+ upload_id: &MultipartId,
+ part_idx: usize,
+ data: Bytes,
+ ) -> Result<PartId> {
+ let content_id = self
+ .put_request(
+ path,
+ data,
+ &[
+ ("partNumber", &format!("{}", part_idx + 1)),
+ ("uploadId", upload_id),
+ ],
+ )
+ .await?;
+
+ Ok(PartId { content_id })
+ }
+
/// Initiate a multi-part upload <https://cloud.google.com/storage/docs/xml-api/post-object-multipart>
pub async fn multipart_initiate(&self, path: &Path) -> Result<MultipartId> {
let credential = self.get_credential().await?;
@@ -240,7 +264,7 @@ impl GoogleCloudStorageClient {
path: &Path,
multipart_id: &MultipartId,
completed_parts: Vec<PartId>,
- ) -> Result<()> {
+ ) -> Result<PutResult> {
let upload_id = multipart_id.clone();
let url = self.object_url(path);
@@ -263,7 +287,8 @@ impl GoogleCloudStorageClient {
// https://github.com/tafia/quick-xml/issues/350
.replace(""", "\"");
- self.client
+ let result = self
+ .client
.request(Method::POST, &url)
.bearer_auth(&credential.bearer)
.query(&[("uploadId", upload_id)])
@@ -274,7 +299,8 @@ impl GoogleCloudStorageClient {
path: path.as_ref(),
})?;
- Ok(())
+ let etag = get_etag(result.headers()).context(MetadataSnafu)?;
+ Ok(PutResult { e_tag: Some(etag) })
}
/// Perform a delete request <https://cloud.google.com/storage/docs/xml-api/delete-object>
diff --git a/object_store/src/gcp/mod.rs b/object_store/src/gcp/mod.rs
index 6512a8b036..0eb3e9c23c 100644
--- a/object_store/src/gcp/mod.rs
+++ b/object_store/src/gcp/mod.rs
@@ -45,6 +45,7 @@ use tokio::io::AsyncWrite;
use crate::client::get::GetClientExt;
use crate::client::list::ListClientExt;
+use crate::multipart::MultiPartStore;
pub use builder::{GoogleCloudStorageBuilder, GoogleConfigKey};
pub use credential::GcpCredential;
@@ -90,27 +91,17 @@ struct GCSMultipartUpload {
impl PutPart for GCSMultipartUpload {
/// Upload an object part <https://cloud.google.com/storage/docs/xml-api/put-object-multipart>
async fn put_part(&self, buf: Vec<u8>, part_idx: usize) -> Result<PartId> {
- let upload_id = self.multipart_id.clone();
- let content_id = self
- .client
- .put_request(
- &self.path,
- buf.into(),
- &[
- ("partNumber", format!("{}", part_idx + 1)),
- ("uploadId", upload_id),
- ],
- )
- .await?;
-
- Ok(PartId { content_id })
+ self.client
+ .put_part(&self.path, &self.multipart_id, part_idx, buf.into())
+ .await
}
/// Complete a multipart upload <https://cloud.google.com/storage/docs/xml-api/post-object-complete>
async fn complete(&self, completed_parts: Vec<PartId>) -> Result<()> {
self.client
.multipart_complete(&self.path, &self.multipart_id, completed_parts)
- .await
+ .await?;
+ Ok(())
}
}
@@ -169,6 +160,36 @@ impl ObjectStore for GoogleCloudStorage {
}
}
+#[async_trait]
+impl MultiPartStore for GoogleCloudStorage {
+ async fn create_multipart(&self, path: &Path) -> Result<MultipartId> {
+ self.client.multipart_initiate(path).await
+ }
+
+ async fn put_part(
+ &self,
+ path: &Path,
+ id: &MultipartId,
+ part_idx: usize,
+ data: Bytes,
+ ) -> Result<PartId> {
+ self.client.put_part(path, id, part_idx, data).await
+ }
+
+ async fn complete_multipart(
+ &self,
+ path: &Path,
+ id: &MultipartId,
+ parts: Vec<PartId>,
+ ) -> Result<PutResult> {
+ self.client.multipart_complete(path, id, parts).await
+ }
+
+ async fn abort_multipart(&self, path: &Path, id: &MultipartId) -> Result<()> {
+ self.client.multipart_cleanup(path, id).await
+ }
+}
+
#[cfg(test)]
mod test {
@@ -197,6 +218,7 @@ mod test {
// Fake GCS server does not yet implement XML Multipart uploads
// https://github.com/fsouza/fake-gcs-server/issues/852
stream_get(&integration).await;
+ multipart(&integration, &integration).await;
// Fake GCS server doesn't currently honor preconditions
get_opts(&integration).await;
}
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index 656b30390a..9a06672298 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -263,7 +263,6 @@ pub use client::{
#[cfg(feature = "cloud")]
mod config;
-#[cfg(feature = "cloud")]
pub mod multipart;
mod parse;
mod util;
@@ -302,18 +301,29 @@ pub trait ObjectStore: std::fmt::Display + Send + Sync + Debug + 'static {
/// should be able to observe a partially written object
async fn put(&self, location: &Path, bytes: Bytes) -> Result<PutResult>;
- /// Get a multi-part upload that allows writing data in chunks
+ /// Get a multi-part upload that allows writing data in chunks.
///
/// Most cloud-based uploads will buffer and upload parts in parallel.
///
/// To complete the upload, [AsyncWrite::poll_shutdown] must be called
/// to completion. This operation is guaranteed to be atomic, it will either
/// make all the written data available at `location`, or fail. No clients
- /// should be able to observe a partially written object
+ /// should be able to observe a partially written object.
///
/// For some object stores (S3, GCS, and local in particular), if the
/// writer fails or panics, you must call [ObjectStore::abort_multipart]
/// to clean up partially written data.
+ ///
+ /// For applications requiring fine-grained control of multipart uploads
+ /// see [`MultiPartStore`], although note that this interface cannot be
+ /// supported by all [`ObjectStore`] backends.
+ ///
+ /// For applications looking to implement this interface for a custom
+ /// multipart API, see [`WriteMultiPart`] which handles the complexities
+ /// of performing parallel uploads of fixed size parts.
+ ///
+ /// [`WriteMultiPart`]: multipart::WriteMultiPart
+ /// [`MultiPartStore`]: multipart::MultiPartStore
async fn put_multipart(
&self,
location: &Path,
@@ -934,6 +944,7 @@ mod test_util {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::multipart::MultiPartStore;
use crate::test_util::flatten_list_stream;
use chrono::TimeZone;
use rand::{thread_rng, Rng};
@@ -1681,6 +1692,31 @@ mod tests {
storage.delete(&path2).await.unwrap();
}
+ pub(crate) async fn multipart(storage: &dyn ObjectStore, multipart: &dyn MultiPartStore) {
+ let path = Path::from("test_multipart");
+ let chunk_size = 5 * 1024 * 1024;
+
+ let chunks = get_chunks(chunk_size, 2);
+
+ let id = multipart.create_multipart(&path).await.unwrap();
+
+ let parts: Vec<_> = futures::stream::iter(chunks)
+ .enumerate()
+ .map(|(idx, b)| multipart.put_part(&path, &id, idx, b))
+ .buffered(2)
+ .try_collect()
+ .await
+ .unwrap();
+
+ multipart
+ .complete_multipart(&path, &id, parts)
+ .await
+ .unwrap();
+
+ let meta = storage.head(&path).await.unwrap();
+ assert_eq!(meta.size, chunk_size * 2);
+ }
+
async fn delete_fixtures(storage: &DynObjectStore) {
let paths = storage.list(None).map_ok(|meta| meta.location).boxed();
storage
diff --git a/object_store/src/multipart.rs b/object_store/src/multipart.rs
index d4c911fcea..1dcd5a6f49 100644
--- a/object_store/src/multipart.rs
+++ b/object_store/src/multipart.rs
@@ -22,17 +22,18 @@
//! especially useful when dealing with large files or high-throughput systems.
use async_trait::async_trait;
+use bytes::Bytes;
use futures::{stream::FuturesUnordered, Future, StreamExt};
use std::{io, pin::Pin, sync::Arc, task::Poll};
use tokio::io::AsyncWrite;
-use crate::Result;
+use crate::path::Path;
+use crate::{MultipartId, PutResult, Result};
type BoxedTryFuture<T> = Pin<Box<dyn Future<Output = Result<T, io::Error>> + Send>>;
-/// A trait that can be implemented by cloud-based object stores
-/// and used in combination with [`WriteMultiPart`] to provide
-/// multipart upload support
+/// A trait used in combination with [`WriteMultiPart`] to implement
+/// [`AsyncWrite`] on top of an API for multipart upload
#[async_trait]
pub trait PutPart: Send + Sync + 'static {
/// Upload a single part
@@ -52,6 +53,9 @@ pub struct PartId {
}
/// Wrapper around a [`PutPart`] that implements [`AsyncWrite`]
+///
+/// Data will be uploaded in fixed size chunks of 10 MiB in parallel,
+/// up to the configured maximum concurrency
pub struct WriteMultiPart<T: PutPart> {
inner: Arc<T>,
/// A list of completed parts, in sequential order.
@@ -263,3 +267,52 @@ impl<T: PutPart> std::fmt::Debug for WriteMultiPart<T> {
.finish()
}
}
+
+/// A low-level interface for interacting with multipart upload APIs
+///
+/// Most use-cases should prefer [`ObjectStore::put_multipart`] as this is supported by more
+/// backends, including [`LocalFileSystem`], and automatically handles uploading fixed
+/// size parts of sufficient size in parallel
+///
+/// [`ObjectStore::put_multipart`]: crate::ObjectStore::put_multipart
+/// [`LocalFileSystem`]: crate::local::LocalFileSystem
+#[async_trait]
+pub trait MultiPartStore: Send + Sync + 'static {
+ /// Creates a new multipart upload, returning the [`MultipartId`]
+ async fn create_multipart(&self, path: &Path) -> Result<MultipartId>;
+
+ /// Uploads a new part with index `part_idx`
+ ///
+ /// `part_idx` should be an integer in the range `0..N` where `N` is the number of
+ /// parts in the upload. Parts may be uploaded concurrently and in any order.
+ ///
+ /// Most stores require that all parts excluding the last are at least 5 MiB, and some
+ /// further require that all parts excluding the last be the same size, e.g. [R2].
+ /// [`WriteMultiPart`] performs writes in fixed size blocks of 10 MiB, and clients wanting
+ /// to maximise compatibility should look to do likewise.
+ ///
+ /// [R2]: https://developers.cloudflare.com/r2/objects/multipart-objects/#limitations
+ async fn put_part(
+ &self,
+ path: &Path,
+ id: &MultipartId,
+ part_idx: usize,
+ data: Bytes,
+ ) -> Result<PartId>;
+
+ /// Completes a multipart upload
+ ///
+ /// The `i`'th value of `parts` must be a [`PartId`] returned by a call to [`Self::put_part`]
+ /// with a `part_idx` of `i`, and the same `path` and `id` as provided to this method. Calling
+ /// this method with out of sequence or repeated [`PartId`], or [`PartId`] returned for other
+ /// values of `path` or `id`, will result in implementation-defined behaviour
+ async fn complete_multipart(
+ &self,
+ path: &Path,
+ id: &MultipartId,
+ parts: Vec<PartId>,
+ ) -> Result<PutResult>;
+
+ /// Aborts a multipart upload
+ async fn abort_multipart(&self, path: &Path, id: &MultipartId) -> Result<()>;
+}
diff --git a/object_store/src/signer.rs b/object_store/src/signer.rs
index f792397a78..ed92e28799 100644
--- a/object_store/src/signer.rs
+++ b/object_store/src/signer.rs
@@ -23,8 +23,7 @@ use reqwest::Method;
use std::{fmt, time::Duration};
use url::Url;
-/// Universal API to presigned URLs generated from multiple object store services. Not supported by
-/// all object store services.
+/// Universal API to generate presigned URLs from multiple object store services.
#[async_trait]
pub trait Signer: Send + Sync + fmt::Debug + 'static {
/// Given the intended [`Method`] and [`Path`] to use and the desired length of time for which