Miscellaneous

alexey-milovidov · alexey-milovidov · commit 779dcb8555f1 · 2024-03-08T07:20:40.000+01:00
diff --git a/src/Interpreters/ThreadStatusExt.cpp b/src/Interpreters/ThreadStatusExt.cpp
@@ -151,15 +151,21 @@ ThreadGroupSwitcher::ThreadGroupSwitcher(ThreadGroupPtr thread_group)
     /// might be nullptr
     prev_thread_group = CurrentThread::getGroup();
 
-    CurrentThread::detachFromGroupIfNotDetached();
-    CurrentThread::attachToGroup(thread_group);
+    if (thread_group != prev_thread_group)
+    {
+        CurrentThread::detachFromGroupIfNotDetached();
+        CurrentThread::attachToGroup(thread_group);
+    }
 }
 
 ThreadGroupSwitcher::~ThreadGroupSwitcher()
 {
-    CurrentThread::detachFromGroupIfNotDetached();
-    if (prev_thread_group)
-        CurrentThread::attachToGroup(prev_thread_group);
+    if (CurrentThread::getGroup() != prev_thread_group)
+    {
+        CurrentThread::detachFromGroupIfNotDetached();
+        if (prev_thread_group)
+            CurrentThread::attachToGroup(prev_thread_group);
+    }
 }
 
 void ThreadStatus::attachInternalProfileEventsQueue(const InternalProfileEventsQueuePtr & profile_queue)
diff --git a/src/Storages/MergeTree/MergePlainMergeTreeTask.cpp b/src/Storages/MergeTree/MergePlainMergeTreeTask.cpp
@@ -38,19 +38,17 @@ bool MergePlainMergeTreeTask::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);
-    }
 
     switch (state)
     {
-        case State::NEED_PREPARE :
+        case State::NEED_PREPARE:
         {
             prepare();
             state = State::NEED_EXECUTE;
             return true;
         }
-        case State::NEED_EXECUTE :
+        case State::NEED_EXECUTE:
         {
             try
             {
@@ -66,7 +64,7 @@ bool MergePlainMergeTreeTask::executeStep()
                 throw;
             }
         }
-        case State::NEED_FINISH :
+        case State::NEED_FINISH:
         {
             finish();
 
diff --git a/src/Storages/MergeTree/MergeTask.h b/src/Storages/MergeTree/MergeTask.h
@@ -56,7 +56,6 @@ using MergeTaskPtr = std::shared_ptr<MergeTask>;
 class MergeTask
 {
 public:
-
     MergeTask(
         FutureMergedMutatedPartPtr future_part_,
         StorageMetadataPtr metadata_snapshot_,
diff --git a/src/Storages/MergeTree/MutateTask.cpp b/src/Storages/MergeTree/MutateTask.cpp
@@ -1005,7 +1005,6 @@ using MutationContextPtr = std::shared_ptr<MutationContext>;
 class MergeProjectionPartsTask : public IExecutableTask
 {
 public:
-
     MergeProjectionPartsTask(
         String name_,
         MergeTreeData::MutableDataPartsVector && parts_,
diff --git a/src/Storages/MergeTree/ReplicatedMergeMutateTaskBase.cpp b/src/Storages/MergeTree/ReplicatedMergeMutateTaskBase.cpp
@@ -161,7 +161,6 @@ bool ReplicatedMergeMutateTaskBase::executeImpl()
         return false;
     };
 
-
     auto execute_fetch = [&] (bool need_to_check_missing_part) -> bool
     {
         if (storage.executeFetch(entry, need_to_check_missing_part))
@@ -170,10 +169,9 @@ bool ReplicatedMergeMutateTaskBase::executeImpl()
         return false;
     };
 
-
     switch (state)
     {
-        case State::NEED_PREPARE :
+        case State::NEED_PREPARE:
         {
             {
                 auto res = checkExistingPart();
@@ -193,7 +191,7 @@ bool ReplicatedMergeMutateTaskBase::executeImpl()
             state = State::NEED_EXECUTE_INNER_MERGE;
             return true;
         }
-        case State::NEED_EXECUTE_INNER_MERGE :
+        case State::NEED_EXECUTE_INNER_MERGE:
         {
             try
             {
@@ -212,7 +210,7 @@ bool ReplicatedMergeMutateTaskBase::executeImpl()
 
             return true;
         }
-        case State::NEED_FINALIZE :
+        case State::NEED_FINALIZE:
         {
             try
             {
@@ -228,7 +226,7 @@ bool ReplicatedMergeMutateTaskBase::executeImpl()
 
             return remove_processed_entry();
         }
-        case State::SUCCESS :
+        case State::SUCCESS:
         {
             throw Exception(ErrorCodes::LOGICAL_ERROR, "Do not call execute on previously succeeded task");
         }

Original file line number	Diff line number	Diff line change
`@@ -38,19 +38,17 @@ bool MergePlainMergeTreeTask::executeStep()`
`38`	`38`	`/// Make out memory tracker a parent of current thread memory tracker`
`39`	`39`	`std::optional<ThreadGroupSwitcher> switcher;`
`40`	`40`	`if (merge_list_entry)`
`41`		`- {`
`42`	`41`	`switcher.emplace((*merge_list_entry)->thread_group);`
`43`		`- }`
`44`	`42`
`45`	`43`	`switch (state)`
`46`	`44`	`{`
`47`		`- case State::NEED_PREPARE :`
	`45`	`+ case State::NEED_PREPARE:`
`48`	`46`	`{`
`49`	`47`	`prepare();`
`50`	`48`	`state = State::NEED_EXECUTE;`
`51`	`49`	`return true;`
`52`	`50`	`}`
`53`		`- case State::NEED_EXECUTE :`
	`51`	`+ case State::NEED_EXECUTE:`
`54`	`52`	`{`
`55`	`53`	`try`
`56`	`54`	`{`
`@@ -66,7 +64,7 @@ bool MergePlainMergeTreeTask::executeStep()`
`66`	`64`	`throw;`
`67`	`65`	`}`
`68`	`66`	`}`
`69`		`- case State::NEED_FINISH :`
	`67`	`+ case State::NEED_FINISH:`
`70`	`68`	`{`
`71`	`69`	`finish();`
`72`	`70`
Original file line number	Diff line number	Diff line change
`@@ -56,7 +56,6 @@ using MergeTaskPtr = std::shared_ptr<MergeTask>;`
`56`	`56`	`class MergeTask`
`57`	`57`	`{`
`58`	`58`	`public:`
`59`		`-`
`60`	`59`	`MergeTask(`
`61`	`60`	`FutureMergedMutatedPartPtr future_part_,`
`62`	`61`	`StorageMetadataPtr metadata_snapshot_,`
Original file line number	Diff line number	Diff line change
`@@ -1005,7 +1005,6 @@ using MutationContextPtr = std::shared_ptr<MutationContext>;`
`1005`	`1005`	`class MergeProjectionPartsTask : public IExecutableTask`
`1006`	`1006`	`{`
`1007`	`1007`	`public:`
`1008`		`-`
`1009`	`1008`	`MergeProjectionPartsTask(`
`1010`	`1009`	`String name_,`
`1011`	`1010`	`MergeTreeData::MutableDataPartsVector && parts_,`
Original file line number	Diff line number	Diff line change
`@@ -161,7 +161,6 @@ bool ReplicatedMergeMutateTaskBase::executeImpl()`
`161`	`161`	`return false;`
`162`	`162`	`};`
`163`	`163`
`164`		`-`
`165`	`164`	`auto execute_fetch = [&] (bool need_to_check_missing_part) -> bool`
`166`	`165`	`{`
`167`	`166`	`if (storage.executeFetch(entry, need_to_check_missing_part))`
`@@ -170,10 +169,9 @@ bool ReplicatedMergeMutateTaskBase::executeImpl()`
`170`	`169`	`return false;`
`171`	`170`	`};`
`172`	`171`
`173`		`-`
`174`	`172`	`switch (state)`
`175`	`173`	`{`
`176`		`- case State::NEED_PREPARE :`
	`174`	`+ case State::NEED_PREPARE:`
`177`	`175`	`{`
`178`	`176`	`{`
`179`	`177`	`auto res = checkExistingPart();`
`@@ -193,7 +191,7 @@ bool ReplicatedMergeMutateTaskBase::executeImpl()`
`193`	`191`	`state = State::NEED_EXECUTE_INNER_MERGE;`
`194`	`192`	`return true;`
`195`	`193`	`}`
`196`		`- case State::NEED_EXECUTE_INNER_MERGE :`
	`194`	`+ case State::NEED_EXECUTE_INNER_MERGE:`
`197`	`195`	`{`
`198`	`196`	`try`
`199`	`197`	`{`
`@@ -212,7 +210,7 @@ bool ReplicatedMergeMutateTaskBase::executeImpl()`
`212`	`210`
`213`	`211`	`return true;`
`214`	`212`	`}`
`215`		`- case State::NEED_FINALIZE :`
	`213`	`+ case State::NEED_FINALIZE:`
`216`	`214`	`{`
`217`	`215`	`try`
`218`	`216`	`{`
`@@ -228,7 +226,7 @@ bool ReplicatedMergeMutateTaskBase::executeImpl()`
`228`	`226`
`229`	`227`	`return remove_processed_entry();`
`230`	`228`	`}`
`231`		`- case State::SUCCESS :`
	`229`	`+ case State::SUCCESS:`
`232`	`230`	`{`
`233`	`231`	`throw Exception(ErrorCodes::LOGICAL_ERROR, "Do not call execute on previously succeeded task");`
`234`	`232`	`}`