Add ability to throttle merges/mutations

azat · azat · commit 837f4ea67666 · 2023-12-24T22:31:49.000+01:00
Main motivation was to has an ability to throttle background tasks, to
avoid affecting queries.

To new server settings had been added for this:
- max_mutations_bandwidth_for_server
- max_merges_bandwidth_for_server

Note, that they limit only reading, since usually you will not write
more data then you read, but sometimes it is possible in case of ALTER
UPDATE.

But for now, to keep things simple, I decided to limit this with only
2 settings instead of 4.

Note, that if the write throttling will be needed, then they can use the
same settings, and just create new throttler for write.

Signed-off-by: Azat Khuzhin &lt;a.khuzhin@semrush.com&gt;
diff --git a/src/Core/ServerSettings.h b/src/Core/ServerSettings.h
@@ -26,6 +26,8 @@ namespace DB
     M(UInt64, max_active_parts_loading_thread_pool_size, 64, "The number of threads to load active set of data parts (Active ones) at startup.", 0) \
     M(UInt64, max_outdated_parts_loading_thread_pool_size, 32, "The number of threads to load inactive set of data parts (Outdated ones) at startup.", 0) \
     M(UInt64, max_parts_cleaning_thread_pool_size, 128, "The number of threads for concurrent removal of inactive data parts.", 0) \
+    M(UInt64, max_mutations_bandwidth_for_server, 0, "The maximum read speed of all mutations on server in bytes per second. Zero means unlimited.", 0) \
+    M(UInt64, max_merges_bandwidth_for_server, 0, "The maximum read speed of all merges on server in bytes per second. Zero means unlimited.", 0) \
     M(UInt64, max_replicated_fetches_network_bandwidth_for_server, 0, "The maximum speed of data exchange over the network in bytes per second for replicated fetches. Zero means unlimited.", 0) \
     M(UInt64, max_replicated_sends_network_bandwidth_for_server, 0, "The maximum speed of data exchange over the network in bytes per second for replicated sends. Zero means unlimited.", 0) \
     M(UInt64, max_remote_read_network_bandwidth_for_server, 0, "The maximum speed of data exchange over the network in bytes per second for read. Zero means unlimited.", 0) \
diff --git a/src/Interpreters/Context.cpp b/src/Interpreters/Context.cpp
@@ -325,6 +325,9 @@ struct ContextSharedPart : boost::noncopyable
 
     mutable ThrottlerPtr backups_server_throttler;          /// A server-wide throttler for BACKUPs
 
+    mutable ThrottlerPtr mutations_throttler;               /// A server-wide throttler for mutations
+    mutable ThrottlerPtr merges_throttler;                  /// A server-wide throttler for merges
+
     MultiVersion<Macros> macros;                            /// Substitutions extracted from config.
     std::unique_ptr<DDLWorker> ddl_worker TSA_GUARDED_BY(mutex); /// Process ddl commands from zk.
     LoadTaskPtr ddl_worker_startup_task;                         /// To postpone `ddl_worker->startup()` after all tables startup
@@ -733,6 +736,12 @@ struct ContextSharedPart : boost::noncopyable
 
         if (auto bandwidth = server_settings.max_backup_bandwidth_for_server)
             backups_server_throttler = std::make_shared<Throttler>(bandwidth);
+
+        if (auto bandwidth = server_settings.max_mutations_bandwidth_for_server)
+            mutations_throttler = std::make_shared<Throttler>(bandwidth);
+
+        if (auto bandwidth = server_settings.max_merges_bandwidth_for_server)
+            merges_throttler = std::make_shared<Throttler>(bandwidth);
     }
 };
 
@@ -2994,6 +3003,16 @@ ThrottlerPtr Context::getBackupsThrottler() const
     return throttler;
 }
 
+ThrottlerPtr Context::getMutationsThrottler() const
+{
+    return shared->mutations_throttler;
+}
+
+ThrottlerPtr Context::getMergesThrottler() const
+{
+    return shared->merges_throttler;
+}
+
 bool Context::hasDistributedDDL() const
 {
     return getConfigRef().has("distributed_ddl");
diff --git a/src/Interpreters/Context.h b/src/Interpreters/Context.h
@@ -1324,6 +1324,9 @@ class Context: public ContextData, public std::enable_shared_from_this<Context>
 
     ThrottlerPtr getBackupsThrottler() const;
 
+    ThrottlerPtr getMutationsThrottler() const;
+    ThrottlerPtr getMergesThrottler() const;
+
     /// Kitchen sink
     using ContextData::KitchenSink;
     using ContextData::kitchen_sink;
diff --git a/src/Interpreters/MutationsInterpreter.cpp b/src/Interpreters/MutationsInterpreter.cpp
@@ -1280,6 +1280,7 @@ void MutationsInterpreter::Source::read(
         VirtualColumns virtual_columns(std::move(required_columns), part);
 
         createReadFromPartStep(
+            MergeTreeSequentialSourceType::Mutation,
             plan, *data, storage_snapshot, part,
             std::move(virtual_columns.columns_to_read),
             apply_deleted_mask_, filter, context_,
diff --git a/src/Storages/MergeTree/MergeTask.cpp b/src/Storages/MergeTree/MergeTask.cpp
@@ -566,6 +566,7 @@ void MergeTask::VerticalMergeStage::prepareVerticalMergeForOneColumn() const
     for (size_t part_num = 0; part_num < global_ctx->future_part->parts.size(); ++part_num)
     {
         Pipe pipe = createMergeTreeSequentialSource(
+            MergeTreeSequentialSourceType::Merge,
             *global_ctx->data,
             global_ctx->storage_snapshot,
             global_ctx->future_part->parts[part_num],
@@ -920,6 +921,7 @@ void MergeTask::ExecuteAndFinalizeHorizontalPart::createMergedStream()
     for (const auto & part : global_ctx->future_part->parts)
     {
         Pipe pipe = createMergeTreeSequentialSource(
+            MergeTreeSequentialSourceType::Merge,
             *global_ctx->data,
             global_ctx->storage_snapshot,
             part,
diff --git a/src/Storages/MergeTree/MergeTreeSequentialSource.cpp b/src/Storages/MergeTree/MergeTreeSequentialSource.cpp
@@ -30,6 +30,7 @@ class MergeTreeSequentialSource : public ISource
 {
 public:
     MergeTreeSequentialSource(
+        MergeTreeSequentialSourceType type,
         const MergeTreeData & storage_,
         const StorageSnapshotPtr & storage_snapshot_,
         MergeTreeData::DataPartPtr data_part_,
@@ -85,6 +86,7 @@ class MergeTreeSequentialSource : public ISource
 
 
 MergeTreeSequentialSource::MergeTreeSequentialSource(
+    MergeTreeSequentialSourceType type,
     const MergeTreeData & storage_,
     const StorageSnapshotPtr & storage_snapshot_,
     MergeTreeData::DataPartPtr data_part_,
@@ -152,6 +154,17 @@ MergeTreeSequentialSource::MergeTreeSequentialSource(
     read_settings.local_fs_method = LocalFSReadMethod::pread;
     if (read_with_direct_io)
         read_settings.direct_io_threshold = 1;
+    /// Configure throttling
+    switch (type)
+    {
+        case Mutation:
+            read_settings.local_throttler = context->getMutationsThrottler();
+            break;
+        case Merge:
+            read_settings.local_throttler = context->getMergesThrottler();
+            break;
+    }
+    read_settings.remote_throttler = read_settings.local_throttler;
 
     MergeTreeReaderSettings reader_settings =
     {
@@ -244,6 +257,7 @@ MergeTreeSequentialSource::~MergeTreeSequentialSource() = default;
 
 
 Pipe createMergeTreeSequentialSource(
+    MergeTreeSequentialSourceType type,
     const MergeTreeData & storage,
     const StorageSnapshotPtr & storage_snapshot,
     MergeTreeData::DataPartPtr data_part,
@@ -264,7 +278,7 @@ Pipe createMergeTreeSequentialSource(
     if (need_to_filter_deleted_rows && !has_filter_column)
         columns_to_read.emplace_back(filter_column.name);
 
-    auto column_part_source = std::make_shared<MergeTreeSequentialSource>(
+    auto column_part_source = std::make_shared<MergeTreeSequentialSource>(type,
         storage, storage_snapshot, data_part, columns_to_read, std::move(mark_ranges),
         /*apply_deleted_mask=*/ false, read_with_direct_io, take_column_types_from_storage, quiet);
 
@@ -292,6 +306,7 @@ class ReadFromPart final : public ISourceStep
 {
 public:
     ReadFromPart(
+        MergeTreeSequentialSourceType type_,
         const MergeTreeData & storage_,
         const StorageSnapshotPtr & storage_snapshot_,
         MergeTreeData::DataPartPtr data_part_,
@@ -301,6 +316,7 @@ class ReadFromPart final : public ISourceStep
         ContextPtr context_,
         Poco::Logger * log_)
         : ISourceStep(DataStream{.header = storage_snapshot_->getSampleBlockForColumns(columns_to_read_)})
+        , type(type_)
         , storage(storage_)
         , storage_snapshot(storage_snapshot_)
         , data_part(std::move(data_part_))
@@ -337,7 +353,7 @@ class ReadFromPart final : public ISourceStep
             }
         }
 
-        auto source = createMergeTreeSequentialSource(
+        auto source = createMergeTreeSequentialSource(type,
             storage,
             storage_snapshot,
             data_part,
@@ -353,6 +369,7 @@ class ReadFromPart final : public ISourceStep
     }
 
 private:
+    MergeTreeSequentialSourceType type;
     const MergeTreeData & storage;
     StorageSnapshotPtr storage_snapshot;
     MergeTreeData::DataPartPtr data_part;
@@ -364,6 +381,7 @@ class ReadFromPart final : public ISourceStep
 };
 
 void createReadFromPartStep(
+    MergeTreeSequentialSourceType type,
     QueryPlan & plan,
     const MergeTreeData & storage,
     const StorageSnapshotPtr & storage_snapshot,
@@ -374,7 +392,7 @@ void createReadFromPartStep(
     ContextPtr context,
     Poco::Logger * log)
 {
-    auto reading = std::make_unique<ReadFromPart>(
+    auto reading = std::make_unique<ReadFromPart>(type,
         storage, storage_snapshot, std::move(data_part),
         std::move(columns_to_read), apply_deleted_mask,
         filter, std::move(context), log);
diff --git a/src/Storages/MergeTree/MergeTreeSequentialSource.h b/src/Storages/MergeTree/MergeTreeSequentialSource.h
@@ -8,9 +8,16 @@
 namespace DB
 {
 
+enum MergeTreeSequentialSourceType
+{
+    Mutation,
+    Merge,
+};
+
 /// Create stream for reading single part from MergeTree.
 /// If the part has lightweight delete mask then the deleted rows are filtered out.
 Pipe createMergeTreeSequentialSource(
+    MergeTreeSequentialSourceType type,
     const MergeTreeData & storage,
     const StorageSnapshotPtr & storage_snapshot,
     MergeTreeData::DataPartPtr data_part,
@@ -25,6 +32,7 @@ Pipe createMergeTreeSequentialSource(
 class QueryPlan;
 
 void createReadFromPartStep(
+    MergeTreeSequentialSourceType type,
     QueryPlan & plan,
     const MergeTreeData & storage,
     const StorageSnapshotPtr & storage_snapshot,
diff --git a/tests/integration/test_throttling/configs/static_overrides.xml b/tests/integration/test_throttling/configs/static_overrides.xml
@@ -31,4 +31,7 @@
         <allowed_disk>default</allowed_disk>
         <allowed_path>/backups/</allowed_path>
     </backups>
+
+    <max_mutations_bandwidth_for_server>1000000</max_mutations_bandwidth_for_server> <!-- 1M -->
+    <max_merges_bandwidth_for_server>1000000</max_merges_bandwidth_for_server> <!-- 1M -->
 </clickhouse>
diff --git a/tests/integration/test_throttling/test.py b/tests/integration/test_throttling/test.py
@@ -430,3 +430,32 @@ def test_write_throttling(policy, mode, setting, value, should_took):
     )
     _, took = elapsed(node.query, f"insert into data select * from numbers(1e6)")
     assert_took(took, should_took)
+
+
+def test_max_mutations_bandwidth_for_server():
+    node.query(
+        """
+        drop table if exists data;
+        create table data (key UInt64 CODEC(NONE)) engine=MergeTree() order by tuple() settings min_bytes_for_wide_part=1e9;
+    """
+    )
+    node.query("insert into data select * from numbers(1e6)")
+    _, took = elapsed(
+        node.query,
+        "alter table data update key = -key where 1 settings mutations_sync = 1",
+    )
+    # reading 1e6*8 bytes with 1M/s bandwith should take (8-1)/1=7 seconds
+    assert_took(took, 7)
+
+
+def test_max_merges_bandwidth_for_server():
+    node.query(
+        """
+        drop table if exists data;
+        create table data (key UInt64 CODEC(NONE)) engine=MergeTree() order by tuple() settings min_bytes_for_wide_part=1e9;
+    """
+    )
+    node.query("insert into data select * from numbers(1e6)")
+    _, took = elapsed(node.query, "optimize table data final")
+    # reading 1e6*8 bytes with 1M/s bandwith should take (8-1)/1=7 seconds
+    assert_took(took, 7)
