Asynchronous WriteBuffer for AzureBlobStorage

src/Backups/BackupIO_AzureBlobStorage.cpp

@@ -275,10 +275,9 @@ std::unique_ptr<WriteBuffer> BackupWriterAzureBlobStorage::writeFile(const Strin
     return std::make_unique<WriteBufferFromAzureBlobStorage>(
         client,
         key,
-        settings->max_single_part_upload_size,
-        settings->max_unexpected_write_error_retries,
         DBMS_DEFAULT_BUFFER_SIZE,
-        write_settings);
+        write_settings,
+        settings);
 }
 
 void BackupWriterAzureBlobStorage::removeFile(const String & file_name)

src/IO/WriteBufferFromS3BufferAllocationPolicy.cpp → src/Common/BufferAllocationPolicy.cpp

@@ -1,21 +1,17 @@
-#include "config.h"
-
-#if USE_AWS_S3
-
-#include <IO/WriteBufferFromS3.h>
+#include "BufferAllocationPolicy.h"
 
 #include <memory>
 
-namespace
+namespace DB
 {
 
-class FixedSizeBufferAllocationPolicy : public DB::WriteBufferFromS3::IBufferAllocationPolicy
+class FixedSizeBufferAllocationPolicy : public IBufferAllocationPolicy
 {
     const size_t buffer_size = 0;
     size_t buffer_number = 0;
 
 public:
-    explicit FixedSizeBufferAllocationPolicy(const DB::S3Settings::RequestSettings::PartUploadSettings & settings_)
+    explicit FixedSizeBufferAllocationPolicy(const BufferAllocationSettings & settings_)
         : buffer_size(settings_.strict_upload_part_size)
     {
         chassert(buffer_size > 0);
@@ -36,7 +32,7 @@ class FixedSizeBufferAllocationPolicy : public DB::WriteBufferFromS3::IBufferAll
 };
 
 
-class ExpBufferAllocationPolicy : public DB::WriteBufferFromS3::IBufferAllocationPolicy
+class ExpBufferAllocationPolicy : public DB::IBufferAllocationPolicy
 {
     const size_t first_size = 0;
     const size_t second_size = 0;
@@ -49,7 +45,7 @@ class ExpBufferAllocationPolicy : public DB::WriteBufferFromS3::IBufferAllocatio
     size_t buffer_number = 0;
 
 public:
-    explicit ExpBufferAllocationPolicy(const DB::S3Settings::RequestSettings::PartUploadSettings & settings_)
+    explicit ExpBufferAllocationPolicy(const BufferAllocationSettings & settings_)
         : first_size(std::max(settings_.max_single_part_upload_size, settings_.min_upload_part_size))
         , second_size(settings_.min_upload_part_size)
         , multiply_factor(settings_.upload_part_size_multiply_factor)
@@ -92,14 +88,10 @@ class ExpBufferAllocationPolicy : public DB::WriteBufferFromS3::IBufferAllocatio
     }
 };
 
-}
-
-namespace DB
-{
 
-WriteBufferFromS3::IBufferAllocationPolicy::~IBufferAllocationPolicy() = default;
+IBufferAllocationPolicy::~IBufferAllocationPolicy() = default;
 
-WriteBufferFromS3::IBufferAllocationPolicyPtr WriteBufferFromS3::ChooseBufferPolicy(const S3Settings::RequestSettings::PartUploadSettings & settings_)
+IBufferAllocationPolicyPtr ChooseBufferPolicy(BufferAllocationSettings settings_)
 {
     if (settings_.strict_upload_part_size > 0)
         return std::make_unique<FixedSizeBufferAllocationPolicy>(settings_);
@@ -109,4 +101,3 @@ WriteBufferFromS3::IBufferAllocationPolicyPtr WriteBufferFromS3::ChooseBufferPol
 
 }
 
-#endif

src/Common/BufferAllocationPolicy.h

@@ -0,0 +1,35 @@
+#pragma once
+
+#include "config.h"
+
+#include "logger_useful.h"
+
+#include <list>
+
+namespace DB
+{
+
+struct BufferAllocationSettings
+{
+    size_t strict_upload_part_size = 0;
+    size_t min_upload_part_size = 16 * 1024 * 1024;
+    size_t max_upload_part_size = 5ULL * 1024 * 1024 * 1024;
+    size_t upload_part_size_multiply_factor = 2;
+    size_t upload_part_size_multiply_parts_count_threshold = 500;
+    size_t max_single_part_upload_size = 32 * 1024 * 1024;
+};
+
+class IBufferAllocationPolicy
+{
+    public:
+        virtual size_t getBufferNumber() const = 0;
+        virtual size_t getBufferSize() const = 0;
+        virtual void nextBuffer() = 0;
+        virtual ~IBufferAllocationPolicy() = 0;
+};
+
+using IBufferAllocationPolicyPtr = std::unique_ptr<IBufferAllocationPolicy>;
+
+IBufferAllocationPolicyPtr ChooseBufferPolicy(BufferAllocationSettings settings_);
+
+}

src/IO/WriteBufferFromS3TaskTracker.cpp → src/Common/ThreadPoolTaskTracker.cpp

@@ -1,8 +1,6 @@
 #include "config.h"
 
-#if USE_AWS_S3
-
-#include <IO/WriteBufferFromS3TaskTracker.h>
+#include "ThreadPoolTaskTracker.h"
 
 namespace ProfileEvents
 {
@@ -12,19 +10,19 @@ namespace ProfileEvents
 namespace DB
 {
 
-WriteBufferFromS3::TaskTracker::TaskTracker(ThreadPoolCallbackRunner<void> scheduler_, size_t max_tasks_inflight_, LogSeriesLimiterPtr limitedLog_)
+TaskTracker::TaskTracker(ThreadPoolCallbackRunner<void> scheduler_, size_t max_tasks_inflight_, LogSeriesLimiterPtr limitedLog_)
     : is_async(bool(scheduler_))
     , scheduler(scheduler_ ? std::move(scheduler_) : syncRunner())
     , max_tasks_inflight(max_tasks_inflight_)
     , limitedLog(limitedLog_)
 {}
 
-WriteBufferFromS3::TaskTracker::~TaskTracker()
+TaskTracker::~TaskTracker()
 {
     safeWaitAll();
 }
 
-ThreadPoolCallbackRunner<void> WriteBufferFromS3::TaskTracker::syncRunner()
+ThreadPoolCallbackRunner<void> TaskTracker::syncRunner()
 {
     return [](Callback && callback, int64_t) mutable -> std::future<void>
     {
@@ -35,7 +33,7 @@ ThreadPoolCallbackRunner<void> WriteBufferFromS3::TaskTracker::syncRunner()
     };
 }
 
-void WriteBufferFromS3::TaskTracker::waitAll()
+void TaskTracker::waitAll()
 {
     /// Exceptions are propagated
     for (auto & future : futures)
@@ -48,7 +46,7 @@ void WriteBufferFromS3::TaskTracker::waitAll()
     finished_futures.clear();
 }
 
-void WriteBufferFromS3::TaskTracker::safeWaitAll()
+void TaskTracker::safeWaitAll()
 {
     for (auto & future : futures)
     {
@@ -71,7 +69,7 @@ void WriteBufferFromS3::TaskTracker::safeWaitAll()
     finished_futures.clear();
 }
 
-void WriteBufferFromS3::TaskTracker::waitIfAny()
+void TaskTracker::waitIfAny()
 {
     if (futures.empty())
         return;
@@ -99,7 +97,7 @@ void WriteBufferFromS3::TaskTracker::waitIfAny()
     ProfileEvents::increment(ProfileEvents::WriteBufferFromS3WaitInflightLimitMicroseconds, watch.elapsedMicroseconds());
 }
 
-void WriteBufferFromS3::TaskTracker::add(Callback && func)
+void TaskTracker::add(Callback && func)
 {
     /// All this fuzz is about 2 things. This is the most critical place of TaskTracker.
     /// The first is not to fail insertion in the list `futures`.
@@ -134,7 +132,7 @@ void WriteBufferFromS3::TaskTracker::add(Callback && func)
     waitTilInflightShrink();
 }
 
-void WriteBufferFromS3::TaskTracker::waitTilInflightShrink()
+void TaskTracker::waitTilInflightShrink()
 {
     if (!max_tasks_inflight)
         return;
@@ -166,11 +164,10 @@ void WriteBufferFromS3::TaskTracker::waitTilInflightShrink()
     ProfileEvents::increment(ProfileEvents::WriteBufferFromS3WaitInflightLimitMicroseconds, watch.elapsedMicroseconds());
 }
 
-bool WriteBufferFromS3::TaskTracker::isAsync() const
+bool TaskTracker::isAsync() const
 {
     return is_async;
 }
 
 }
 
-#endif

src/IO/WriteBufferFromS3TaskTracker.h → src/Common/ThreadPoolTaskTracker.h

@@ -1,28 +1,24 @@
 #pragma once
 
 #include "config.h"
+#include "threadPoolCallbackRunner.h"
+#include "IO/WriteBufferFromS3.h"
 
-#if USE_AWS_S3
-
-#include "WriteBufferFromS3.h"
-
-#include <Common/logger_useful.h>
+#include "logger_useful.h"
 
 #include <list>
 
 namespace DB
 {
 
-/// That class is used only in WriteBufferFromS3 for now.
-/// Therefore it declared as a part of  WriteBufferFromS3.
 /// TaskTracker takes a Callback which is run by scheduler in some external shared ThreadPool.
 /// TaskTracker brings the methods waitIfAny, waitAll/safeWaitAll
 /// to help with coordination of the running tasks.
 
 /// Basic exception safety is provided. If exception occurred the object has to be destroyed.
 /// No thread safety is provided. Use this object with no concurrency.
 
-class WriteBufferFromS3::TaskTracker
+class TaskTracker
 {
 public:
     using Callback = std::function<void()>;
@@ -68,5 +64,3 @@ class WriteBufferFromS3::TaskTracker
 };
 
 }
-
-#endif

src/Core/Settings.h

@@ -78,11 +78,17 @@ class IColumn;
     M(UInt64, distributed_connections_pool_size, 1024, "Maximum number of connections with one remote server in the pool.", 0) \
     M(UInt64, connections_with_failover_max_tries, 3, "The maximum number of attempts to connect to replicas.", 0) \
     M(UInt64, s3_strict_upload_part_size, 0, "The exact size of part to upload during multipart upload to S3 (some implementations does not supports variable size parts).", 0) \
+    M(UInt64, azure_strict_upload_part_size, 0, "The exact size of part to upload during multipart upload to Azure blob storage.", 0) \
     M(UInt64, s3_min_upload_part_size, 16*1024*1024, "The minimum size of part to upload during multipart upload to S3.", 0) \
     M(UInt64, s3_max_upload_part_size, 5ull*1024*1024*1024, "The maximum size of part to upload during multipart upload to S3.", 0) \
+    M(UInt64, azure_min_upload_part_size, 16*1024*1024, "The minimum size of part to upload during multipart upload to Azure blob storage.", 0) \
+    M(UInt64, azure_max_upload_part_size, 5ull*1024*1024*1024, "The maximum size of part to upload during multipart upload to Azure blob storage.", 0) \
     M(UInt64, s3_upload_part_size_multiply_factor, 2, "Multiply s3_min_upload_part_size by this factor each time s3_multiply_parts_count_threshold parts were uploaded from a single write to S3.", 0) \
     M(UInt64, s3_upload_part_size_multiply_parts_count_threshold, 500, "Each time this number of parts was uploaded to S3, s3_min_upload_part_size is multiplied by s3_upload_part_size_multiply_factor.", 0) \
-    M(UInt64, s3_max_inflight_parts_for_one_file, 20, "The maximum number of a concurrent loaded parts in multipart upload request. 0 means unlimited. You ", 0) \
+    M(UInt64, azure_upload_part_size_multiply_factor, 2, "Multiply azure_min_upload_part_size by this factor each time azure_multiply_parts_count_threshold parts were uploaded from a single write to Azure blob storage.", 0) \
+    M(UInt64, azure_upload_part_size_multiply_parts_count_threshold, 500, "Each time this number of parts was uploaded to Azure blob storage, azure_min_upload_part_size is multiplied by azure_upload_part_size_multiply_factor.", 0) \
+    M(UInt64, s3_max_inflight_parts_for_one_file, 20, "The maximum number of a concurrent loaded parts in multipart upload request. 0 means unlimited.", 0) \
+    M(UInt64, azure_max_inflight_parts_for_one_file, 20, "The maximum number of a concurrent loaded parts in multipart upload request. 0 means unlimited.", 0) \
     M(UInt64, s3_max_single_part_upload_size, 32*1024*1024, "The maximum size of object to upload using singlepart upload to S3.", 0) \
     M(UInt64, azure_max_single_part_upload_size, 100*1024*1024, "The maximum size of object to upload using singlepart upload to Azure blob storage.", 0)                                                                             \
     M(UInt64, azure_max_single_part_copy_size, 256*1024*1024, "The maximum size of object to copy using single part copy to Azure blob storage.", 0) \

src/Core/SettingsChangesHistory.h

@@ -114,6 +114,7 @@ static std::map<ClickHouseVersion, SettingsChangesHistory::SettingsChanges> sett
               {"default_normal_view_sql_security", "INVOKER", "INVOKER", "Allows to set default `SQL SECURITY` option while creating a normal view"},
               {"mysql_map_string_to_text_in_show_columns", false, true, "Reduce the configuration effort to connect ClickHouse with BI tools."},
               {"mysql_map_fixed_string_to_text_in_show_columns", false, true, "Reduce the configuration effort to connect ClickHouse with BI tools."},
+              {"azure_max_inflight_parts_for_one_file", 20, 20, "The maximum number of a concurrent loaded parts in multipart upload request. 0 means unlimited."},
               }},
     {"24.1", {{"print_pretty_type_names", false, true, "Better user experience."},
               {"input_format_json_read_bools_as_strings", false, true, "Allow to read bools as strings in JSON formats by default"},