Merge pull request ClickHouse#86468 from ClickHouse/backport/25.8/86422

Backport ClickHouse#86422 to 25.8: Fix leaking of MergesMutationsMemoryTracking and fix query_views_log for streaming

Author: clickhouse-gh[bot]

src/Common/ThreadStatus.h

@@ -117,9 +117,11 @@ class ThreadGroup
     /// When new query starts, new thread group is created for it, current thread becomes master thread of the query
     static ThreadGroupPtr createForQuery(ContextPtr query_context_, FatalErrorCallback fatal_error_callback_ = {});
 
-    static ThreadGroupPtr createForBackgroundProcess(ContextPtr storage_context);
+    /// NOTE: The caller should call background_memory_tracker.adjustOnBackgroundTaskEnd() at the end (see existing callers),
+    /// and make sure that you are the only user of this shared_ptr (usually it is managed via ThreadGroupSwitcher)
+    static ThreadGroupPtr createForMergeMutate(ContextPtr storage_context);
 
-    static ThreadGroupPtr createForMaterializedView();
+    static ThreadGroupPtr createForMaterializedView(ContextPtr context);
 
     std::vector<UInt64> getInvolvedThreadIds() const;
     size_t getPeakThreadsUsage() const;
@@ -144,6 +146,8 @@ class ThreadGroup
     size_t peak_threads_usage TSA_GUARDED_BY(mutex) = 0;
 
     UInt64 elapsed_total_threads_counter_ms TSA_GUARDED_BY(mutex) = 0;
+
+    static ThreadGroupPtr create(ContextPtr context);
 };
 
 /**

src/Interpreters/InsertDependenciesBuilder.cpp

@@ -964,7 +964,11 @@ bool InsertDependenciesBuilder::observePath(const DependencyPath & path)
         select_contexts[root_view] = select_context;
         insert_contexts[root_view] = insert_context;
         input_headers[root_view] = init_header;
-        thread_groups[root_view] = CurrentThread::getGroup();
+        /// For background tasks (i.e. Buffer flush) there may not be any group
+        if (auto thread_group = CurrentThread::getGroup())
+            thread_groups[root_view] = thread_group;
+        else
+            thread_groups[root_view] = ThreadGroup::createForMaterializedView(init_context);
         views_error_registry->init(root_view);
         dependent_views[root_view] = {};
     };
@@ -990,7 +994,7 @@ bool InsertDependenciesBuilder::observePath(const DependencyPath & path)
 
         inner_tables[current] = materialized_view->getTargetTableId();
         source_tables[current] = parent;
-        thread_groups[current] = ThreadGroup::createForMaterializedView();
+        thread_groups[current] = ThreadGroup::createForMaterializedView(init_context);
         view_types[current] = QueryViewsLogElement::ViewType::MATERIALIZED;
         views_error_registry->init(current);
 
@@ -1019,7 +1023,7 @@ bool InsertDependenciesBuilder::observePath(const DependencyPath & path)
         inner_tables[current] = current;
         select_queries[current] = live_view->getInnerQuery();
         input_headers[current] = output_headers.at(path.parent(2));
-        thread_groups[current] = ThreadGroup::createForMaterializedView();
+        thread_groups[current] = ThreadGroup::createForMaterializedView(init_context);
         view_types[current] = QueryViewsLogElement::ViewType::LIVE;
         views_error_registry->init(current);
 
@@ -1051,7 +1055,7 @@ bool InsertDependenciesBuilder::observePath(const DependencyPath & path)
         inner_tables[current] = current;
         select_queries[current] = window_view->getMergeableQuery();
         input_headers[current] = output_headers.at(path.parent(2));
-        thread_groups[current] = ThreadGroup::createForMaterializedView();
+        thread_groups[current] = ThreadGroup::createForMaterializedView(init_context);
         view_types[current] = QueryViewsLogElement::ViewType::WINDOW;
         views_error_registry->init(current);
 

src/Interpreters/ThreadStatusExt.cpp

@@ -145,36 +145,42 @@ ThreadGroupPtr ThreadGroup::createForQuery(ContextPtr query_context_, std::funct
     return group;
 }
 
-ThreadGroupPtr ThreadGroup::createForBackgroundProcess(ContextPtr storage_context)
+ThreadGroupPtr ThreadGroup::create(ContextPtr context)
 {
-    auto group = std::make_shared<ThreadGroup>(storage_context);
+    auto group = std::make_shared<ThreadGroup>(context);
 
-    group->memory_tracker.setDescription("Background process (mutate/merge)");
     /// However settings from storage context have to be applied
-    const Settings & settings = storage_context->getSettingsRef();
+    const Settings & settings = context->getSettingsRef();
     group->memory_tracker.setProfilerStep(settings[Setting::memory_profiler_step]);
     group->memory_tracker.setSampleProbability(settings[Setting::memory_profiler_sample_probability]);
     group->memory_tracker.setSampleMinAllocationSize(settings[Setting::memory_profiler_sample_min_allocation_size]);
     group->memory_tracker.setSampleMaxAllocationSize(settings[Setting::memory_profiler_sample_max_allocation_size]);
     group->memory_tracker.setSoftLimit(settings[Setting::memory_overcommit_ratio_denominator]);
-    group->memory_tracker.setParent(&background_memory_tracker);
     if (settings[Setting::memory_tracker_fault_probability] > 0.0)
         group->memory_tracker.setFaultProbability(settings[Setting::memory_tracker_fault_probability]);
 
     return group;
 }
 
-ThreadGroupPtr ThreadGroup::createForMaterializedView()
+ThreadGroupPtr ThreadGroup::createForMergeMutate(ContextPtr storage_context)
 {
-    auto current_group = CurrentThread::getGroup();
-    if (!current_group)
-        return nullptr;
-
-    auto group = std::make_shared<ThreadGroup>(current_group);
-    group->memory_tracker.setDescription("MaterializeView");
+    auto group = create(storage_context);
+    group->memory_tracker.setDescription("Background process (mutate/merge)");
+    group->memory_tracker.setParent(&background_memory_tracker);
     return group;
 }
 
+ThreadGroupPtr ThreadGroup::createForMaterializedView(ContextPtr context)
+{
+    ThreadGroupPtr res_group;
+    if (auto current_group = CurrentThread::getGroup())
+        res_group = std::make_shared<ThreadGroup>(current_group);
+    else
+        res_group = create(context);
+    res_group->memory_tracker.setDescription("MaterializeView");
+    return res_group;
+}
+
 void ThreadGroup::attachQueryForLog(const String & query_, UInt64 normalized_hash)
 {
     auto hash = normalized_hash ? normalized_hash : normalizedQueryHash(query_, false);

src/Storages/MergeTree/MergeList.cpp

@@ -74,7 +74,7 @@ MergeListElement::MergeListElement(const StorageID & table_id_, FutureMergedMuta
         throw Exception(ErrorCodes::LOGICAL_ERROR, "Got {} source parts for mutation {}: {}", future_part->parts.size(),
                         result_part_info.getPartNameV1(), fmt::join(source_part_names, ", "));
 
-    thread_group = ThreadGroup::createForBackgroundProcess(context);
+    thread_group = ThreadGroup::createForMergeMutate(context);
 }
 
 MergeInfo MergeListElement::getInfo() const

src/Storages/StorageBuffer.cpp

@@ -1089,18 +1089,13 @@ void StorageBuffer::writeBlockToDestination(const Block & block, StoragePtr tabl
 
 void StorageBuffer::backgroundFlush()
 {
+    try
     {
-        auto thread_group = ThreadGroup::createForBackgroundProcess(getContext());
-        ThreadGroupSwitcher group_switcher(thread_group, "BufferBgrFlush");
-
-        try
-        {
-            flushAllBuffers(true);
-        }
-        catch (...)
-        {
-            tryLogCurrentException(__PRETTY_FUNCTION__);
-        }
+        flushAllBuffers(true);
+    }
+    catch (...)
+    {
+        tryLogCurrentException(__PRETTY_FUNCTION__);
     }
 
     reschedule();