Increase the default multipart target size

Author: rdettai

quickwit/quickwit-storage/src/lib.rs

@@ -74,7 +74,7 @@ pub use self::object_storage::{
 #[cfg(feature = "gcs")]
 pub use self::opendal_storage::GoogleCloudStorageFactory;
 #[cfg(all(feature = "gcs", feature = "integration-testsuite"))]
-pub use self::opendal_storage::{DummyTokenLoader, new_emulated_google_cloud_storage};
+pub use self::opendal_storage::test_credentials;
 pub use self::ram_storage::{RamStorage, RamStorageBuilder};
 pub use self::split::{SplitPayload, SplitPayloadBuilder};
 #[cfg(any(test, feature = "testsuite"))]

quickwit/quickwit-storage/src/object_storage/policy.rs

@@ -63,8 +63,6 @@ impl MultiPartPolicy {
     }
 }
 
-// Default values from https://github.com/apache/hadoop/blob/trunk/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java
-// The best default value may however differ depending on vendors.
 impl Default for MultiPartPolicy {
     fn default() -> Self {
         MultiPartPolicy {

quickwit/quickwit-storage/src/opendal_storage/base.rs

@@ -26,20 +26,19 @@ use tokio_util::compat::{FuturesAsyncReadCompatExt, FuturesAsyncWriteCompatExt};
 use crate::metrics::object_storage_get_slice_in_flight_guards;
 use crate::storage::SendableAsync;
 use crate::{
-    BulkDeleteError, OwnedBytes, PutPayload, Storage, StorageError, StorageErrorKind,
-    StorageResolverError, StorageResult,
+    BulkDeleteError, MultiPartPolicy, OwnedBytes, PutPayload, Storage, StorageError,
+    StorageErrorKind, StorageResolverError, StorageResult,
 };
 
 /// OpenDAL based storage implementation.
 /// # TODO
 ///
 /// - Implement REQUEST_SEMAPHORE to control the concurrency.
 /// - Implement STORAGE_METRICS for metrics.
-/// - Add multipart_policy to control write at once or via multiple.
-#[derive(Clone)]
 pub struct OpendalStorage {
     uri: Uri,
     op: Operator,
+    multipart_policy: MultiPartPolicy,
 }
 
 impl fmt::Debug for OpendalStorage {
@@ -58,7 +57,16 @@ impl OpendalStorage {
         cfg: opendal::services::Gcs,
     ) -> Result<Self, StorageResolverError> {
         let op = Operator::new(cfg)?.finish();
-        Ok(Self { uri, op })
+        Ok(Self {
+            uri,
+            op,
+            // limits are the same as on S3
+            multipart_policy: MultiPartPolicy::default(),
+        })
+    }
+
+    pub fn set_policy(&mut self, multipart_policy: MultiPartPolicy) {
+        self.multipart_policy = multipart_policy;
     }
 }
 
@@ -69,17 +77,14 @@ impl Storage for OpendalStorage {
         Ok(())
     }
 
-    /// # TODO
-    ///
-    /// We can implement something like `multipart_policy` determine whether to use copy.
-    /// If the payload is small enough, we can call `op.write()` at once.
     async fn put(&self, path: &Path, payload: Box<dyn PutPayload>) -> StorageResult<()> {
         let path = path.as_os_str().to_string_lossy();
         let mut payload_reader = payload.byte_stream().await?.into_async_read();
 
         let mut storage_writer = self
             .op
             .writer_with(&path)
+            .chunk(self.multipart_policy.part_num_bytes(payload.len()) as usize)
             .await?
             .into_futures_async_write()
             .compat_write();

quickwit/quickwit-storage/src/opendal_storage/google_cloud_storage.rs

@@ -50,10 +50,14 @@ impl StorageFactory for GoogleCloudStorageFactory {
     }
 }
 
+/// Helpers to configure the GCP local test setup.
 #[cfg(feature = "integration-testsuite")]
-pub mod test_credentials {
+pub mod test_config_helpers {
     use super::*;
 
+    /// URL of the local GCP emulator.
+    pub const LOCAL_GCP_EMULATOR_ENDPOINT: &str = "http://127.0.0.1:4443";
+
     /// reqsign::GoogleTokenLoad implementation for testing.
     #[derive(Debug)]
     pub struct DummyTokenLoader;
@@ -78,8 +82,8 @@ pub mod test_credentials {
         let cfg = opendal::services::Gcs::default()
             .bucket(&bucket)
             .root(&root.to_string_lossy())
-            .endpoint("http://127.0.0.1:4443")
-            .customized_token_loader(Box::new(test_credentials::DummyTokenLoader));
+            .endpoint(LOCAL_GCP_EMULATOR_ENDPOINT)
+            .customized_token_loader(Box::new(test_config_helpers::DummyTokenLoader));
         let store = OpendalStorage::new_google_cloud_storage(uri.clone(), cfg)?;
         Ok(store)
     }

quickwit/quickwit-storage/src/opendal_storage/mod.rs

@@ -19,6 +19,4 @@ mod google_cloud_storage;
 
 pub use google_cloud_storage::GoogleCloudStorageFactory;
 #[cfg(feature = "integration-testsuite")]
-pub use google_cloud_storage::test_credentials::{
-    DummyTokenLoader, new_emulated_google_cloud_storage,
-};
+pub use google_cloud_storage::test_config_helpers;

quickwit/quickwit-storage/tests/google_cloud_storage.rs

@@ -17,17 +17,19 @@
 
 #[cfg(all(feature = "integration-testsuite", feature = "gcs"))]
 #[cfg_attr(not(feature = "ci-test"), ignore)]
-mod tests {
+mod gcp_storage_test_suite {
     use std::str::FromStr;
 
     use anyhow::Context;
     use quickwit_common::rand::append_random_suffix;
     use quickwit_common::setup_logging_for_tests;
     use quickwit_common::uri::Uri;
-    use quickwit_storage::{DummyTokenLoader, new_emulated_google_cloud_storage};
+    use quickwit_storage::test_config_helpers::{
+        DummyTokenLoader, LOCAL_GCP_EMULATOR_ENDPOINT, new_emulated_google_cloud_storage,
+    };
     use reqsign::GoogleTokenLoad;
 
-    pub async fn sign_gcs_requet(req: &mut reqwest::Request) -> anyhow::Result<()> {
+    pub async fn sign_gcs_request(req: &mut reqwest::Request) -> anyhow::Result<()> {
         let client = reqwest::Client::new();
         let token = DummyTokenLoader
             .load(client.clone())
@@ -42,18 +44,16 @@ mod tests {
 
     async fn create_gcs_bucket(bucket_name: &str) -> anyhow::Result<()> {
         let client = reqwest::Client::new();
-        let url = "http://127.0.0.1:4443/storage/v1/b";
+        let url = format!("{LOCAL_GCP_EMULATOR_ENDPOINT}/storage/v1/b");
         let mut request = client
             .post(url)
             .body(serde_json::to_vec(&serde_json::json!({
                 "name": bucket_name,
-                // "location": "us-central1",
-                // "storageClass": "STANDARD",
             }))?)
             .header(reqwest::header::CONTENT_TYPE, "application/json")
             .build()?;
 
-        sign_gcs_requet(&mut request).await?;
+        sign_gcs_request(&mut request).await?;
 
         let response = client.execute(request).await?;
 
@@ -90,6 +90,13 @@ mod tests {
         // supported in the emulated GCS server
         // (https://github.com/fsouza/fake-gcs-server/pull/1164)
 
+        // object_storage.set_policy(MultiPartPolicy {
+        //     target_part_num_bytes: 5 * 1_024 * 1_024,
+        //     max_num_parts: 10_000,
+        //     multipart_threshold_num_bytes: 10_000_000,
+        //     max_object_num_bytes: 5_000_000_000_000,
+        //     max_concurrent_uploads: 100,
+        // });
         // quickwit_storage::storage_test_multi_part_upload(&mut object_storage)
         //     .await
         //     .context("test multipart upload failed")?;