[Chore](debug) add rf merger info to PipelineFragmentContext::debug_string() (#56405)

add rf merger info to PipelineFragmentContext::debug_string()

Author: BiteTheDDDDt

be/src/pipeline/pipeline_fragment_context.cpp

@@ -1936,7 +1936,9 @@ std::string PipelineFragmentContext::debug_string() {
     for (size_t j = 0; j < _tasks.size(); j++) {
         fmt::format_to(debug_string_buffer, "Tasks in instance {}:\n", j);
         for (size_t i = 0; i < _tasks[j].size(); i++) {
-            fmt::format_to(debug_string_buffer, "Task {}: {}\n", i, _tasks[j][i]->debug_string());
+            fmt::format_to(debug_string_buffer, "Task {}: {}\n{}\n", i,
+                           _tasks[j][i]->debug_string(),
+                           _task_runtime_states[i][j]->local_runtime_filter_mgr()->debug_string());
         }
     }
 

be/src/runtime_filter/runtime_filter_mgr.cpp

@@ -283,6 +283,15 @@ Status RuntimeFilterMgr::sync_filter_size(const PSyncFilterSizeRequest* request)
     return Status::OK();
 }
 
+std::string RuntimeFilterMgr::debug_string() {
+    std::string result = "Merger Info:\n";
+    std::lock_guard l(_lock);
+    for (const auto& [filter_id, merger] : _local_merge_map) {
+        result += fmt::format("merger: {}\n", filter_id, merger.merger->debug_string());
+    }
+    return result;
+}
+
 // merge data
 Status RuntimeFilterMergeControllerEntity::merge(std::shared_ptr<QueryContext> query_ctx,
                                                  const PMergeFilterRequest* request,
@@ -398,4 +407,5 @@ Status RuntimeFilterMergeControllerEntity::merge(std::shared_ptr<QueryContext> q
     }
     return st;
 }
+
 } // namespace doris

be/src/runtime_filter/runtime_filter_mgr.h

@@ -100,6 +100,8 @@ class RuntimeFilterMgr {
     Status get_merge_addr(TNetworkAddress* addr);
     Status sync_filter_size(const PSyncFilterSizeRequest* request);
 
+    std::string debug_string();
+
 private:
     /**
      * `_is_global = true` means this runtime filter manager menages query-level runtime filters.