Avoid sending ComposeObject requests after upload to GCS

This should not be required anymore, but leave it as an option, since
likely this is required for old files.

Signed-off-by: Azat Khuzhin <a.khuzhin@semrush.com>

src/Backups/BackupIO_S3.cpp

@@ -72,6 +72,7 @@ namespace
         S3::ClientSettings client_settings{
             .use_virtual_addressing = s3_uri.is_virtual_hosted_style,
             .disable_checksum = local_settings.s3_disable_checksum,
+            .gcs_issue_compose_request = context->getConfigRef().getBool("s3.gcs_issue_compose_request", false),
         };
 
         return S3::ClientFactory::instance().create(

src/Coordination/KeeperSnapshotManagerS3.cpp

@@ -102,6 +102,7 @@ void KeeperSnapshotManagerS3::updateS3Configuration(const Poco::Util::AbstractCo
         S3::ClientSettings client_settings{
             .use_virtual_addressing = new_uri.is_virtual_hosted_style,
             .disable_checksum = false,
+            .gcs_issue_compose_request = false,
         };
 
         auto client = S3::ClientFactory::instance().create(

src/Disks/ObjectStorages/S3/diskSettings.cpp

@@ -97,6 +97,7 @@ std::unique_ptr<S3::Client> getClient(
     S3::ClientSettings client_settings{
         .use_virtual_addressing = uri.is_virtual_hosted_style,
         .disable_checksum = local_settings.s3_disable_checksum,
+        .gcs_issue_compose_request = config.getBool("s3.gcs_issue_compose_request", false),
     };
 
     return S3::ClientFactory::instance().create(

src/IO/S3/Client.cpp

@@ -413,7 +413,7 @@ Model::CompleteMultipartUploadOutcome Client::CompleteMultipartUpload(CompleteMu
             outcome = Aws::S3::Model::CompleteMultipartUploadOutcome(Aws::S3::Model::CompleteMultipartUploadResult());
     }
 
-    if (outcome.IsSuccess() && provider_type == ProviderType::GCS)
+    if (outcome.IsSuccess() && provider_type == ProviderType::GCS && client_settings.gcs_issue_compose_request)
     {
         /// For GCS we will try to compose object at the end, otherwise we cannot do a native copy
         /// for the object (e.g. for backups)

src/IO/S3/Client.h

@@ -97,6 +97,16 @@ struct ClientSettings
     bool use_virtual_addressing;
     /// Disable checksum to avoid extra read of the input stream
     bool disable_checksum;
+    /// Should client send ComposeObject request after upload to GCS.
+    ///
+    /// Previously ComposeObject request was required to make Copy possible,
+    /// but not anymore (see [1]).
+    ///
+    ///   [1]: https://cloud.google.com/storage/docs/release-notes#June_23_2023
+    ///
+    /// Ability to enable it preserved since likely it is required for old
+    /// files.
+    bool gcs_issue_compose_request;
 };
 
 /// Client that improves the client from the AWS SDK

src/IO/S3/tests/gtest_aws_s3_client.cpp

@@ -143,6 +143,7 @@ void testServerSideEncryption(
     DB::S3::ClientSettings client_settings{
         .use_virtual_addressing = uri.is_virtual_hosted_style,
         .disable_checksum = disable_checksum,
+        .gcs_issue_compose_request = false,
     };
 
     std::shared_ptr<DB::S3::Client> client = DB::S3::ClientFactory::instance().create(

src/IO/tests/gtest_writebuffer_s3.cpp

@@ -213,6 +213,7 @@ struct Client : DB::S3::Client
                DB::S3::ClientSettings{
                    .use_virtual_addressing = true,
                    .disable_checksum= false,
+                   .gcs_issue_compose_request = false,
                })
         , store(mock_s3_store)
     {}

src/Storages/StorageS3.cpp

@@ -1464,6 +1464,7 @@ void StorageS3::Configuration::connect(ContextPtr context)
     S3::ClientSettings client_settings{
         .use_virtual_addressing = url.is_virtual_hosted_style,
         .disable_checksum = local_settings.s3_disable_checksum,
+        .gcs_issue_compose_request = context->getConfigRef().getBool("s3.gcs_issue_compose_request", false),
     };
 
     auto credentials = Aws::Auth::AWSCredentials(auth_settings.access_key_id, auth_settings.secret_access_key, auth_settings.session_token);