Merge pull request ClickHouse#77671 from azat/faster-object-queues

Improve performance of S3Queue/AzureQueue by allowing INSERTs data in parallel

Author: azat

docs/en/engines/table-engines/integrations/azure-queue.md

@@ -82,7 +82,7 @@ For more information about virtual columns see [here](../../../engines/table-eng
 
 ## Introspection {#introspection}
 
-Enable logging for the table via the table setting `enable_logging_to_s3queue_log=1`.
+Enable logging for the table via the table setting `enable_logging_to_queue_log=1`.
 
 Introspection capabilities are the same as the [S3Queue table engine](/engines/table-engines/integrations/s3queue#introspection) with several distinct differences:
 

docs/en/engines/table-engines/integrations/s3queue.md

@@ -24,15 +24,24 @@ CREATE TABLE s3_queue_engine_table (name String, value UInt32)
     [after_processing = 'keep',]
     [keeper_path = '',]
     [loading_retries = 0,]
-    [processing_threads_num = 1,]
-    [enable_logging_to_s3queue_log = 0,]
+    [processing_threads_num = 16,]
+    [parallel_inserts = false,]
+    [enable_logging_to_queue_log = true,]
+    [last_processed_path = "",]
+    [tracked_files_limit = 1000,]
+    [tracked_file_ttl_sec = 0,]
     [polling_min_timeout_ms = 1000,]
     [polling_max_timeout_ms = 10000,]
     [polling_backoff_ms = 0,]
-    [tracked_file_ttl_sec = 0,]
-    [tracked_files_limit = 1000,]
     [cleanup_interval_min_ms = 10000,]
     [cleanup_interval_max_ms = 30000,]
+    [buckets = 0,]
+    [list_objects_batch_size = 1000,]
+    [enable_hash_ring_filtering = 0,]
+    [max_processed_files_before_commit = 100,]
+    [max_processed_rows_before_commit = 0,]
+    [max_processed_bytes_before_commit = 0,]
+    [max_processing_time_sec_before_commit = 0,]
 ```
 
 :::warning
@@ -118,7 +127,18 @@ Default value: `0`.
 
 Number of threads to perform processing. Applies only for `Unordered` mode.
 
-Default value: `1`.
+Default value: Number of CPUs or 16.
+
+### s3queue_parallel_inserts {#parallel_inserts}
+
+By default `processing_threads_num` will produce one `INSERT`, so it will only download files and parse in multiple threads.
+But this limits the parallelism, so for better throughput use `parallel_inserts=true`, this will allow to insert data in parallel (but keep in mind that it will result in higher number of generated data parts for MergeTree family).
+
+:::note
+`INSERT`s will be spawned with respect to `max_process*_before_commit` settings.
+:::
+
+Default value: `false`.
 
 ### s3queue_enable_logging_to_s3queue_log {#enable_logging_to_s3queue_log}
 

src/Storages/ObjectStorageQueue/ObjectStorageQueueSettings.cpp

@@ -26,7 +26,8 @@ namespace ErrorCodes
     DECLARE(ObjectStorageQueueAction, after_processing, ObjectStorageQueueAction::KEEP, "Delete or keep file in after successful processing", 0) \
     DECLARE(String, keeper_path, "", "Zookeeper node path", 0) \
     DECLARE(UInt64, loading_retries, 10, "Retry loading up to specified number of times", 0) \
-    DECLARE(UInt64, processing_threads_num, 1, "Number of processing threads", 0) \
+    DECLARE(UInt64, processing_threads_num, 1, "Number of processing threads (default number of available CPUs or 16)", 0) \
+    DECLARE(Bool, parallel_inserts, false, "By default processing_threads_num will produce one INSERT, but this limits the parallel execution, so better scalability enable this setting (note this will create not one INSERT per file, but one per max_process*_before_commit)", 0) \
     DECLARE(UInt32, enable_logging_to_queue_log, 1, "Enable logging to system table system.(s3/azure_)queue_log", 0) \
     DECLARE(String, last_processed_path, "", "For Ordered mode. Files that have lexicographically smaller file name are considered already processed", 0) \
     DECLARE(UInt64, tracked_files_limit, 1000, "For unordered mode. Max set size for tracking processed files in ZooKeeper", 0) \
@@ -37,12 +38,12 @@ namespace ErrorCodes
     DECLARE(UInt32, cleanup_interval_min_ms, 60000, "For unordered mode. Polling backoff min for cleanup", 0) \
     DECLARE(UInt32, cleanup_interval_max_ms, 60000, "For unordered mode. Polling backoff max for cleanup", 0) \
     DECLARE(UInt64, buckets, 0, "Number of buckets for Ordered mode parallel processing", 0) \
-    DECLARE(UInt64, list_objects_batch_size, 1000, "Size of a list batcn in object storage", 0) \
+    DECLARE(UInt64, list_objects_batch_size, 1000, "Size of a list batch in object storage", 0) \
     DECLARE(Bool, enable_hash_ring_filtering, 0, "Enable filtering files among replicas according to hash ring for Unordered mode", 0) \
-    DECLARE(UInt64, max_processed_files_before_commit, 100, "Number of files which can be processed before being committed to keeper", 0) \
-    DECLARE(UInt64, max_processed_rows_before_commit, 0, "Number of rows which can be processed before being committed to keeper", 0) \
-    DECLARE(UInt64, max_processed_bytes_before_commit, 0, "Number of bytes which can be processed before being committed to keeper", 0) \
-    DECLARE(UInt64, max_processing_time_sec_before_commit, 0, "Timeout in seconds after which to commit files committed to keeper", 0) \
+    DECLARE(UInt64, max_processed_files_before_commit, 100, "Number of files which can be processed before being committed to keeper (in case of parallel_inserts=true, works on a per-thread basis)", 0) \
+    DECLARE(UInt64, max_processed_rows_before_commit, 0, "Number of rows which can be processed before being committed to keeper (in case of parallel_inserts=true, works on a per-thread basis)", 0) \
+    DECLARE(UInt64, max_processed_bytes_before_commit, 0, "Number of bytes which can be processed before being committed to keeper (in case of parallel_inserts=true, works on a per-thread basis)", 0) \
+    DECLARE(UInt64, max_processing_time_sec_before_commit, 0, "Timeout in seconds after which to commit files committed to keeper (in case of parallel_inserts=true, works on a per-thread basis)", 0) \
 
 #define LIST_OF_OBJECT_STORAGE_QUEUE_SETTINGS(M, ALIAS) \
     OBJECT_STORAGE_QUEUE_RELATED_SETTINGS(M, ALIAS) \

src/Storages/ObjectStorageQueue/ObjectStorageQueueSource.cpp

@@ -129,16 +129,18 @@ ObjectStorageQueueSource::FileIterator::FileIterator(
     }
 }
 
-bool ObjectStorageQueueSource::FileIterator::isFinished() const
+bool ObjectStorageQueueSource::FileIterator::isFinished()
 {
-     LOG_TEST(log, "Iterator finished: {}, objects to retry: {}", iterator_finished.load(), objects_to_retry.size());
-     return iterator_finished
-         && std::all_of(listed_keys_cache.begin(), listed_keys_cache.end(), [](const auto & v) { return v.second.keys.empty(); })
-         && objects_to_retry.empty();
+    std::lock_guard lock(mutex);
+    LOG_TEST(log, "Iterator finished: {}, objects to retry: {}", iterator_finished.load(), objects_to_retry.size());
+    return iterator_finished
+        && std::all_of(listed_keys_cache.begin(), listed_keys_cache.end(), [](const auto & v) { return v.second.keys.empty(); })
+        && objects_to_retry.empty();
 }
 
 size_t ObjectStorageQueueSource::FileIterator::estimatedKeysCount()
 {
+    std::lock_guard lock(next_mutex);
     /// Copied from StorageObjectStorageSource::estimateKeysCount().
     if (object_infos.empty() && !is_finished && object_storage_iterator->isValid())
         return std::numeric_limits<size_t>::max();
@@ -456,6 +458,7 @@ void ObjectStorageQueueSource::FileIterator::returnForRetry(ObjectInfoPtr object
 
 void ObjectStorageQueueSource::FileIterator::releaseFinishedBuckets()
 {
+    std::lock_guard lock(mutex);
     for (const auto & [processor, holders] : bucket_holders)
     {
         LOG_TEST(log, "Releasing {} bucket holders for processor {}", holders.size(), processor);
@@ -870,7 +873,6 @@ Chunk ObjectStorageQueueSource::generateImpl()
 
             if (commit_settings.max_processed_files_before_commit)
             {
-                std::lock_guard lock(progress->processed_files_mutex);
                 if (progress->processed_files.load(std::memory_order_relaxed) >= commit_settings.max_processed_files_before_commit)
                 {
                     LOG_TRACE(log, "Number of max processed files before commit reached "

src/Storages/ObjectStorageQueue/ObjectStorageQueueSource.h

@@ -7,6 +7,7 @@
 #include <Storages/ObjectStorage/StorageObjectStorageSource.h>
 #include <Storages/ObjectStorageQueue/ObjectStorageQueueMetadata.h>
 #include <Storages/ObjectStorageQueue/ObjectStorageQueueSettings.h>
+#include <base/defines.h>
 #include <Common/ZooKeeper/ZooKeeper.h>
 
 
@@ -52,7 +53,7 @@ class ObjectStorageQueueSource : public ISource, WithContext
             bool file_deletion_on_processed_enabled_,
             std::atomic<bool> & shutdown_called_);
 
-        bool isFinished() const;
+        bool isFinished();
 
         ObjectInfoPtr next(size_t processor) override;
 
@@ -86,7 +87,7 @@ class ObjectStorageQueueSource : public ISource, WithContext
         ExpressionActionsPtr filter_expr;
         bool recursive{false};
 
-        Source::ObjectInfos object_infos;
+        Source::ObjectInfos object_infos TSA_GUARDED_BY(next_mutex);
         std::vector<FileMetadataPtr> file_metadatas;
         bool is_finished = false;
         std::mutex next_mutex;
@@ -106,24 +107,24 @@ class ObjectStorageQueueSource : public ISource, WithContext
             std::optional<Processor> processor;
         };
         /// A cache of keys which were iterated via glob_iterator, but not taken for processing.
-        std::unordered_map<Bucket, ListedKeys> listed_keys_cache;
+        std::unordered_map<Bucket, ListedKeys> listed_keys_cache TSA_GUARDED_BY(mutex);
 
         /// We store a vector of holders, because we cannot release them until processed files are committed.
-        std::unordered_map<size_t, std::vector<BucketHolderPtr>> bucket_holders;
+        std::unordered_map<size_t, std::vector<BucketHolderPtr>> bucket_holders TSA_GUARDED_BY(mutex);
 
         /// Is glob_iterator finished?
         std::atomic_bool iterator_finished = false;
 
         /// Only for processing without buckets.
-        std::deque<std::pair<ObjectInfoPtr, FileMetadataPtr>> objects_to_retry;
+        std::deque<std::pair<ObjectInfoPtr, FileMetadataPtr>> objects_to_retry TSA_GUARDED_BY(mutex);
 
         struct NextKeyFromBucket
         {
             ObjectInfoPtr object_info;
             FileMetadataPtr file_metadata;
             ObjectStorageQueueOrderedFileMetadata::BucketInfoPtr bucket_info;
         };
-        NextKeyFromBucket getNextKeyFromAcquiredBucket(size_t processor);
+        NextKeyFromBucket getNextKeyFromAcquiredBucket(size_t processor) TSA_REQUIRES(mutex);
         bool hasKeysForProcessor(const Processor & processor) const;
     };
 
@@ -141,8 +142,6 @@ class ObjectStorageQueueSource : public ISource, WithContext
         std::atomic<size_t> processed_rows = 0;
         std::atomic<size_t> processed_bytes = 0;
         Stopwatch elapsed_time{CLOCK_MONOTONIC_COARSE};
-
-        std::mutex processed_files_mutex;
     };
     using ProcessingProgressPtr = std::shared_ptr<ProcessingProgress>;
 

src/Storages/ObjectStorageQueue/ObjectStorageQueueTableMetadata.h

@@ -28,6 +28,7 @@ struct ObjectStorageQueueTableMetadata
     std::atomic<ObjectStorageQueueAction> after_processing;
     std::atomic<UInt64> loading_retries;
     std::atomic<UInt64> processing_threads_num;
+    std::atomic<bool> parallel_inserts;
     std::atomic<UInt64> tracked_files_limit;
     std::atomic<UInt64> tracked_files_ttl_sec;
     std::atomic<UInt64> buckets;
@@ -47,6 +48,7 @@ struct ObjectStorageQueueTableMetadata
         , after_processing(other.after_processing.load())
         , loading_retries(other.loading_retries.load())
         , processing_threads_num(other.processing_threads_num.load())
+        , parallel_inserts(other.parallel_inserts.load())
         , tracked_files_limit(other.tracked_files_limit.load())
         , tracked_files_ttl_sec(other.tracked_files_ttl_sec.load())
         , buckets(other.buckets.load())
@@ -89,6 +91,7 @@ struct ObjectStorageQueueTableMetadata
             "after_processing",
             "loading_retries",
             "processing_threads_num",
+            "parallel_inserts",
             "tracked_files_limit",
             "tracked_file_ttl_sec",
             "tracked_files_ttl_sec",