bump object_store dependency to 0.7.1

Also points to a new branch for direct multi-part upload that uses more existing code.
ArroyoSystems · Oct 23, 2023 · 1115613 · 1115613
1 parent d0e6e63
commit 1115613
Show file tree

Hide file tree

Showing 8 changed files with 83 additions and 63 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -38,7 +38,7 @@ arrow = { version = "43.0.0" }
 arrow-buffer = { version = "43.0.0" }
 arrow-array = { version = "43.0.0" }
 arrow-schema = { version = "43.0.0" }
-object_store = { version = "0.6.1" }
+object_store = { version = "0.7.1" }
 parquet = { version = "43.0.0" }
 
 [profile.release]
@@ -56,4 +56,4 @@ arrow = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '43.0.0/arr
 arrow-buffer = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '43.0.0/arroyo_patches'}
 arrow-array = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '43.0.0/arroyo_patches'}
 arrow-schema = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '43.0.0/arroyo_patches'}
-object_store = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '43.0.0/arroyo_patches'}
+object_store = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = 'object_store/put_part_api'}
diff --git a/arroyo-controller/src/compiler.rs b/arroyo-controller/src/compiler.rs
@@ -134,7 +134,7 @@ arrow = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '43.0.0/arr
 arrow-buffer = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '43.0.0/arroyo_patches'}
 arrow-array = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '43.0.0/arroyo_patches'}
 arrow-schema = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = '43.0.0/arroyo_patches'}
-object_store = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = 'multipart_gcp' }
+object_store = {git = 'https://github.com/ArroyoSystems/arrow-rs', branch = 'object_store/put_part_api' }
 "#;
 
         // NOTE: These must be kept in sync with the Cargo configs in docker/build_base and build_dir/Cargo.toml

diff --git a/arroyo-storage/Cargo.toml b/arroyo-storage/Cargo.toml
@@ -14,7 +14,7 @@ bytes = "1.4.0"
 # better way to do this
 rusoto_core = "0.48.0"
 
-object_store = {version = "0.6.1", features = ["aws", "gcp"]}
+object_store = {workspace = true, features = ["aws", "gcp"]}
 regex = "1.9.5"
 thiserror = "1"
 tokio = { version = "1", features = ["fs"] }

diff --git a/arroyo-storage/src/lib.rs b/arroyo-storage/src/lib.rs
@@ -10,9 +10,10 @@ use aws::ArroyoCredentialProvider;
 use bytes::Bytes;
 use object_store::aws::AmazonS3ConfigKey;
 use object_store::gcp::GoogleCloudStorageBuilder;
+use object_store::multipart::PartId;
 use object_store::path::Path;
+use object_store::MultipartId;
 use object_store::{aws::AmazonS3Builder, local::LocalFileSystem, ObjectStore};
-use object_store::{MultipartId, UploadPart};
 use regex::{Captures, Regex};
 use thiserror::Error;
 
@@ -440,9 +441,10 @@ impl StorageProvider {
     pub async fn start_multipart(&self, path: &Path) -> Result<MultipartId, StorageError> {
         Ok(self
             .object_store
-            .start_multipart(path)
+            .initiate_multipart_upload(path)
             .await
-            .map_err(|e| Into::<StorageError>::into(e))?)
+            .map_err(|e| Into::<StorageError>::into(e))?
+            .0)
     }
 
     pub async fn add_multipart(
@@ -451,10 +453,12 @@ impl StorageProvider {
         multipart_id: &MultipartId,
         part_number: usize,
         bytes: Bytes,
-    ) -> Result<UploadPart, StorageError> {
+    ) -> Result<PartId, StorageError> {
         Ok(self
             .object_store
-            .add_multipart(path, multipart_id, part_number, bytes)
+            .get_put_part(path, multipart_id)
+            .await?
+            .put_part(bytes, part_number)
             .await
             .map_err(|e| Into::<StorageError>::into(e))?)
     }
@@ -463,11 +467,13 @@ impl StorageProvider {
         &self,
         path: &Path,
         multipart_id: &MultipartId,
-        parts: Vec<UploadPart>,
+        parts: Vec<PartId>,
     ) -> Result<(), StorageError> {
         Ok(self
             .object_store
-            .close_multipart(path, multipart_id, parts)
+            .get_put_part(path, multipart_id)
+            .await?
+            .complete(parts)
             .await
             .map_err(|e| Into::<StorageError>::into(e))?)
     }