Even more checks and error messages

al13n321 · al13n321 · commit f0eae2c07efe · 2025-02-04T20:54:07.000Z
diff --git a/src/AggregateFunctions/UniqExactSet.h b/src/AggregateFunctions/UniqExactSet.h
@@ -65,8 +65,7 @@ class UniqExactSet
                 auto data_vec_atomic_index = std::make_shared<std::atomic_uint32_t>(0);
                 auto thread_func = [data_vec, data_vec_atomic_index, &is_cancelled, thread_group = CurrentThread::getGroup()]()
                 {
-                    ThreadGroupSwitcher switcher(thread_group);
-                    setThreadName("UniqExaConvert");
+                    ThreadGroupSwitcher switcher(thread_group, "UniqExaConvert");
 
                     while (true)
                     {
@@ -129,8 +128,7 @@ class UniqExactSet
 
                     auto thread_func = [&lhs, &rhs, next_bucket_to_merge, is_cancelled, thread_group = CurrentThread::getGroup()]()
                     {
-                        ThreadGroupSwitcher switcher(thread_group);
-                        setThreadName("UniqExactMerger");
+                        ThreadGroupSwitcher switcher(thread_group, "UniqExactMerger");
 
                         while (true)
                         {
diff --git a/src/Common/ThreadPool.cpp b/src/Common/ThreadPool.cpp
@@ -1,4 +1,6 @@
 #include <Common/ThreadPool.h>
+
+#include <Common/CurrentThread.h>
 #include <Common/ProfileEvents.h>
 #include <Common/setThreadName.h>
 #include <Common/Exception.h>
@@ -769,6 +771,11 @@ void ThreadPoolImpl<Thread>::ThreadFromThreadPool::worker()
 
             CurrentMetrics::Increment metric_active_pool_threads(parent_pool.metric_active_threads);
 
+#ifdef DEBUG_OR_SANITIZER_BUILD
+            DB::ThreadStatus * initial_thread = DB::current_thread;
+            DB::ThreadGroupPtr initial_thread_group = DB::CurrentThread::getGroup();
+#endif
+
             if constexpr (!std::is_same_v<Thread, std::thread>)
             {
                 Stopwatch watch;
@@ -785,6 +792,12 @@ void ThreadPoolImpl<Thread>::ThreadFromThreadPool::worker()
                 job_data->job();
             }
 
+#ifdef DEBUG_OR_SANITIZER_BUILD
+            DB::ThreadStatus * final_thread = DB::current_thread;
+            DB::ThreadGroupPtr final_thread_group = DB::CurrentThread::getGroup();
+            if (final_thread != initial_thread || final_thread_group != initial_thread_group)
+                throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Thread pool job changed current ThreadStatus pointer ({} -> {}) or ThreadGroup ({} -> {}).", initial_thread ? "non-nullptr" : "nullptr", final_thread ? "non-nullptr" : "nullptr", initial_thread_group ? "master_thread_id " + std::to_string(initial_thread_group->master_thread_id) : "nullptr", final_thread_group ? "master_thread_id " + std::to_string(final_thread_group->master_thread_id) : "nullptr");
+#endif
 
             if (thread_trace_context.root_span.isTraceEnabled())
             {
diff --git a/src/Common/ThreadStatus.h b/src/Common/ThreadStatus.h
@@ -144,7 +144,7 @@ class ThreadGroup
  * Typically used for inheriting thread group when scheduling tasks on a thread pool:
  *   pool->scheduleOrThrow([thread_group = CurrentThread::getGroup()]()
  *       {
- *           ThreadGroupSwitcher switcher(thread_group);
+ *           ThreadGroupSwitcher switcher(thread_group, "MyThread");
  *           ...
  *       });
  */
@@ -156,7 +156,8 @@ class ThreadGroupSwitcher : private boost::noncopyable
     ///  * If false, asserts that the thread is not already attached to a different group.
     ///    Use this when running a task in a thread pool.
     ///  * If true, remembers the current group and restores it in destructor.
-    explicit ThreadGroupSwitcher(ThreadGroupPtr thread_group_, bool allow_existing_group = false) noexcept;
+    /// If thread_name is not empty, calls setThreadName along the way; should be at most 15 bytes long.
+    explicit ThreadGroupSwitcher(ThreadGroupPtr thread_group_, std::string_view thread_name, bool allow_existing_group = false) noexcept;
     ~ThreadGroupSwitcher() noexcept;
 
 private:
diff --git a/src/Common/threadPoolCallbackRunner.h b/src/Common/threadPoolCallbackRunner.h
@@ -32,8 +32,7 @@ ThreadPoolCallbackRunnerUnsafe<Result, Callback> threadPoolCallbackRunnerUnsafe(
     {
         auto task = std::make_shared<std::packaged_task<Result()>>([thread_group, thread_name, my_callback = std::move(callback)]() mutable -> Result
         {
-            ThreadGroupSwitcher switcher(thread_group);
-            setThreadName(thread_name.data());
+            ThreadGroupSwitcher switcher(thread_group, thread_name);
 
             SCOPE_EXIT_SAFE(
             {
@@ -157,8 +156,7 @@ class ThreadPoolCallbackRunnerLocal final
 
         auto task_func = [task, thread_group = CurrentThread::getGroup(), my_thread_name = thread_name, my_callback = std::move(callback), promise]() mutable -> void
         {
-            ThreadGroupSwitcher switcher(thread_group);
-            setThreadName(my_thread_name.data());
+            ThreadGroupSwitcher switcher(thread_group, my_thread_name);
 
             TaskState expected = SCHEDULED;
             if (!task->state.compare_exchange_strong(expected, RUNNING))
diff --git a/src/Compression/ParallelCompressedWriteBuffer.cpp b/src/Compression/ParallelCompressedWriteBuffer.cpp
@@ -48,8 +48,7 @@ void ParallelCompressedWriteBuffer::nextImpl()
     current_buffer->uncompressed_size = offset();
     pool.scheduleOrThrowOnError([this, my_current_buffer = current_buffer, thread_group = CurrentThread::getGroup()]
     {
-        ThreadGroupSwitcher switcher(thread_group);
-        setThreadName("ParallelCompres");
+        ThreadGroupSwitcher switcher(thread_group, "ParallelCompres");
 
         compress(my_current_buffer);
     });
diff --git a/src/Dictionaries/HashedDictionary.h b/src/Dictionaries/HashedDictionary.h
@@ -336,8 +336,7 @@ HashedDictionary<dictionary_key_type, sparse, sharded>::~HashedDictionary()
 
         if (!pool.trySchedule([&container, thread_group = CurrentThread::getGroup()]
             {
-                ThreadGroupSwitcher switcher(thread_group);
-                setThreadName("HashedDictDtor");
+                ThreadGroupSwitcher switcher(thread_group, "HashedDictDtor");
 
                 /// Do not account memory that was occupied by the dictionaries for the query/user context.
                 MemoryTrackerBlockerInThread memory_blocker;
diff --git a/src/Dictionaries/HashedDictionaryParallelLoader.h b/src/Dictionaries/HashedDictionaryParallelLoader.h
@@ -68,8 +68,7 @@ class HashedDictionaryParallelLoader : public boost::noncopyable
             {
                 pool.scheduleOrThrowOnError([this, shard, thread_group = CurrentThread::getGroup()]
                 {
-                    ThreadGroupSwitcher switcher(thread_group);
-                    setThreadName("HashedDictLoad");
+                    ThreadGroupSwitcher switcher(thread_group, "HashedDictLoad");
 
                     WorkerStatistic statistic;
                     SCOPE_EXIT_SAFE(
diff --git a/src/Interpreters/Aggregator.cpp b/src/Interpreters/Aggregator.cpp
@@ -2356,7 +2356,7 @@ BlocksList Aggregator::prepareBlocksAndFillTwoLevelImpl(
             if (thread_pool)
                 thread_pool->scheduleOrThrowOnError([thread_id, &tasks, thread_group = CurrentThread::getGroup()]
                     {
-                        ThreadGroupSwitcher switcher(thread_group);
+                        ThreadGroupSwitcher switcher(thread_group, "");
                         tasks[thread_id]();
                     });
             else
@@ -3123,7 +3123,7 @@ void Aggregator::mergeBlocks(BucketToBlocks bucket_to_blocks, AggregatedDataVari
             if (thread_pool)
                 thread_pool->scheduleOrThrowOnError([bucket, &merge_bucket, aggregates_pool, thread_group = CurrentThread::getGroup()]
                 {
-                    ThreadGroupSwitcher switcher(thread_group);
+                    ThreadGroupSwitcher switcher(thread_group, "");
                     merge_bucket(bucket, aggregates_pool);
                 });
             else
diff --git a/src/Interpreters/ConcurrentHashJoin.cpp b/src/Interpreters/ConcurrentHashJoin.cpp
@@ -134,8 +134,7 @@ void reserveSpaceInHashMaps(
             pool->scheduleOrThrow(
                 [&, i, thread_group = CurrentThread::getGroup()]()
                 {
-                    ThreadGroupSwitcher switcher(thread_group);
-                    setThreadName("ConcurrentJoin");
+                    ThreadGroupSwitcher switcher(thread_group, "ConcurrentJoin");
 
                     const auto & right_data = getData(hash_joins[i]);
                     std::visit([&](auto & maps) { return reserve_space_in_buckets(maps, right_data->type, i); }, right_data->maps.at(0));
@@ -189,8 +188,7 @@ ConcurrentHashJoin::ConcurrentHashJoin(
             pool->scheduleOrThrow(
                 [&, i, thread_group = CurrentThread::getGroup()]()
                 {
-                    ThreadGroupSwitcher switcher(thread_group);
-                    setThreadName("ConcurrentJoin");
+                    ThreadGroupSwitcher switcher(thread_group, "ConcurrentJoin");
 
                     /// reserve is not needed anyway - either we will use fixed-size hash map or shared two-level map (then reserve will be done in a special way below)
                     const size_t reserve_size = 0;
@@ -239,8 +237,7 @@ ConcurrentHashJoin::~ConcurrentHashJoin()
             pool->scheduleOrThrow(
                 [join = hash_joins[0], i, this, thread_group = CurrentThread::getGroup()]()
                 {
-                    ThreadGroupSwitcher switcher(thread_group);
-                    setThreadName("ConcurrentJoin");
+                    ThreadGroupSwitcher switcher(thread_group, "ConcurrentJoin");
 
                     auto clear_space_in_buckets = [&](auto & maps, HashJoin::Type type, size_t idx)
                     {
diff --git a/src/Interpreters/ExternalLoader.cpp b/src/Interpreters/ExternalLoader.cpp
@@ -969,7 +969,7 @@ class ExternalLoader::LoadingDispatcher : private boost::noncopyable
     /// Does the loading, possibly in the separate thread.
     void doLoading(const String & name, size_t loading_id, bool forced_to_reload, size_t min_id_to_finish_loading_dependencies_, bool async, ThreadGroupPtr thread_group = {})
     {
-        ThreadGroupSwitcher switcher(thread_group);
+        ThreadGroupSwitcher switcher(thread_group, "");
 
         /// Do not account memory that was occupied by the dictionaries for the query/user context.
         MemoryTrackerBlockerInThread memory_blocker;
diff --git a/src/Interpreters/ThreadStatusExt.cpp b/src/Interpreters/ThreadStatusExt.cpp
@@ -168,7 +168,7 @@ void ThreadGroup::attachInternalProfileEventsQueue(const InternalProfileEventsQu
     shared_data.profile_queue_ptr = profile_queue;
 }
 
-ThreadGroupSwitcher::ThreadGroupSwitcher(ThreadGroupPtr thread_group_, bool allow_existing_group) noexcept : thread_group(std::move(thread_group_))
+ThreadGroupSwitcher::ThreadGroupSwitcher(ThreadGroupPtr thread_group_, std::string_view thread_name, bool allow_existing_group) noexcept : thread_group(std::move(thread_group_))
 {
     try
     {
@@ -182,7 +182,7 @@ ThreadGroupSwitcher::ThreadGroupSwitcher(ThreadGroupPtr thread_group_, bool allo
                 prev_thread_group = nullptr;
             }
             else if (!allow_existing_group)
-                throw Exception(ErrorCodes::LOGICAL_ERROR, "Thread is already attached to a group");
+                throw Exception(ErrorCodes::LOGICAL_ERROR, "Thread ({}) is already attached to a group (master_thread_id {})", thread_name, prev_thread_group->master_thread_id);
             else
                 CurrentThread::detachFromGroupIfNotDetached();
         }
@@ -191,11 +191,13 @@ ThreadGroupSwitcher::ThreadGroupSwitcher(ThreadGroupPtr thread_group_, bool allo
             return;
 
         if (!prev_thread)
-            throw Exception(ErrorCodes::LOGICAL_ERROR, "No current thread");
+            throw Exception(ErrorCodes::LOGICAL_ERROR, "Tried to attach thread ({}) to a group, but the ThreadStatus is not initialized", thread_name);
 
         LockMemoryExceptionInThread lock_memory_tracker(VariableContext::Global);
 
         CurrentThread::attachToGroup(thread_group);
+        if (!thread_name.empty())
+            setThreadName(thread_name.data());
     }
     catch (...)
     {
diff --git a/src/Processors/Executors/CompletedPipelineExecutor.cpp b/src/Processors/Executors/CompletedPipelineExecutor.cpp
@@ -37,8 +37,7 @@ static void threadFunction(
 {
     try
     {
-        ThreadGroupSwitcher switcher(thread_group);
-        setThreadName("QueryCompPipeEx");
+        ThreadGroupSwitcher switcher(thread_group, "QueryCompPipeEx");
 
         data.executor->execute(num_threads, concurrency_control);
     }
diff --git a/src/Processors/Executors/PipelineExecutor.cpp b/src/Processors/Executors/PipelineExecutor.cpp
@@ -403,8 +403,7 @@ void PipelineExecutor::spawnThreadsImpl()
         /// Start new thread
         pool->scheduleOrThrowOnError([this, thread_num, thread_group = CurrentThread::getGroup(), my_slot = std::move(slot)]
         {
-            ThreadGroupSwitcher switcher(thread_group);
-            setThreadName("QueryPipelineEx");
+            ThreadGroupSwitcher switcher(thread_group, "QueryPipelineEx");
 
             try
             {
diff --git a/src/Processors/Executors/PullingAsyncPipelineExecutor.cpp b/src/Processors/Executors/PullingAsyncPipelineExecutor.cpp
@@ -71,8 +71,7 @@ static void threadFunction(
 {
     try
     {
-        ThreadGroupSwitcher switcher(thread_group);
-        setThreadName("QueryPullPipeEx");
+        ThreadGroupSwitcher switcher(thread_group, "QueryPullPipeEx");
 
         data.executor->execute(num_threads, concurrency_control);
     }
diff --git a/src/Processors/Executors/PushingAsyncPipelineExecutor.cpp b/src/Processors/Executors/PushingAsyncPipelineExecutor.cpp
@@ -101,8 +101,7 @@ static void threadFunction(
 {
     try
     {
-        ThreadGroupSwitcher switcher(thread_group);
-        setThreadName("QueryPushPipeEx");
+        ThreadGroupSwitcher switcher(thread_group, "QueryPushPipeEx");
 
         data.executor->execute(num_threads, concurrency_control);
     }
diff --git a/src/Processors/Formats/Impl/DWARFBlockInputFormat.cpp b/src/Processors/Formats/Impl/DWARFBlockInputFormat.cpp
@@ -245,8 +245,7 @@ void DWARFBlockInputFormat::initializeIfNeeded()
             {
                 try
                 {
-                    ThreadGroupSwitcher switcher(thread_group);
-                    setThreadName("DWARFDecoder");
+                    ThreadGroupSwitcher switcher(thread_group, "DWARFDecoder");
 
                     std::unique_lock lock(mutex);
                     while (!units_queue.empty() && !is_stopped)
diff --git a/src/Processors/Formats/Impl/ParallelFormattingOutputFormat.cpp b/src/Processors/Formats/Impl/ParallelFormattingOutputFormat.cpp
@@ -129,8 +129,7 @@ namespace DB
 
     void ParallelFormattingOutputFormat::collectorThreadFunction(const ThreadGroupPtr & thread_group)
     {
-        ThreadGroupSwitcher switcher(thread_group);
-        setThreadName("Collector");
+        ThreadGroupSwitcher switcher(thread_group, "Collector");
 
         try
         {
@@ -195,8 +194,7 @@ namespace DB
 
     void ParallelFormattingOutputFormat::formatterThreadFunction(size_t current_unit_number, size_t first_row_num, const ThreadGroupPtr & thread_group)
     {
-        ThreadGroupSwitcher switcher(thread_group);
-        setThreadName("Formatter");
+        ThreadGroupSwitcher switcher(thread_group, "Formatter");
 
         try
         {
diff --git a/src/Processors/Formats/Impl/ParallelParsingInputFormat.cpp b/src/Processors/Formats/Impl/ParallelParsingInputFormat.cpp
@@ -10,8 +10,7 @@ namespace DB
 
 void ParallelParsingInputFormat::segmentatorThreadFunction(ThreadGroupPtr thread_group)
 {
-    ThreadGroupSwitcher switcher(thread_group);
-    setThreadName("Segmentator");
+    ThreadGroupSwitcher switcher(thread_group, "Segmentator");
 
     try
     {
@@ -61,8 +60,7 @@ void ParallelParsingInputFormat::segmentatorThreadFunction(ThreadGroupPtr thread
 
 void ParallelParsingInputFormat::parserThreadFunction(ThreadGroupPtr thread_group, size_t current_ticket_number)
 {
-    ThreadGroupSwitcher switcher(thread_group);
-    setThreadName("ChunkParser");
+    ThreadGroupSwitcher switcher(thread_group, "ChunkParser");
 
     const auto parser_unit_number = current_ticket_number % processing_units.size();
     auto & unit = processing_units[parser_unit_number];
diff --git a/src/Processors/Formats/Impl/ParquetBlockInputFormat.cpp b/src/Processors/Formats/Impl/ParquetBlockInputFormat.cpp
@@ -877,8 +877,7 @@ void ParquetBlockInputFormat::scheduleRowGroup(size_t row_group_batch_idx)
         {
             try
             {
-                ThreadGroupSwitcher switcher(thread_group);
-                setThreadName("ParquetDecoder");
+                ThreadGroupSwitcher switcher(thread_group, "ParquetDecoder");
 
                 threadFunction(row_group_batch_idx);
             }
diff --git a/src/Processors/Formats/Impl/ParquetBlockOutputFormat.cpp b/src/Processors/Formats/Impl/ParquetBlockOutputFormat.cpp
@@ -490,8 +490,7 @@ void ParquetBlockOutputFormat::startMoreThreadsIfNeeded(const std::unique_lock<s
         {
             try
             {
-                ThreadGroupSwitcher switcher(thread_group);
-                setThreadName("ParquetEncoder");
+                ThreadGroupSwitcher switcher(thread_group, "ParquetEncoder");
 
                 threadFunction();
             }
diff --git a/src/Processors/Formats/Impl/PrettyBlockOutputFormat.cpp b/src/Processors/Formats/Impl/PrettyBlockOutputFormat.cpp
@@ -159,8 +159,7 @@ void PrettyBlockOutputFormat::write(Chunk chunk, PortKind port_kind)
             {
                 thread.emplace([this, thread_group = CurrentThread::getGroup()]
                 {
-                    ThreadGroupSwitcher switcher(thread_group);
-                    setThreadName("PrettyWriter");
+                    ThreadGroupSwitcher switcher(thread_group, "PrettyWriter");
 
                     writingThread();
                 });
diff --git a/src/Processors/Transforms/AggregatingTransform.h b/src/Processors/Transforms/AggregatingTransform.h
@@ -109,8 +109,7 @@ struct ManyAggregatedData
                     pool->scheduleOrThrowOnError(
                         [my_variant = std::move(variant), thread_group = CurrentThread::getGroup()]() mutable
                         {
-                            ThreadGroupSwitcher switcher(thread_group);
-                            setThreadName("AggregDestruct");
+                            ThreadGroupSwitcher switcher(thread_group, "AggregDestruct");
 
                             my_variant.reset();
                         });
diff --git a/src/Storages/Distributed/DistributedSink.cpp b/src/Storages/Distributed/DistributedSink.cpp
@@ -331,8 +331,7 @@ DistributedSink::runWritingJob(JobReplica & job, const Block & current_block, si
         if (isCancelled())
             throw Exception(ErrorCodes::ABORTED, "Writing job was cancelled");
 
-        ThreadGroupSwitcher switcher(thread_group);
-        setThreadName("DistrOutStrProc");
+        ThreadGroupSwitcher switcher(thread_group, "DistrOutStrProc");
 
         OpenTelemetry::SpanHolder span(__PRETTY_FUNCTION__);
 
@@ -594,7 +593,7 @@ void DistributedSink::onFinish()
                     {
                         pool->scheduleOrThrowOnError([&job, thread_group = CurrentThread::getGroup()]()
                         {
-                            ThreadGroupSwitcher switcher(thread_group);
+                            ThreadGroupSwitcher switcher(thread_group, "");
 
                             job.executor->finish();
                         });
diff --git a/src/Storages/MergeTree/MergePlainMergeTreeTask.cpp b/src/Storages/MergeTree/MergePlainMergeTreeTask.cpp
@@ -39,7 +39,7 @@ bool MergePlainMergeTreeTask::executeStep()
     std::optional<ThreadGroupSwitcher> switcher;
     if (merge_list_entry)
     {
-        switcher.emplace((*merge_list_entry)->thread_group, /*allow_existing_group*/ true);
+        switcher.emplace((*merge_list_entry)->thread_group, "", /*allow_existing_group*/ true);
     }
 
     switch (state)
diff --git a/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp b/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp
@@ -803,8 +803,7 @@ RangesInDataParts MergeTreeDataSelectExecutor::filterPartsByPrimaryKeyAndSkipInd
                 pool.scheduleOrThrow(
                     [&, part_index, thread_group = CurrentThread::getGroup()]
                     {
-                        ThreadGroupSwitcher switcher(thread_group);
-                        setThreadName("MergeTreeIndex");
+                        ThreadGroupSwitcher switcher(thread_group, "MergeTreeIndex");
 
                         process_part(part_index);
                     },
diff --git a/src/Storages/MergeTree/MutatePlainMergeTreeTask.cpp b/src/Storages/MergeTree/MutatePlainMergeTreeTask.cpp
@@ -83,7 +83,7 @@ bool MutatePlainMergeTreeTask::executeStep()
     /// Make out memory tracker a parent of current thread memory tracker
     std::optional<ThreadGroupSwitcher> switcher;
     if (merge_list_entry)
-        switcher.emplace((*merge_list_entry)->thread_group, /*allow_existing_group*/ true);
+        switcher.emplace((*merge_list_entry)->thread_group, "", /*allow_existing_group*/ true);
 
     switch (state)
     {
diff --git a/src/Storages/MergeTree/ReplicatedMergeMutateTaskBase.cpp b/src/Storages/MergeTree/ReplicatedMergeMutateTaskBase.cpp
@@ -144,7 +144,7 @@ bool ReplicatedMergeMutateTaskBase::executeImpl()
 {
     std::optional<ThreadGroupSwitcher> switcher;
     if (merge_mutate_entry)
-        switcher.emplace((*merge_mutate_entry)->thread_group, /*allow_existing_group*/ true);
+        switcher.emplace((*merge_mutate_entry)->thread_group, "", /*allow_existing_group*/ true);
 
     auto remove_processed_entry = [&] () -> bool
     {
diff --git a/src/Storages/System/StorageSystemReplicas.cpp b/src/Storages/System/StorageSystemReplicas.cpp

Original file line number	Diff line number	Diff line change
`@@ -65,8 +65,7 @@ class UniqExactSet`
`65`	`65`	`auto data_vec_atomic_index = std::make_shared<std::atomic_uint32_t>(0);`
`66`	`66`	`auto thread_func = [data_vec, data_vec_atomic_index, &is_cancelled, thread_group = CurrentThread::getGroup()]()`
`67`	`67`	`{`
`68`		`- ThreadGroupSwitcher switcher(thread_group);`
`69`		`- setThreadName("UniqExaConvert");`
	`68`	`+ ThreadGroupSwitcher switcher(thread_group, "UniqExaConvert");`
`70`	`69`
`71`	`70`	`while (true)`
`72`	`71`	`{`
`@@ -129,8 +128,7 @@ class UniqExactSet`
`129`	`128`
`130`	`129`	`auto thread_func = [&lhs, &rhs, next_bucket_to_merge, is_cancelled, thread_group = CurrentThread::getGroup()]()`
`131`	`130`	`{`
`132`		`- ThreadGroupSwitcher switcher(thread_group);`
`133`		`- setThreadName("UniqExactMerger");`
	`131`	`+ ThreadGroupSwitcher switcher(thread_group, "UniqExactMerger");`
`134`	`132`
`135`	`133`	`while (true)`
`136`	`134`	`{`
Original file line number	Diff line number	Diff line change
`@@ -32,8 +32,7 @@ ThreadPoolCallbackRunnerUnsafe<Result, Callback> threadPoolCallbackRunnerUnsafe(`
`32`	`32`	`{`
`33`	`33`	`auto task = std::make_shared<std::packaged_task<Result()>>([thread_group, thread_name, my_callback = std::move(callback)]() mutable -> Result`
`34`	`34`	`{`
`35`		`- ThreadGroupSwitcher switcher(thread_group);`
`36`		`- setThreadName(thread_name.data());`
	`35`	`+ ThreadGroupSwitcher switcher(thread_group, thread_name);`
`37`	`36`
`38`	`37`	`SCOPE_EXIT_SAFE(`
`39`	`38`	`{`
`@@ -157,8 +156,7 @@ class ThreadPoolCallbackRunnerLocal final`
`157`	`156`
`158`	`157`	`auto task_func = [task, thread_group = CurrentThread::getGroup(), my_thread_name = thread_name, my_callback = std::move(callback), promise]() mutable -> void`
`159`	`158`	`{`
`160`		`- ThreadGroupSwitcher switcher(thread_group);`
`161`		`- setThreadName(my_thread_name.data());`
	`159`	`+ ThreadGroupSwitcher switcher(thread_group, my_thread_name);`
`162`	`160`
`163`	`161`	`TaskState expected = SCHEDULED;`
`164`	`162`	`if (!task->state.compare_exchange_strong(expected, RUNNING))`
Original file line number	Diff line number	Diff line change
`@@ -48,8 +48,7 @@ void ParallelCompressedWriteBuffer::nextImpl()`
`48`	`48`	`current_buffer->uncompressed_size = offset();`
`49`	`49`	`pool.scheduleOrThrowOnError([this, my_current_buffer = current_buffer, thread_group = CurrentThread::getGroup()]`
`50`	`50`	`{`
`51`		`- ThreadGroupSwitcher switcher(thread_group);`
`52`		`- setThreadName("ParallelCompres");`
	`51`	`+ ThreadGroupSwitcher switcher(thread_group, "ParallelCompres");`
`53`	`52`
`54`	`53`	`compress(my_current_buffer);`
`55`	`54`	`});`
Original file line number	Diff line number	Diff line change
`@@ -336,8 +336,7 @@ HashedDictionary<dictionary_key_type, sparse, sharded>::~HashedDictionary()`
`336`	`336`
`337`	`337`	`if (!pool.trySchedule([&container, thread_group = CurrentThread::getGroup()]`
`338`	`338`	`{`
`339`		`- ThreadGroupSwitcher switcher(thread_group);`
`340`		`- setThreadName("HashedDictDtor");`
	`339`	`+ ThreadGroupSwitcher switcher(thread_group, "HashedDictDtor");`
`341`	`340`
`342`	`341`	`/// Do not account memory that was occupied by the dictionaries for the query/user context.`
`343`	`342`	`MemoryTrackerBlockerInThread memory_blocker;`
Original file line number	Diff line number	Diff line change
`@@ -68,8 +68,7 @@ class HashedDictionaryParallelLoader : public boost::noncopyable`
`68`	`68`	`{`
`69`	`69`	`pool.scheduleOrThrowOnError([this, shard, thread_group = CurrentThread::getGroup()]`
`70`	`70`	`{`
`71`		`- ThreadGroupSwitcher switcher(thread_group);`
`72`		`- setThreadName("HashedDictLoad");`
	`71`	`+ ThreadGroupSwitcher switcher(thread_group, "HashedDictLoad");`
`73`	`72`
`74`	`73`	`WorkerStatistic statistic;`
`75`	`74`	`SCOPE_EXIT_SAFE(`
Original file line number	Diff line number	Diff line change
`@@ -969,7 +969,7 @@ class ExternalLoader::LoadingDispatcher : private boost::noncopyable`
`969`	`969`	`/// Does the loading, possibly in the separate thread.`
`970`	`970`	`void doLoading(const String & name, size_t loading_id, bool forced_to_reload, size_t min_id_to_finish_loading_dependencies_, bool async, ThreadGroupPtr thread_group = {})`
`971`	`971`	`{`
`972`		`- ThreadGroupSwitcher switcher(thread_group);`
	`972`	`+ ThreadGroupSwitcher switcher(thread_group, "");`
`973`	`973`
`974`	`974`	`/// Do not account memory that was occupied by the dictionaries for the query/user context.`
`975`	`975`	`MemoryTrackerBlockerInThread memory_blocker;`