You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "zhiqiang-hhhh (via GitHub)" <gi...@apache.org> on 2024/03/29 06:07:38 UTC
[PR] [refactor](profile) WIP [doris]
zhiqiang-hhhh opened a new pull request, #33015:
URL: https://github.com/apache/doris/pull/33015
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](profile) WIP [doris]
Posted by "zhiqiang-hhhh (via GitHub)" <gi...@apache.org>.
zhiqiang-hhhh commented on PR #33015:
URL: https://github.com/apache/doris/pull/33015#issuecomment-2031918493
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](profile) WIP [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #33015:
URL: https://github.com/apache/doris/pull/33015#discussion_r1547791651
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -17,14 +17,307 @@
#include "runtime/runtime_query_statistics_mgr.h"
+#include <gen_cpp/FrontendService_types.h>
+#include <gen_cpp/RuntimeProfile_types.h>
+#include <gen_cpp/Status_types.h>
+#include <gen_cpp/Types_types.h>
+#include <thrift/TApplicationException.h>
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <random>
+#include <shared_mutex>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "common/logging.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "service/backend_options.h"
#include "util/debug_util.h"
#include "util/time.h"
+#include "util/uid_util.h"
#include "vec/core/block.h"
namespace doris {
+static Status _doReportExecStatsRpc(const TNetworkAddress& coor_addr,
+ const TReportExecStatusParams& req,
+ TReportExecStatusResult& res) {
+ Status client_status;
+ FrontendServiceConnection rpc_client(ExecEnv::GetInstance()->frontend_client_cache(), coor_addr,
+ &client_status);
+ if (!client_status.ok()) {
+ LOG_WARNING(
+ "could not get client rpc client of {} when reporting profiles, reason is {}, "
+ "not reporting, profile will be lost",
+ PrintThriftNetworkAddress(coor_addr), client_status.to_string());
+ return Status::RpcError("Client rpc client failed");
+ }
+
+ try {
+ try {
+ rpc_client->reportExecStatus(res, req);
+ } catch (const apache::thrift::transport::TTransportException& e) {
+ LOG_WARNING("Transport exception from {}, reason: {}, reopening",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ client_status = rpc_client.reopen(config::thrift_rpc_timeout_ms);
+ if (!client_status.ok()) {
+ LOG_WARNING("Reopen failed, reason: {}", client_status.to_string());
+ return Status::RpcError("Open rpc client failed");
+ }
+
+ rpc_client->reportExecStatus(res, req);
+ }
+ } catch (apache::thrift::TApplicationException& e) {
+ if (e.getType() == e.UNKNOWN_METHOD) {
+ LOG_WARNING(
+ "Failed to send statistics to {} due to {}, usually because the frontend "
+ "is not upgraded, check the version",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ } else {
+ LOG_WARNING(
+ "Failed to send statistics to {}, reason: {}, you can see fe log for "
+ "details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ }
+ return Status::RpcError("Send stats failed");
+ } catch (std::exception& e) {
+ LOG_WARNING("Failed to send statistics to {}, reason: {}, you can see fe log for details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ return Status::RpcError("Send stats failed");
+ }
+
+ return Status::OK();
+}
+
+TReportExecStatusParams RuntimeQueryStatiticsMgr::create_report_exec_status_params(
+ const TUniqueId& query_id, int32 fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile) {
+ TReportExecStatusParams req;
+ req.__set_query_id(query_id);
+ req.__set_backend_id(ExecEnv::GetInstance()->master_info()->backend_id);
+ req.__set_fragment_id(fragment_id);
+ std::vector<TDetailedReportParams> detailed_params;
+ for (const auto& p_profile : f_profile) {
+ TDetailedReportParams tmp;
+ tmp.__set_profile(*p_profile);
+ // tmp.fragment_instance_id is not needed for pipeline x
+ detailed_params.push_back(tmp);
+ }
+
+ req.__set_detailed_report(detailed_params);
+ return req;
+}
+
+void RuntimeQueryStatiticsMgr::start_report_thread() {
+ if (started.load()) {
+ DCHECK(false) << "report thread has been started";
+ LOG_ERROR("report thread has been started");
+ return;
+ }
+
+ started.store(true);
+
+ for (size_t i = 0; i < config::report_exec_status_thread_num; ++i) {
+ this->_report_profile_threads.emplace_back(std::make_unique<std::thread>(
+ &RuntimeQueryStatiticsMgr::report_query_profiles_thread, this));
+ }
+}
+
+void RuntimeQueryStatiticsMgr::report_query_profiles_thread() {
+ while (true) {
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+
+ while (_profile_map.empty() && _profile_map_x.empty() && !_report_profile_thread_stop) {
+ _report_profile_cv.wait(lock);
+ }
+
+ if (_report_profile_thread_stop) {
+ LOG_INFO("Report profile thread stopped");
+ return;
+ }
+ }
+
+ _report_query_profiles_function();
+ }
+}
+
+void RuntimeQueryStatiticsMgr::force_report_profile() {
Review Comment:
warning: method 'force_report_profile' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/runtime_query_statistics_mgr.h:90:
```diff
- void force_report_profile();
+ static void force_report_profile();
```
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -17,14 +17,307 @@
#include "runtime/runtime_query_statistics_mgr.h"
+#include <gen_cpp/FrontendService_types.h>
+#include <gen_cpp/RuntimeProfile_types.h>
+#include <gen_cpp/Status_types.h>
+#include <gen_cpp/Types_types.h>
+#include <thrift/TApplicationException.h>
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <random>
+#include <shared_mutex>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "common/logging.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "service/backend_options.h"
#include "util/debug_util.h"
#include "util/time.h"
+#include "util/uid_util.h"
#include "vec/core/block.h"
namespace doris {
+static Status _doReportExecStatsRpc(const TNetworkAddress& coor_addr,
+ const TReportExecStatusParams& req,
+ TReportExecStatusResult& res) {
+ Status client_status;
+ FrontendServiceConnection rpc_client(ExecEnv::GetInstance()->frontend_client_cache(), coor_addr,
+ &client_status);
+ if (!client_status.ok()) {
+ LOG_WARNING(
+ "could not get client rpc client of {} when reporting profiles, reason is {}, "
+ "not reporting, profile will be lost",
+ PrintThriftNetworkAddress(coor_addr), client_status.to_string());
+ return Status::RpcError("Client rpc client failed");
+ }
+
+ try {
+ try {
+ rpc_client->reportExecStatus(res, req);
+ } catch (const apache::thrift::transport::TTransportException& e) {
+ LOG_WARNING("Transport exception from {}, reason: {}, reopening",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ client_status = rpc_client.reopen(config::thrift_rpc_timeout_ms);
+ if (!client_status.ok()) {
+ LOG_WARNING("Reopen failed, reason: {}", client_status.to_string());
+ return Status::RpcError("Open rpc client failed");
+ }
+
+ rpc_client->reportExecStatus(res, req);
+ }
+ } catch (apache::thrift::TApplicationException& e) {
+ if (e.getType() == e.UNKNOWN_METHOD) {
+ LOG_WARNING(
+ "Failed to send statistics to {} due to {}, usually because the frontend "
+ "is not upgraded, check the version",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ } else {
+ LOG_WARNING(
+ "Failed to send statistics to {}, reason: {}, you can see fe log for "
+ "details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ }
+ return Status::RpcError("Send stats failed");
+ } catch (std::exception& e) {
+ LOG_WARNING("Failed to send statistics to {}, reason: {}, you can see fe log for details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ return Status::RpcError("Send stats failed");
+ }
+
+ return Status::OK();
+}
+
+TReportExecStatusParams RuntimeQueryStatiticsMgr::create_report_exec_status_params(
+ const TUniqueId& query_id, int32 fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile) {
+ TReportExecStatusParams req;
+ req.__set_query_id(query_id);
+ req.__set_backend_id(ExecEnv::GetInstance()->master_info()->backend_id);
+ req.__set_fragment_id(fragment_id);
+ std::vector<TDetailedReportParams> detailed_params;
+ for (const auto& p_profile : f_profile) {
+ TDetailedReportParams tmp;
+ tmp.__set_profile(*p_profile);
+ // tmp.fragment_instance_id is not needed for pipeline x
+ detailed_params.push_back(tmp);
+ }
+
+ req.__set_detailed_report(detailed_params);
+ return req;
+}
+
+void RuntimeQueryStatiticsMgr::start_report_thread() {
Review Comment:
warning: method 'start_report_thread' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/runtime_query_statistics_mgr.h:88:
```diff
- void start_report_thread();
+ static void start_report_thread();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](profile) WIP [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #33015:
URL: https://github.com/apache/doris/pull/33015#discussion_r1544120481
##########
be/src/pipeline/pipeline_fragment_context.cpp:
##########
@@ -981,4 +992,50 @@ std::string PipelineFragmentContext::debug_string() {
return fmt::to_string(debug_string_buffer);
}
+std::vector<profile::TRuntimeProfilePtr> PipelineFragmentContext::collect_profile_x() const {
+ std::vector<profile::TRuntimeProfilePtr> res;
+ DCHECK(_query_ctx->enable_pipeline_x_exec() == true)
+ << fmt::format("Query {} calling a pipeline X function, but its pipeline X is disabled",
+ print_id(this->_query_id));
+
+ std::stringstream ss;
+
+ for (auto& pipeline_profile : _runtime_state->pipeline_id_to_profile()) {
+ auto profile_ptr = std::make_shared<TRuntimeProfileTree>();
+ pipeline_profile->to_thrift(&(*profile_ptr));
+ res.push_back(profile_ptr);
+
+ std::vector<RuntimeProfile*> task_x_profile;
+ pipeline_profile->get_all_children(&task_x_profile);
+ for (RuntimeProfile* p : task_x_profile) {
+ if (p->name().find("PipelineXTask") != std::string::npos) {
+ ss << p->name() << '\n';
+ } else {
+ ss << '\t' << p->name() << '\n';
+ }
+ }
+ }
+
+ LOG_INFO("Query X {} fragment {} profile\n{} ", print_id(this->_query_id), this->_fragment_id,
+ ss.str());
+ return res;
+}
+
+profile::TRuntimeProfilePtr PipelineFragmentContext::collect_profile() const {
+ profile::TRuntimeProfilePtr res = std::make_shared<TRuntimeProfileTree>();
Review Comment:
warning: method 'collect_profile' can be made static [readability-convert-member-functions-to-static]
be/src/pipeline/pipeline_fragment_context.h:150:
```diff
- profile::TRuntimeProfilePtr collect_profile() const;
+ static profile::TRuntimeProfilePtr collect_profile() ;
```
```suggestion
t {
```
##########
be/src/runtime/fragment_mgr.cpp:
##########
@@ -193,7 +193,7 @@ Status FragmentMgr::trigger_pipeline_context_report(
// it is only invoked from the executor's reporting thread.
// Also, the reported status will always reflect the most recent execution status,
// including the final status when execution finishes.
-void FragmentMgr::coordinator_callback(const ReportStatusRequest& req) {
+void FragmentMgr::report_status_callback(const ReportStatusRequest& req) {
Review Comment:
warning: function 'report_status_callback' exceeds recommended size/complexity thresholds [readability-function-size]
```cpp
void FragmentMgr::report_status_callback(const ReportStatusRequest& req) {
^
```
<details>
<summary>Additional context</summary>
**be/src/runtime/fragment_mgr.cpp:195:** 264 lines including whitespace and comments (threshold 80)
```cpp
void FragmentMgr::report_status_callback(const ReportStatusRequest& req) {
^
```
</details>
##########
be/src/runtime/fragment_mgr.cpp:
##########
@@ -1546,4 +1577,17 @@
}
}
+void FragmentMgr::async_get_query_statics(const TUniqueId& query_id) {
+ std::lock_guard<std::mutex> lock(_lock);
+
+ auto iter = _query_ctx_map.find(query_id);
+
+ if (iter == _query_ctx_map.end()) {
+ throw doris::Exception(ErrorCode::NOT_FOUND, "query id not found");
+ }
+
+ iter->second->async_report_profile();
+ return;
+}
Review Comment:
warning: redundant return statement at the end of a function with a void return type [readability-redundant-control-flow]
```suggestion
}
```
##########
be/src/runtime/fragment_mgr.cpp:
##########
@@ -1546,4 +1577,17 @@
}
}
+void FragmentMgr::async_get_query_statics(const TUniqueId& query_id) {
Review Comment:
warning: method 'async_get_query_statics' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void FragmentMgr::async_get_query_statics(const TUniqueId& query_id) {
```
##########
be/src/pipeline/pipeline_fragment_context.h:
##########
@@ -17,21 +17,25 @@
#pragma once
+#include <gen_cpp/RuntimeProfile_types.h>
Review Comment:
warning: 'gen_cpp/RuntimeProfile_types.h' file not found [clang-diagnostic-error]
```cpp
#include <gen_cpp/RuntimeProfile_types.h>
^
```
##########
be/src/runtime/profile/profile.h:
##########
@@ -0,0 +1,58 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <gen_cpp/RuntimeProfile_types.h>
Review Comment:
warning: 'gen_cpp/RuntimeProfile_types.h' file not found [clang-diagnostic-error]
```cpp
#include <gen_cpp/RuntimeProfile_types.h>
^
```
##########
be/src/runtime/query_context.h:
##########
@@ -18,15 +18,18 @@
#pragma once
#include <gen_cpp/PaloInternalService_types.h>
Review Comment:
warning: 'gen_cpp/PaloInternalService_types.h' file not found [clang-diagnostic-error]
```cpp
#include <gen_cpp/PaloInternalService_types.h>
^
```
##########
be/src/runtime/query_context.cpp:
##########
@@ -281,4 +293,77 @@ Status QueryContext::set_workload_group(WorkloadGroupPtr& tg) {
return Status::OK();
}
+void QueryContext::async_report_profile_x() {
Review Comment:
warning: method 'async_report_profile_x' can be made const [readability-make-member-function-const]
```suggestion
void QueryContext::async_report_profile_x() const {
```
be/src/runtime/query_context.h:337:
```diff
- void async_report_profile_x();
+ void async_report_profile_x() const;
```
##########
be/src/service/backend_service.cpp:
##########
@@ -1125,4 +1128,39 @@ void BaseBackendService::warm_up_tablets(TWarmUpTabletsResponse& response,
response.__set_status(Status::NotSupported("warm_up_tablets is not implemented").to_thrift());
}
+void BaseBackendService::async_get_query_statics(TAsyncGetQueryStaticsResponse& _return,
Review Comment:
warning: method 'async_get_query_statics' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void BaseBackendService::async_get_query_statics(TAsyncGetQueryStaticsResponse& _return,
```
##########
be/src/runtime/runtime_query_statistics_mgr.h:
##########
@@ -18,10 +18,15 @@
#pragma once
#include <gen_cpp/Data_types.h>
Review Comment:
warning: 'gen_cpp/Data_types.h' file not found [clang-diagnostic-error]
```cpp
#include <gen_cpp/Data_types.h>
^
```
##########
be/src/service/backend_service.cpp:
##########
@@ -1125,4 +1128,39 @@
response.__set_status(Status::NotSupported("warm_up_tablets is not implemented").to_thrift());
}
+void BaseBackendService::async_get_query_statics(TAsyncGetQueryStaticsResponse& _return,
+ const TAsyncGetQueryStaticsRequest& request) {
+ _return = TAsyncGetQueryStaticsResponse();
+
+ if (!request.__isset.query_id) {
+ LOG_WARNING("Query_id is empty");
+ _return.__set_status(Status::InvalidArgument("query_id is empty").to_thrift());
+ return;
+ }
+
+ auto fragment_mgr = ExecEnv::GetInstance()->fragment_mgr();
+ if (fragment_mgr == nullptr) {
+ LOG_ERROR("Fragment manager is has not been created");
+ _return.__set_status(
+ Status::InternalError("Fragment manager has not been created").to_thrift());
+ return;
+ }
+
+ try {
+ fragment_mgr->async_get_query_statics(request.query_id);
+
+ ExecEnv::GetInstance()->runtime_query_statistics_mgr()->force_report_profile();
+
+ LOG_INFO("Async get {} query statics finished", print_id(request.query_id));
+ } catch (const doris::Exception& e) {
+ LOG_WARNING("Failed to async get {} query statics. error: {}", e.what());
+ _return.__set_status(
+ Status::NotFound("Query {} not found", print_id(request.query_id)).to_thrift());
+ return;
+ }
+
+ _return.__set_status(Status::OK().to_thrift());
+ return;
+}
Review Comment:
warning: redundant return statement at the end of a function with a void return type [readability-redundant-control-flow]
```suggestion
}
```
##########
be/src/runtime/query_context.cpp:
##########
@@ -281,4 +293,77 @@
return Status::OK();
}
+void QueryContext::async_report_profile_x() {
+ if (!enable_pipeline_x_exec()) {
+ return;
+ }
+
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ LOG_INFO(
+ "Pipeline x query context, register query profile, query {}, fragment profile count {}",
+ print_id(_query_id), _profile_map_x.size());
+
+ for (auto& [fid, f_profile] : _profile_map_x) {
+ auto tmp_f_profile = std::make_shared<profile::FragmentProfileX>();
+
+ for (auto p_profile : f_profile.second) {
+ tmp_f_profile->pipeline_profiles.push_back(
+ std::make_shared<profile::PipelineProfileX>(fid, f_profile.first, p_profile));
+ }
+
+ ExecEnv::GetInstance()->runtime_query_statistics_mgr()->register_fragment_profile_x(
+ _query_id, fid, this->coord_addr, tmp_f_profile);
+ }
+
+ _profile_map_x.clear();
+}
+
+void QueryContext::add_pipeline_profile_x(int f_id, bool finished,
+ profile::TRuntimeProfilePtr profile) {
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ LOG_INFO("Query X {} add pipeline profile, fid {}", print_id(this->_query_id), f_id);
+ _profile_map_x[f_id].first = finished;
+ _profile_map_x[f_id].second.push_back(profile);
+}
+
+void QueryContext::add_fragment_profile_x(
+ int f_id, bool finished, const std::vector<profile::TRuntimeProfilePtr>& pipeline_profile) {
+ LOG_INFO("Query X add fragment profile, query {}, fragment {}, pipeline profile count {} ",
+ print_id(this->_query_id), f_id, pipeline_profile.size());
+
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ _profile_map_x[f_id] = std::make_pair(finished, pipeline_profile);
+}
+
+void QueryContext::add_instance_profile(const TUniqueId& iid, bool finished,
+ profile::TRuntimeProfilePtr profile) {
+ // LOG_INFO("Query {} add instance profile, iid {}, finished {}", print_id(this->_query_id),
+ // print_id(iid), finished);
+ DCHECK(profile != nullptr) << print_id(iid);
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ _profile_map[print_id(iid)] =
+ std::make_pair(finished, std::make_shared<profile::InstanceProfile>(iid, profile));
+}
+
+void QueryContext::async_report_profile() {
Review Comment:
warning: method 'async_report_profile' can be made const [readability-make-member-function-const]
```suggestion
void QueryContext::async_report_profile() const {
```
be/src/runtime/query_context.h:344:
```diff
- void async_report_profile();
+ void async_report_profile() const;
```
##########
be/src/service/backend_service.cpp:
##########
@@ -1125,4 +1128,39 @@
response.__set_status(Status::NotSupported("warm_up_tablets is not implemented").to_thrift());
}
+void BaseBackendService::async_get_query_statics(TAsyncGetQueryStaticsResponse& _return,
+ const TAsyncGetQueryStaticsRequest& request) {
+ _return = TAsyncGetQueryStaticsResponse();
+
+ if (!request.__isset.query_id) {
+ LOG_WARNING("Query_id is empty");
+ _return.__set_status(Status::InvalidArgument("query_id is empty").to_thrift());
+ return;
+ }
+
+ auto fragment_mgr = ExecEnv::GetInstance()->fragment_mgr();
Review Comment:
warning: 'auto fragment_mgr' can be declared as 'auto *fragment_mgr' [readability-qualified-auto]
```suggestion
auto *fragment_mgr = ExecEnv::GetInstance()->fragment_mgr();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](profile) WIP [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #33015:
URL: https://github.com/apache/doris/pull/33015#discussion_r1547111706
##########
be/src/runtime/query_context.cpp:
##########
@@ -293,4 +310,113 @@
return Status::OK();
}
+void QueryContext::add_pipeline_profile_x(int f_id, std::shared_ptr<TRuntimeProfileTree> profile) {
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ LOG_INFO("Query X {} add pipeline profile, fid {}", print_id(this->_query_id), f_id);
+ DCHECK(_profile_map_x != nullptr && profile != nullptr) << fmt::format(
+ "Add pipeline profile failed, query {}, fragment {}", print_id(this->_query_id), f_id);
+
+ if (!_profile_map_x->contains(f_id)) {
+ LOG_WARNING("Query X {} add pipeline profile, fid {} not found", print_id(this->_query_id),
+ f_id);
+ return;
+ }
+
+ (*_profile_map_x)[f_id].push_back(profile);
+}
+
+void QueryContext::add_fragment_profile_x(
+ int f_id, const std::vector<std::shared_ptr<TRuntimeProfileTree>>& pipeline_profiles) {
+ LOG_INFO("Query X add fragment profile, query {}, fragment {}, pipeline profile count {} ",
+ print_id(this->_query_id), f_id, pipeline_profiles.size());
+
+ std::lock_guard<std::mutex> l(_profile_mutex);
+
+ DCHECK(_profile_map_x != nullptr) << fmt::format(
+ "Add pipeline profile failed, query {}, fragment {}", print_id(this->_query_id), f_id);
+#ifndef NDEBUG
+ for (const auto& p : pipeline_profiles) {
+ DCHECK(p != nullptr) << fmt::format("Add pipeline profile failed, query {}, fragment {}",
+ print_id(this->_query_id), f_id);
+ }
+#endif
+ _profile_map_x->insert(std::make_pair(f_id, pipeline_profiles));
+}
+
+void QueryContext::add_instance_profile(const TUniqueId& i_id,
+ std::shared_ptr<TRuntimeProfileTree> profile) {
+ DCHECK(profile != nullptr) << print_id(i_id);
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ DCHECK(_profile_map != nullptr)
+ << fmt::format("Add pipeline profile failed, query {}, instance {}",
+ print_id(this->_query_id), print_id(i_id));
+
+ _profile_map->insert(std::make_pair(i_id, profile));
+}
+
+void QueryContext::async_report_profile() {
+ _async_report_profile_x();
+ _async_report_profile_p();
+}
+
+void QueryContext::_async_report_profile_p() {
Review Comment:
warning: method '_async_report_profile_p' can be made const [readability-make-member-function-const]
```suggestion
void QueryContext::_async_report_profile_p() const {
```
be/src/runtime/query_context.h:367:
```diff
- void _async_report_profile_p();
+ void _async_report_profile_p() const;
```
##########
be/src/runtime/query_context.cpp:
##########
@@ -293,4 +310,113 @@
return Status::OK();
}
+void QueryContext::add_pipeline_profile_x(int f_id, std::shared_ptr<TRuntimeProfileTree> profile) {
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ LOG_INFO("Query X {} add pipeline profile, fid {}", print_id(this->_query_id), f_id);
+ DCHECK(_profile_map_x != nullptr && profile != nullptr) << fmt::format(
+ "Add pipeline profile failed, query {}, fragment {}", print_id(this->_query_id), f_id);
+
+ if (!_profile_map_x->contains(f_id)) {
+ LOG_WARNING("Query X {} add pipeline profile, fid {} not found", print_id(this->_query_id),
+ f_id);
+ return;
+ }
+
+ (*_profile_map_x)[f_id].push_back(profile);
+}
+
+void QueryContext::add_fragment_profile_x(
+ int f_id, const std::vector<std::shared_ptr<TRuntimeProfileTree>>& pipeline_profiles) {
+ LOG_INFO("Query X add fragment profile, query {}, fragment {}, pipeline profile count {} ",
+ print_id(this->_query_id), f_id, pipeline_profiles.size());
+
+ std::lock_guard<std::mutex> l(_profile_mutex);
+
+ DCHECK(_profile_map_x != nullptr) << fmt::format(
+ "Add pipeline profile failed, query {}, fragment {}", print_id(this->_query_id), f_id);
+#ifndef NDEBUG
+ for (const auto& p : pipeline_profiles) {
+ DCHECK(p != nullptr) << fmt::format("Add pipeline profile failed, query {}, fragment {}",
+ print_id(this->_query_id), f_id);
+ }
+#endif
+ _profile_map_x->insert(std::make_pair(f_id, pipeline_profiles));
+}
+
+void QueryContext::add_instance_profile(const TUniqueId& i_id,
+ std::shared_ptr<TRuntimeProfileTree> profile) {
+ DCHECK(profile != nullptr) << print_id(i_id);
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ DCHECK(_profile_map != nullptr)
+ << fmt::format("Add pipeline profile failed, query {}, instance {}",
+ print_id(this->_query_id), print_id(i_id));
+
+ _profile_map->insert(std::make_pair(i_id, profile));
+}
+
+void QueryContext::async_report_profile() {
+ _async_report_profile_x();
+ _async_report_profile_p();
+}
+
+void QueryContext::_async_report_profile_p() {
+ if (enable_pipeline_x_exec()) {
+ return;
+ }
+
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ LOG_INFO("Query {}, register query profile, instance profile count {}", print_id(_query_id),
+ _profile_map->size());
+
+ for (auto& [i_id, i_profile] : *_profile_map) {
+ ExecEnv::GetInstance()->runtime_query_statistics_mgr()->register_instance_profile(
+ _query_id, this->coord_addr, i_id, i_profile);
+ }
+
+ _profile_map->clear();
+}
+
+void QueryContext::_async_report_profile_x() {
Review Comment:
warning: method '_async_report_profile_x' can be made const [readability-make-member-function-const]
```suggestion
void QueryContext::_async_report_profile_x() const {
```
be/src/runtime/query_context.h:368:
```diff
- void _async_report_profile_x();
+ void _async_report_profile_x() const;
```
##########
be/src/runtime/runtime_query_statistics_mgr.h:
##########
@@ -75,9 +82,65 @@ class RuntimeQueryStatiticsMgr {
// used for backend_active_tasks
void get_active_be_tasks_block(vectorized::Block* block);
+ void start_report_thread();
+ void report_query_profiles_thread();
+ void force_report_profile();
+ void stop_report_thread();
+
+ void submit_report_status_task(const TUniqueId& q_id, const TNetworkAddress& coor_addr,
+ int32 f_id, const TUniqueId& i_id, int be_num, bool done,
+ Status exec_status);
+
+ void register_instance_profile(const TUniqueId& query_id, const TNetworkAddress& coor_addr,
+ const TUniqueId& instance_id,
+ std::shared_ptr<TRuntimeProfileTree> i_profile);
+
+ void register_fragment_profile_x(const TUniqueId& query_id, const TNetworkAddress& const_addr,
+ int32_t fragment_id,
+ std::vector<std::shared_ptr<TRuntimeProfileTree>> p_profiles);
+ using CoorAddrWithFragmentProfileMap =
+ std::tuple<TNetworkAddress,
+ std::unordered_map<int, std::vector<std::shared_ptr<TRuntimeProfileTree>>>>;
+
+ using CoorAddrWithInstanceProfileMap =
+ std::tuple<TNetworkAddress,
+ std::unordered_map<TUniqueId, std::shared_ptr<TRuntimeProfileTree>>>;
+
+ static TReportExecStatusParams create_report_exec_status_params(
+ const TUniqueId& q_id, int32 f_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile);
+
+ static void report_query_profile(
+ const TUniqueId&, const TNetworkAddress&,
Review Comment:
warning: parameter 3 is const-qualified in the function declaration; const-qualification of parameters only has an effect in function definitions [readability-avoid-const-params-in-decls]
```suggestion
std::unordered_map<int, std::vector<std::shared_ptr<TRuntimeProfileTree>>>&);
```
##########
be/src/runtime/query_context.cpp:
##########
@@ -293,4 +310,113 @@ Status QueryContext::set_workload_group(WorkloadGroupPtr& tg) {
return Status::OK();
}
+void QueryContext::add_pipeline_profile_x(int f_id, std::shared_ptr<TRuntimeProfileTree> profile) {
Review Comment:
warning: method 'add_pipeline_profile_x' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/query_context.h:373:
```diff
- void add_pipeline_profile_x(int f_id, std::shared_ptr<TRuntimeProfileTree> profile);
+ static void add_pipeline_profile_x(int f_id, std::shared_ptr<TRuntimeProfileTree> profile);
```
##########
be/src/pipeline/pipeline_fragment_context.cpp:
##########
@@ -993,4 +995,50 @@ std::string PipelineFragmentContext::debug_string() {
return fmt::to_string(debug_string_buffer);
}
+std::vector<std::shared_ptr<TRuntimeProfileTree>> PipelineFragmentContext::collect_profile_x()
+ const {
+ std::vector<std::shared_ptr<TRuntimeProfileTree>> res;
+ DCHECK(_query_ctx->enable_pipeline_x_exec() == true)
+ << fmt::format("Query {} calling a pipeline X function, but its pipeline X is disabled",
+ print_id(this->_query_id));
+ std::stringstream ss;
+
+ for (auto& pipeline_profile : _runtime_state->pipeline_id_to_profile()) {
+ auto profile_ptr = std::make_shared<TRuntimeProfileTree>();
+ pipeline_profile->to_thrift(&(*profile_ptr));
+ res.push_back(profile_ptr);
+
+ std::vector<RuntimeProfile*> task_x_profile;
+ pipeline_profile->get_all_children(&task_x_profile);
+ for (RuntimeProfile* p : task_x_profile) {
+ if (p->name().find("PipelineXTask") != std::string::npos) {
+ ss << p->name() << '\n';
+ } else {
+ ss << '\t' << p->name() << '\n';
+ }
+ }
+ }
+
+ LOG_INFO("Query X {} fragment {} profile\n{} ", print_id(this->_query_id), this->_fragment_id,
+ ss.str());
+ return res;
+}
+
+std::shared_ptr<TRuntimeProfileTree> PipelineFragmentContext::collect_profile_p() const {
+ auto res = std::make_shared<TRuntimeProfileTree>();
Review Comment:
warning: method 'collect_profile_p' can be made static [readability-convert-member-functions-to-static]
be/src/pipeline/pipeline_fragment_context.h:149:
```diff
- std::shared_ptr<TRuntimeProfileTree> collect_profile_p() const;
+ static std::shared_ptr<TRuntimeProfileTree> collect_profile_p() ;
```
```suggestion
t {
```
##########
be/src/service/backend_service.cpp:
##########
@@ -1129,4 +1133,51 @@
response.__set_status(Status::NotSupported("warm_up_tablets is not implemented").to_thrift());
}
+void BaseBackendService::get_realtime_query_exec_status(
+ TGetRealTimeQueryExecStatusResponse& _return,
+ const TGetRealTimeQueryExecStatusRequest& request) {
+ _return = TGetRealTimeQueryExecStatusResponse();
+
+ if (!request.__isset.query_id) {
+ LOG_WARNING("Query_id is empty");
+ _return.__set_status(Status::InvalidArgument("query_id is empty").to_thrift());
+ return;
+ }
+
Review Comment:
warning: 'auto fragment_mgr' can be declared as 'auto *fragment_mgr' [readability-qualified-auto]
```suggestion
auto *fragment_mgr = ExecEnv::GetInstance()->fragment_mgr();
```
##########
be/src/service/backend_service.cpp:
##########
@@ -1129,4 +1133,51 @@ void BaseBackendService::warm_up_tablets(TWarmUpTabletsResponse& response,
response.__set_status(Status::NotSupported("warm_up_tablets is not implemented").to_thrift());
}
+void BaseBackendService::get_realtime_query_exec_status(
Review Comment:
warning: method 'get_realtime_query_exec_status' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void BaseBackendService::get_realtime_query_exec_status(TGetRealTimeQueryExecStatusResponse& _return,
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](profile) WIP [doris]
Posted by "zhiqiang-hhhh (via GitHub)" <gi...@apache.org>.
zhiqiang-hhhh commented on PR #33015:
URL: https://github.com/apache/doris/pull/33015#issuecomment-2031909373
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](profile) WIP [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #33015:
URL: https://github.com/apache/doris/pull/33015#discussion_r1544141992
##########
be/src/runtime/query_context.cpp:
##########
@@ -281,4 +293,77 @@ Status QueryContext::set_workload_group(WorkloadGroupPtr& tg) {
return Status::OK();
}
+void QueryContext::async_report_profile_x() {
+ if (!enable_pipeline_x_exec()) {
+ return;
+ }
+
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ LOG_INFO(
+ "Pipeline x query context, register query profile, query {}, fragment profile count {}",
+ print_id(_query_id), _profile_map_x.size());
+
+ for (auto& [fid, f_profile] : _profile_map_x) {
+ auto tmp_f_profile = std::make_shared<profile::FragmentProfileX>();
+
+ for (auto p_profile : f_profile.second) {
+ tmp_f_profile->pipeline_profiles.push_back(
+ std::make_shared<profile::PipelineProfileX>(fid, f_profile.first, p_profile));
+ }
+
+ ExecEnv::GetInstance()->runtime_query_statistics_mgr()->register_fragment_profile_x(
+ _query_id, fid, this->coord_addr, tmp_f_profile);
+ }
+
+ _profile_map_x.clear();
+}
+
+void QueryContext::add_pipeline_profile_x(int f_id, bool finished,
+ profile::TRuntimeProfilePtr profile) {
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ LOG_INFO("Query X {} add pipeline profile, fid {}", print_id(this->_query_id), f_id);
+ _profile_map_x[f_id].first = finished;
+ _profile_map_x[f_id].second.push_back(profile);
+}
+
+void QueryContext::add_fragment_profile_x(
+ int f_id, bool finished, const std::vector<profile::TRuntimeProfilePtr>& pipeline_profile) {
+ LOG_INFO("Query X add fragment profile, query {}, fragment {}, pipeline profile count {} ",
+ print_id(this->_query_id), f_id, pipeline_profile.size());
+
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ _profile_map_x[f_id] = std::make_pair(finished, pipeline_profile);
+}
+
+void QueryContext::add_instance_profile(const TUniqueId& iid, bool finished,
+ profile::TRuntimeProfilePtr profile) {
+ // LOG_INFO("Query {} add instance profile, iid {}, finished {}", print_id(this->_query_id),
+ // print_id(iid), finished);
+ DCHECK(profile != nullptr) << print_id(iid);
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ _profile_map[print_id(iid)] =
+ std::make_pair(finished, std::make_shared<profile::InstanceProfile>(iid, profile));
+}
Review Comment:
PlanFragmentExecutor
PipelineXFragmentContext
Pipeline...
collect_profile ---> vector<fragmentid/ instanceid TRuntimeProfile>
QueryContext
std::vector<std::weak_ptr<PipelineXFragmentContext>>
std::vector<std::weak_ptr<PipelineXFragmentContext>>
std::vector<std::weak_ptr<PipelineXFragmentContext>>
map<> // 析构的时候，给我的
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](profile) WIP [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #33015:
URL: https://github.com/apache/doris/pull/33015#discussion_r1544457074
##########
be/src/runtime/query_context.cpp:
##########
@@ -281,4 +295,145 @@ Status QueryContext::set_workload_group(WorkloadGroupPtr& tg) {
return Status::OK();
}
+void QueryContext::add_pipeline_profile_x(int f_id, bool finished,
+ profile::TRuntimeProfilePtr profile) {
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ LOG_INFO("Query X {} add pipeline profile, fid {}", print_id(this->_query_id), f_id);
+ _profile_map_x[f_id].first = finished;
+ _profile_map_x[f_id].second.push_back(profile);
+}
+
+void QueryContext::add_fragment_profile_x(
+ int f_id, bool finished, const std::vector<profile::TRuntimeProfilePtr>& pipeline_profile) {
+ LOG_INFO("Query X add fragment profile, query {}, fragment {}, pipeline profile count {} ",
+ print_id(this->_query_id), f_id, pipeline_profile.size());
+
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ _profile_map_x[f_id] = std::make_pair(finished, pipeline_profile);
+}
+
+void QueryContext::add_instance_profile(const TUniqueId& iid, bool finished,
+ profile::TRuntimeProfilePtr profile) {
+ // LOG_INFO("Query {} add instance profile, iid {}, finished {}", print_id(this->_query_id),
+ // print_id(iid), finished);
+ DCHECK(profile != nullptr) << print_id(iid);
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ _profile_map[print_id(iid)] =
+ std::make_pair(finished, std::make_shared<profile::InstanceProfile>(iid, profile));
+}
+
+void QueryContext::async_report_profile() {
+ _async_report_profile_x();
+ _async_report_profile_p();
+}
+
+void QueryContext::_async_report_profile_p() {
Review Comment:
warning: method '_async_report_profile_p' can be made const [readability-make-member-function-const]
```suggestion
void QueryContext::_async_report_profile_p() const {
```
be/src/runtime/query_context.h:339:
```diff
- void _async_report_profile_p();
+ void _async_report_profile_p() const;
```
##########
be/src/runtime/query_context.cpp:
##########
@@ -281,4 +295,145 @@
return Status::OK();
}
+void QueryContext::add_pipeline_profile_x(int f_id, bool finished,
+ profile::TRuntimeProfilePtr profile) {
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ LOG_INFO("Query X {} add pipeline profile, fid {}", print_id(this->_query_id), f_id);
+ _profile_map_x[f_id].first = finished;
+ _profile_map_x[f_id].second.push_back(profile);
+}
+
+void QueryContext::add_fragment_profile_x(
+ int f_id, bool finished, const std::vector<profile::TRuntimeProfilePtr>& pipeline_profile) {
+ LOG_INFO("Query X add fragment profile, query {}, fragment {}, pipeline profile count {} ",
+ print_id(this->_query_id), f_id, pipeline_profile.size());
+
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ _profile_map_x[f_id] = std::make_pair(finished, pipeline_profile);
+}
+
+void QueryContext::add_instance_profile(const TUniqueId& iid, bool finished,
+ profile::TRuntimeProfilePtr profile) {
+ // LOG_INFO("Query {} add instance profile, iid {}, finished {}", print_id(this->_query_id),
+ // print_id(iid), finished);
+ DCHECK(profile != nullptr) << print_id(iid);
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ _profile_map[print_id(iid)] =
+ std::make_pair(finished, std::make_shared<profile::InstanceProfile>(iid, profile));
+}
+
+void QueryContext::async_report_profile() {
+ _async_report_profile_x();
+ _async_report_profile_p();
+}
+
+void QueryContext::_async_report_profile_p() {
+ if (enable_pipeline_x_exec()) {
+ return;
+ }
+
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ LOG_INFO("Query {}, register query profile, instance profile count {}", print_id(_query_id),
+ _profile_map.size());
+
+ for (auto& [i_id, i_profile] : _profile_map) {
+ ExecEnv::GetInstance()->runtime_query_statistics_mgr()->register_instance_profile(
+ _query_id, i_id, this->coord_addr, i_profile.second);
+ }
+
+ _profile_map.clear();
+}
+
+void QueryContext::_async_report_profile_x() {
Review Comment:
warning: method '_async_report_profile_x' can be made const [readability-make-member-function-const]
```suggestion
void QueryContext::_async_report_profile_x() const {
```
be/src/runtime/query_context.h:340:
```diff
- void _async_report_profile_x();
+ void _async_report_profile_x() const;
```
##########
be/src/runtime/fragment_mgr.cpp:
##########
@@ -1546,4 +1516,31 @@ void FragmentMgr::get_runtime_query_info(std::vector<WorkloadQueryInfo>* query_i
}
}
+profile::TRuntimeProfilePtr FragmentMgr::get_instance_profile(const TUniqueId& instance_id) {
Review Comment:
warning: method 'get_instance_profile' can be made static [readability-convert-member-functions-to-static]
```suggestion
static profile::TRuntimeProfilePtr FragmentMgr::get_instance_profile(const TUniqueId& instance_id) {
```
##########
be/src/pipeline/pipeline_fragment_context.cpp:
##########
@@ -981,4 +983,49 @@ std::string PipelineFragmentContext::debug_string() {
return fmt::to_string(debug_string_buffer);
}
+std::vector<profile::TRuntimeProfilePtr> PipelineFragmentContext::collect_profile_x() const {
+ std::vector<profile::TRuntimeProfilePtr> res;
+ DCHECK(_query_ctx->enable_pipeline_x_exec() == true)
+ << fmt::format("Query {} calling a pipeline X function, but its pipeline X is disabled",
+ print_id(this->_query_id));
+ std::stringstream ss;
+
+ for (auto& pipeline_profile : _runtime_state->pipeline_id_to_profile()) {
+ auto profile_ptr = std::make_shared<TRuntimeProfileTree>();
+ pipeline_profile->to_thrift(&(*profile_ptr));
+ res.push_back(profile_ptr);
+
+ std::vector<RuntimeProfile*> task_x_profile;
+ pipeline_profile->get_all_children(&task_x_profile);
+ for (RuntimeProfile* p : task_x_profile) {
+ if (p->name().find("PipelineXTask") != std::string::npos) {
+ ss << p->name() << '\n';
+ } else {
+ ss << '\t' << p->name() << '\n';
+ }
+ }
+ }
+
+ LOG_INFO("Query X {} fragment {} profile\n{} ", print_id(this->_query_id), this->_fragment_id,
+ ss.str());
+ return res;
+}
+
+profile::TRuntimeProfilePtr PipelineFragmentContext::collect_profile_p() const {
+ profile::TRuntimeProfilePtr res = std::make_shared<TRuntimeProfileTree>();
Review Comment:
warning: method 'collect_profile_p' can be made static [readability-convert-member-functions-to-static]
be/src/pipeline/pipeline_fragment_context.h:150:
```diff
- profile::TRuntimeProfilePtr collect_profile_p() const;
+ static profile::TRuntimeProfilePtr collect_profile_p() ;
```
```suggestion
t {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](profile) WIP [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #33015:
URL: https://github.com/apache/doris/pull/33015#issuecomment-2026723262
Thank you for your contribution to Apache Doris.
Don't know what should be done next? See [How to process your PR](https://cwiki.apache.org/confluence/display/DORIS/How+to+process+your+PR)
Since 2024-03-18, the Document has been moved to [doris-website](https://github.com/apache/doris-website).
See [Doris Document](https://cwiki.apache.org/confluence/display/DORIS/Doris+Document).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](profile) WIP [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #33015:
URL: https://github.com/apache/doris/pull/33015#discussion_r1544134997
##########
be/src/runtime/query_context.cpp:
##########
@@ -281,4 +293,77 @@ Status QueryContext::set_workload_group(WorkloadGroupPtr& tg) {
return Status::OK();
}
+void QueryContext::async_report_profile_x() {
+ if (!enable_pipeline_x_exec()) {
+ return;
+ }
+
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ LOG_INFO(
+ "Pipeline x query context, register query profile, query {}, fragment profile count {}",
+ print_id(_query_id), _profile_map_x.size());
+
+ for (auto& [fid, f_profile] : _profile_map_x) {
+ auto tmp_f_profile = std::make_shared<profile::FragmentProfileX>();
+
+ for (auto p_profile : f_profile.second) {
+ tmp_f_profile->pipeline_profiles.push_back(
+ std::make_shared<profile::PipelineProfileX>(fid, f_profile.first, p_profile));
+ }
+
+ ExecEnv::GetInstance()->runtime_query_statistics_mgr()->register_fragment_profile_x(
+ _query_id, fid, this->coord_addr, tmp_f_profile);
+ }
+
+ _profile_map_x.clear();
+}
+
+void QueryContext::add_pipeline_profile_x(int f_id, bool finished,
+ profile::TRuntimeProfilePtr profile) {
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ LOG_INFO("Query X {} add pipeline profile, fid {}", print_id(this->_query_id), f_id);
+ _profile_map_x[f_id].first = finished;
+ _profile_map_x[f_id].second.push_back(profile);
+}
+
+void QueryContext::add_fragment_profile_x(
+ int f_id, bool finished, const std::vector<profile::TRuntimeProfilePtr>& pipeline_profile) {
Review Comment:
不用定义profile 这个namespace
不要定义TRuntimeProfilePtr
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](profile) WIP [doris]
Posted by "zhiqiang-hhhh (via GitHub)" <gi...@apache.org>.
zhiqiang-hhhh closed pull request #33015: [refactor](profile) WIP
URL: https://github.com/apache/doris/pull/33015
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](profile) WIP [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #33015:
URL: https://github.com/apache/doris/pull/33015#discussion_r1549021906
##########
be/src/runtime/plan_fragment_executor.cpp:
##########
@@ -644,4 +646,19 @@ void PlanFragmentExecutor::close() {
_closed = true;
}
+std::shared_ptr<TRuntimeProfileTree> PlanFragmentExecutor::collect_realtime_query_profile() const {
Review Comment:
warning: method 'collect_realtime_query_profile' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/plan_fragment_executor.h:149:
```diff
- std::shared_ptr<TRuntimeProfileTree> collect_realtime_query_profile() const;
+ static std::shared_ptr<TRuntimeProfileTree> collect_realtime_query_profile() ;
```
```suggestion
std::shared_ptr<TRuntimeProfileTree> PlanFragmentExecutor::collect_realtime_query_profile() {
```
##########
be/src/runtime/query_context.cpp:
##########
@@ -293,4 +310,90 @@ Status QueryContext::set_workload_group(WorkloadGroupPtr& tg) {
return Status::OK();
}
+void QueryContext::add_fragment_profile_x(
Review Comment:
warning: method 'add_fragment_profile_x' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void QueryContext::add_fragment_profile_x(
```
##########
be/src/runtime/query_context.cpp:
##########
@@ -293,4 +310,90 @@
return Status::OK();
}
+void QueryContext::add_fragment_profile_x(
+ int fragment_id, const std::vector<std::shared_ptr<TRuntimeProfileTree>>& pipeline_profiles,
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile) {
+#ifndef NDEBUG
+ for (const auto& p : pipeline_profiles) {
+ DCHECK(p != nullptr) << fmt::format("Add pipeline profile failed, query {}, fragment {}",
+ print_id(this->_query_id), fragment_id);
+ }
+#endif
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ LOG_INFO("Query X add fragment profile, query {}, fragment {}, pipeline profile count {} ",
+ print_id(this->_query_id), fragment_id, pipeline_profiles.size());
+ _profile_map_x.insert(std::make_pair(fragment_id, pipeline_profiles));
+
+ if (load_channel_profile != nullptr) {
+ _load_channel_profile_map_x.insert(std::make_pair(fragment_id, load_channel_profile));
+ }
+}
+
+void QueryContext::add_instance_profile(const TUniqueId& instance_id,
Review Comment:
warning: method 'add_instance_profile' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void QueryContext::add_instance_profile(const TUniqueId& instance_id,
```
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -17,14 +17,342 @@
#include "runtime/runtime_query_statistics_mgr.h"
+#include <gen_cpp/FrontendService_types.h>
+#include <gen_cpp/RuntimeProfile_types.h>
+#include <gen_cpp/Status_types.h>
+#include <gen_cpp/Types_types.h>
+#include <thrift/TApplicationException.h>
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <random>
+#include <shared_mutex>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "common/logging.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "service/backend_options.h"
#include "util/debug_util.h"
#include "util/time.h"
+#include "util/uid_util.h"
#include "vec/core/block.h"
namespace doris {
+static Status _doReportExecStatsRpc(const TNetworkAddress& coor_addr,
+ const TReportExecStatusParams& req,
+ TReportExecStatusResult& res) {
+ Status client_status;
+ FrontendServiceConnection rpc_client(ExecEnv::GetInstance()->frontend_client_cache(), coor_addr,
+ &client_status);
+ if (!client_status.ok()) {
+ LOG_WARNING(
+ "could not get client rpc client of {} when reporting profiles, reason is {}, "
+ "not reporting, profile will be lost",
+ PrintThriftNetworkAddress(coor_addr), client_status.to_string());
+ return Status::RpcError("Client rpc client failed");
+ }
+
+ try {
+ try {
+ rpc_client->reportExecStatus(res, req);
+ } catch (const apache::thrift::transport::TTransportException& e) {
+ LOG_WARNING("Transport exception from {}, reason: {}, reopening",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ client_status = rpc_client.reopen(config::thrift_rpc_timeout_ms);
+ if (!client_status.ok()) {
+ LOG_WARNING("Reopen failed, reason: {}", client_status.to_string());
+ return Status::RpcError("Open rpc client failed");
+ }
+
+ rpc_client->reportExecStatus(res, req);
+ }
+ } catch (apache::thrift::TApplicationException& e) {
+ if (e.getType() == e.UNKNOWN_METHOD) {
+ LOG_WARNING(
+ "Failed to send statistics to {} due to {}, usually because the frontend "
+ "is not upgraded, check the version",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ } else {
+ LOG_WARNING(
+ "Failed to send statistics to {}, reason: {}, you can see fe log for "
+ "details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ }
+ return Status::RpcError("Send stats failed");
+ } catch (std::exception& e) {
+ LOG_WARNING("Failed to send statistics to {}, reason: {}, you can see fe log for details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ return Status::RpcError("Send stats failed");
+ }
+
+ return Status::OK();
+}
+
+TReportExecStatusParams RuntimeQueryStatiticsMgr::create_report_exec_status_params(
+ const TUniqueId& query_id, int32 fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile,
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile) {
+ TReportExecStatusParams req;
+ req.__set_query_id(query_id);
+ req.__set_backend_id(ExecEnv::GetInstance()->master_info()->backend_id);
+ req.__set_fragment_id(fragment_id);
+
+ std::vector<TDetailedReportParams> detailed_params;
+ for (const auto& p_profile : f_profile) {
+ TDetailedReportParams tmp;
+ tmp.__set_profile(*p_profile);
+ // tmp.fragment_instance_id is not needed for pipeline x
+ detailed_params.push_back(tmp);
+ }
+ req.__set_detailed_report(detailed_params);
+
+ if (load_channel_profile) {
+ req.__set_loadChannelProfile(*load_channel_profile);
+ }
+
+ return req;
+}
+
+void RuntimeQueryStatiticsMgr::start_report_thread() {
+ if (started.load()) {
+ DCHECK(false) << "report thread has been started";
+ LOG_ERROR("report thread has been started");
+ return;
+ }
+
+ started.store(true);
+
+ for (size_t i = 0; i < config::report_exec_status_thread_num; ++i) {
+ this->_report_profile_threads.emplace_back(std::make_unique<std::thread>(
+ &RuntimeQueryStatiticsMgr::report_query_profiles_thread, this));
+ }
+}
+
+void RuntimeQueryStatiticsMgr::report_query_profiles_thread() {
+ while (true) {
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+
+ while (_query_profile_map.empty() && _profile_map_x.empty() &&
+ !_report_profile_thread_stop) {
+ _report_profile_cv.wait(lock);
+ }
+
+ if (_report_profile_thread_stop) {
+ LOG_INFO("Report profile thread stopped");
+ return;
+ }
+ }
+
+ _report_query_profiles_function();
+ }
+}
+
+void RuntimeQueryStatiticsMgr::force_report_profile() {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+ _report_profile_cv.notify_one();
+}
+
+void RuntimeQueryStatiticsMgr::stop_report_thread() {
+ if (!started) {
+ return;
+ }
+
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+ _report_profile_thread_stop = true;
+ LOG_INFO("All report threads are going to stop");
+ _report_profile_cv.notify_all();
+ }
+
+ for (const auto& thread : _report_profile_threads) {
+ thread->join();
+ }
+
+ LOG_INFO("All report threads stopped");
+}
+
+void RuntimeQueryStatiticsMgr::register_instance_profile(
+ const TUniqueId& query_id, const TNetworkAddress& coor_addr, const TUniqueId& instance_id,
+ std::shared_ptr<TRuntimeProfileTree> instance_profile,
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile) {
+ if (instance_profile == nullptr) {
+ auto msg = fmt::format("Register instance profile {} {} failed, profile is null",
+ print_id(query_id), print_id(instance_id));
+ DCHECK(false) << msg;
+ LOG_ERROR(msg);
+ return;
+ }
+
+ std::lock_guard<std::shared_mutex> lg(_query_profile_map_lock);
+
+ if (!_query_profile_map.contains(query_id)) {
+ _query_profile_map[query_id] = std::make_tuple(
+ coor_addr, std::unordered_map<TUniqueId, std::shared_ptr<TRuntimeProfileTree>>());
+ }
+
+ std::get<1>(_query_profile_map[query_id]).insert(std::make_pair(instance_id, instance_profile));
+
+ if (load_channel_profile != nullptr) {
+ _load_channel_profile_map[instance_id] = load_channel_profile;
+ }
+
+ LOG_INFO("Register instance profile {} {}", print_id(query_id), print_id(instance_id));
+}
+
+void RuntimeQueryStatiticsMgr::register_fragment_profile_x(
+ const TUniqueId& query_id, const TNetworkAddress& coor_addr, int32_t fragment_id,
+ std::vector<std::shared_ptr<TRuntimeProfileTree>> p_profiles,
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile_x) {
+ for (const auto& p : p_profiles) {
+ if (p == nullptr) {
+ auto msg = fmt::format("Register fragment profile {} {} failed, profile is null",
+ print_id(query_id), fragment_id);
+ DCHECK(false) << msg;
+ LOG_ERROR(msg);
+ return;
+ }
+ }
+
+ std::lock_guard<std::shared_mutex> lg(_query_profile_map_lock);
+
+ if (!_profile_map_x.contains(query_id)) {
+ _profile_map_x[query_id] = std::make_tuple(
+ coor_addr,
+ std::unordered_map<int, std::vector<std::shared_ptr<TRuntimeProfileTree>>>());
+ }
+
+ std::unordered_map<int, std::vector<std::shared_ptr<TRuntimeProfileTree>>>&
+ fragment_profile_map = std::get<1>(_profile_map_x[query_id]);
+ fragment_profile_map.insert(std::make_pair(fragment_id, p_profiles));
+
+ if (load_channel_profile_x != nullptr) {
+ _load_channel_profile_map_x[std::make_pair(query_id, fragment_id)] = load_channel_profile_x;
+ }
+
+ LOG_INFO("register x profile done {}, fragment {}, profiles {}", print_id(query_id),
+ fragment_id, p_profiles.size());
+}
+
+void RuntimeQueryStatiticsMgr::_report_query_profiles_non_pipeline() {
Review Comment:
warning: method '_report_query_profiles_non_pipeline' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/runtime_query_statistics_mgr.h:124:
```diff
- void _report_query_profiles_non_pipeline();
+ static void _report_query_profiles_non_pipeline();
```
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -17,14 +17,342 @@
#include "runtime/runtime_query_statistics_mgr.h"
+#include <gen_cpp/FrontendService_types.h>
+#include <gen_cpp/RuntimeProfile_types.h>
+#include <gen_cpp/Status_types.h>
+#include <gen_cpp/Types_types.h>
+#include <thrift/TApplicationException.h>
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <random>
+#include <shared_mutex>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "common/logging.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "service/backend_options.h"
#include "util/debug_util.h"
#include "util/time.h"
+#include "util/uid_util.h"
#include "vec/core/block.h"
namespace doris {
+static Status _doReportExecStatsRpc(const TNetworkAddress& coor_addr,
+ const TReportExecStatusParams& req,
+ TReportExecStatusResult& res) {
+ Status client_status;
+ FrontendServiceConnection rpc_client(ExecEnv::GetInstance()->frontend_client_cache(), coor_addr,
+ &client_status);
+ if (!client_status.ok()) {
+ LOG_WARNING(
+ "could not get client rpc client of {} when reporting profiles, reason is {}, "
+ "not reporting, profile will be lost",
+ PrintThriftNetworkAddress(coor_addr), client_status.to_string());
+ return Status::RpcError("Client rpc client failed");
+ }
+
+ try {
+ try {
+ rpc_client->reportExecStatus(res, req);
+ } catch (const apache::thrift::transport::TTransportException& e) {
+ LOG_WARNING("Transport exception from {}, reason: {}, reopening",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ client_status = rpc_client.reopen(config::thrift_rpc_timeout_ms);
+ if (!client_status.ok()) {
+ LOG_WARNING("Reopen failed, reason: {}", client_status.to_string());
+ return Status::RpcError("Open rpc client failed");
+ }
+
+ rpc_client->reportExecStatus(res, req);
+ }
+ } catch (apache::thrift::TApplicationException& e) {
+ if (e.getType() == e.UNKNOWN_METHOD) {
+ LOG_WARNING(
+ "Failed to send statistics to {} due to {}, usually because the frontend "
+ "is not upgraded, check the version",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ } else {
+ LOG_WARNING(
+ "Failed to send statistics to {}, reason: {}, you can see fe log for "
+ "details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ }
+ return Status::RpcError("Send stats failed");
+ } catch (std::exception& e) {
+ LOG_WARNING("Failed to send statistics to {}, reason: {}, you can see fe log for details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ return Status::RpcError("Send stats failed");
+ }
+
+ return Status::OK();
+}
+
+TReportExecStatusParams RuntimeQueryStatiticsMgr::create_report_exec_status_params(
+ const TUniqueId& query_id, int32 fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile,
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile) {
+ TReportExecStatusParams req;
+ req.__set_query_id(query_id);
+ req.__set_backend_id(ExecEnv::GetInstance()->master_info()->backend_id);
+ req.__set_fragment_id(fragment_id);
+
+ std::vector<TDetailedReportParams> detailed_params;
+ for (const auto& p_profile : f_profile) {
+ TDetailedReportParams tmp;
+ tmp.__set_profile(*p_profile);
+ // tmp.fragment_instance_id is not needed for pipeline x
+ detailed_params.push_back(tmp);
+ }
+ req.__set_detailed_report(detailed_params);
+
+ if (load_channel_profile) {
+ req.__set_loadChannelProfile(*load_channel_profile);
+ }
+
+ return req;
+}
+
+void RuntimeQueryStatiticsMgr::start_report_thread() {
+ if (started.load()) {
+ DCHECK(false) << "report thread has been started";
+ LOG_ERROR("report thread has been started");
+ return;
+ }
+
+ started.store(true);
+
+ for (size_t i = 0; i < config::report_exec_status_thread_num; ++i) {
+ this->_report_profile_threads.emplace_back(std::make_unique<std::thread>(
+ &RuntimeQueryStatiticsMgr::report_query_profiles_thread, this));
+ }
+}
+
+void RuntimeQueryStatiticsMgr::report_query_profiles_thread() {
+ while (true) {
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+
+ while (_query_profile_map.empty() && _profile_map_x.empty() &&
+ !_report_profile_thread_stop) {
+ _report_profile_cv.wait(lock);
+ }
+
+ if (_report_profile_thread_stop) {
+ LOG_INFO("Report profile thread stopped");
+ return;
+ }
+ }
+
+ _report_query_profiles_function();
+ }
+}
+
+void RuntimeQueryStatiticsMgr::force_report_profile() {
Review Comment:
warning: method 'force_report_profile' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/runtime_query_statistics_mgr.h:91:
```diff
- void force_report_profile();
+ static void force_report_profile();
```
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -17,14 +17,342 @@
#include "runtime/runtime_query_statistics_mgr.h"
+#include <gen_cpp/FrontendService_types.h>
+#include <gen_cpp/RuntimeProfile_types.h>
+#include <gen_cpp/Status_types.h>
+#include <gen_cpp/Types_types.h>
+#include <thrift/TApplicationException.h>
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <random>
+#include <shared_mutex>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "common/logging.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "service/backend_options.h"
#include "util/debug_util.h"
#include "util/time.h"
+#include "util/uid_util.h"
#include "vec/core/block.h"
namespace doris {
+static Status _doReportExecStatsRpc(const TNetworkAddress& coor_addr,
+ const TReportExecStatusParams& req,
+ TReportExecStatusResult& res) {
+ Status client_status;
+ FrontendServiceConnection rpc_client(ExecEnv::GetInstance()->frontend_client_cache(), coor_addr,
+ &client_status);
+ if (!client_status.ok()) {
+ LOG_WARNING(
+ "could not get client rpc client of {} when reporting profiles, reason is {}, "
+ "not reporting, profile will be lost",
+ PrintThriftNetworkAddress(coor_addr), client_status.to_string());
+ return Status::RpcError("Client rpc client failed");
+ }
+
+ try {
+ try {
+ rpc_client->reportExecStatus(res, req);
+ } catch (const apache::thrift::transport::TTransportException& e) {
+ LOG_WARNING("Transport exception from {}, reason: {}, reopening",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ client_status = rpc_client.reopen(config::thrift_rpc_timeout_ms);
+ if (!client_status.ok()) {
+ LOG_WARNING("Reopen failed, reason: {}", client_status.to_string());
+ return Status::RpcError("Open rpc client failed");
+ }
+
+ rpc_client->reportExecStatus(res, req);
+ }
+ } catch (apache::thrift::TApplicationException& e) {
+ if (e.getType() == e.UNKNOWN_METHOD) {
+ LOG_WARNING(
+ "Failed to send statistics to {} due to {}, usually because the frontend "
+ "is not upgraded, check the version",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ } else {
+ LOG_WARNING(
+ "Failed to send statistics to {}, reason: {}, you can see fe log for "
+ "details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ }
+ return Status::RpcError("Send stats failed");
+ } catch (std::exception& e) {
+ LOG_WARNING("Failed to send statistics to {}, reason: {}, you can see fe log for details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ return Status::RpcError("Send stats failed");
+ }
+
+ return Status::OK();
+}
+
+TReportExecStatusParams RuntimeQueryStatiticsMgr::create_report_exec_status_params(
+ const TUniqueId& query_id, int32 fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile,
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile) {
+ TReportExecStatusParams req;
+ req.__set_query_id(query_id);
+ req.__set_backend_id(ExecEnv::GetInstance()->master_info()->backend_id);
+ req.__set_fragment_id(fragment_id);
+
+ std::vector<TDetailedReportParams> detailed_params;
+ for (const auto& p_profile : f_profile) {
+ TDetailedReportParams tmp;
+ tmp.__set_profile(*p_profile);
+ // tmp.fragment_instance_id is not needed for pipeline x
+ detailed_params.push_back(tmp);
+ }
+ req.__set_detailed_report(detailed_params);
+
+ if (load_channel_profile) {
+ req.__set_loadChannelProfile(*load_channel_profile);
+ }
+
+ return req;
+}
+
+void RuntimeQueryStatiticsMgr::start_report_thread() {
Review Comment:
warning: method 'start_report_thread' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/runtime_query_statistics_mgr.h:89:
```diff
- void start_report_thread();
+ static void start_report_thread();
```
##########
be/src/runtime/query_context.cpp:
##########
@@ -293,4 +310,90 @@
return Status::OK();
}
+void QueryContext::add_fragment_profile_x(
+ int fragment_id, const std::vector<std::shared_ptr<TRuntimeProfileTree>>& pipeline_profiles,
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile) {
+#ifndef NDEBUG
+ for (const auto& p : pipeline_profiles) {
+ DCHECK(p != nullptr) << fmt::format("Add pipeline profile failed, query {}, fragment {}",
+ print_id(this->_query_id), fragment_id);
+ }
+#endif
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ LOG_INFO("Query X add fragment profile, query {}, fragment {}, pipeline profile count {} ",
+ print_id(this->_query_id), fragment_id, pipeline_profiles.size());
+ _profile_map_x.insert(std::make_pair(fragment_id, pipeline_profiles));
+
+ if (load_channel_profile != nullptr) {
+ _load_channel_profile_map_x.insert(std::make_pair(fragment_id, load_channel_profile));
+ }
+}
+
+void QueryContext::add_instance_profile(const TUniqueId& instance_id,
+ std::shared_ptr<TRuntimeProfileTree> profile,
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile) {
+ DCHECK(profile != nullptr) << print_id(instance_id);
+
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ _profile_map.insert(std::make_pair(instance_id, profile));
+ if (load_channel_profile != nullptr) {
+ _load_channel_profile_map.insert(std::make_pair(instance_id, load_channel_profile));
+ }
+}
+
+void QueryContext::_async_report_profile() {
+ if (!enable_profile()) {
+ return;
+ }
+
+ _async_report_profile_x();
+ _async_report_profile_non_pipeline();
+}
+
+void QueryContext::_async_report_profile_non_pipeline() {
+ if (enable_pipeline_exec()) {
+ return;
+ }
+
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ LOG_INFO("Query {}, register query profile, instance profile count {}", print_id(_query_id),
+ _profile_map.size());
+
+ for (auto& [instance_id, instance_profile] : _profile_map) {
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile = nullptr;
+ if (_load_channel_profile_map.contains(instance_id)) {
+ load_channel_profile = _load_channel_profile_map[instance_id];
+ }
+
+ ExecEnv::GetInstance()->runtime_query_statistics_mgr()->register_instance_profile(
+ _query_id, this->coord_addr, instance_id, instance_profile, load_channel_profile);
+ }
+
+ ExecEnv::GetInstance()->runtime_query_statistics_mgr()->force_report_profile();
+}
+
+void QueryContext::_async_report_profile_x() {
Review Comment:
warning: method '_async_report_profile_x' can be made const [readability-make-member-function-const]
```suggestion
void QueryContext::_async_report_profile_x() const {
```
be/src/runtime/query_context.h:385:
```diff
- void _async_report_profile_x();
+ void _async_report_profile_x() const;
```
##########
be/src/runtime/plan_fragment_executor.cpp:
##########
@@ -644,4 +646,19 @@
_closed = true;
}
+std::shared_ptr<TRuntimeProfileTree> PlanFragmentExecutor::collect_realtime_query_profile() const {
+ std::shared_ptr<TRuntimeProfileTree> res = std::make_shared<TRuntimeProfileTree>();
+ _runtime_state->runtime_profile()->compute_time_in_profile();
+ _runtime_state->runtime_profile()->to_thrift(res.get());
+ return res;
+}
+
+std::shared_ptr<TRuntimeProfileTree> PlanFragmentExecutor::collect_realtime_load_channel_profile()
Review Comment:
warning: method 'collect_realtime_load_channel_profile' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/plan_fragment_executor.h:150:
```diff
- std::shared_ptr<TRuntimeProfileTree> collect_realtime_load_channel_profile() const;
+ static std::shared_ptr<TRuntimeProfileTree> collect_realtime_load_channel_profile() ;
```
be/src/runtime/plan_fragment_executor.cpp:656:
```diff
- const {
+ {
```
##########
be/src/runtime/query_context.cpp:
##########
@@ -293,4 +310,90 @@
return Status::OK();
}
+void QueryContext::add_fragment_profile_x(
+ int fragment_id, const std::vector<std::shared_ptr<TRuntimeProfileTree>>& pipeline_profiles,
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile) {
+#ifndef NDEBUG
+ for (const auto& p : pipeline_profiles) {
+ DCHECK(p != nullptr) << fmt::format("Add pipeline profile failed, query {}, fragment {}",
+ print_id(this->_query_id), fragment_id);
+ }
+#endif
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ LOG_INFO("Query X add fragment profile, query {}, fragment {}, pipeline profile count {} ",
+ print_id(this->_query_id), fragment_id, pipeline_profiles.size());
+ _profile_map_x.insert(std::make_pair(fragment_id, pipeline_profiles));
+
+ if (load_channel_profile != nullptr) {
+ _load_channel_profile_map_x.insert(std::make_pair(fragment_id, load_channel_profile));
+ }
+}
+
+void QueryContext::add_instance_profile(const TUniqueId& instance_id,
+ std::shared_ptr<TRuntimeProfileTree> profile,
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile) {
+ DCHECK(profile != nullptr) << print_id(instance_id);
+
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ _profile_map.insert(std::make_pair(instance_id, profile));
+ if (load_channel_profile != nullptr) {
+ _load_channel_profile_map.insert(std::make_pair(instance_id, load_channel_profile));
+ }
+}
+
+void QueryContext::_async_report_profile() {
+ if (!enable_profile()) {
+ return;
+ }
+
+ _async_report_profile_x();
+ _async_report_profile_non_pipeline();
+}
+
+void QueryContext::_async_report_profile_non_pipeline() {
Review Comment:
warning: method '_async_report_profile_non_pipeline' can be made const [readability-make-member-function-const]
```suggestion
void QueryContext::_async_report_profile_non_pipeline() const {
```
be/src/runtime/query_context.h:384:
```diff
- void _async_report_profile_non_pipeline();
+ void _async_report_profile_non_pipeline() const;
```
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -17,14 +17,342 @@
#include "runtime/runtime_query_statistics_mgr.h"
+#include <gen_cpp/FrontendService_types.h>
+#include <gen_cpp/RuntimeProfile_types.h>
+#include <gen_cpp/Status_types.h>
+#include <gen_cpp/Types_types.h>
+#include <thrift/TApplicationException.h>
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <random>
+#include <shared_mutex>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "common/logging.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "service/backend_options.h"
#include "util/debug_util.h"
#include "util/time.h"
+#include "util/uid_util.h"
#include "vec/core/block.h"
namespace doris {
+static Status _doReportExecStatsRpc(const TNetworkAddress& coor_addr,
+ const TReportExecStatusParams& req,
+ TReportExecStatusResult& res) {
+ Status client_status;
+ FrontendServiceConnection rpc_client(ExecEnv::GetInstance()->frontend_client_cache(), coor_addr,
+ &client_status);
+ if (!client_status.ok()) {
+ LOG_WARNING(
+ "could not get client rpc client of {} when reporting profiles, reason is {}, "
+ "not reporting, profile will be lost",
+ PrintThriftNetworkAddress(coor_addr), client_status.to_string());
+ return Status::RpcError("Client rpc client failed");
+ }
+
+ try {
+ try {
+ rpc_client->reportExecStatus(res, req);
+ } catch (const apache::thrift::transport::TTransportException& e) {
+ LOG_WARNING("Transport exception from {}, reason: {}, reopening",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ client_status = rpc_client.reopen(config::thrift_rpc_timeout_ms);
+ if (!client_status.ok()) {
+ LOG_WARNING("Reopen failed, reason: {}", client_status.to_string());
+ return Status::RpcError("Open rpc client failed");
+ }
+
+ rpc_client->reportExecStatus(res, req);
+ }
+ } catch (apache::thrift::TApplicationException& e) {
+ if (e.getType() == e.UNKNOWN_METHOD) {
+ LOG_WARNING(
+ "Failed to send statistics to {} due to {}, usually because the frontend "
+ "is not upgraded, check the version",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ } else {
+ LOG_WARNING(
+ "Failed to send statistics to {}, reason: {}, you can see fe log for "
+ "details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ }
+ return Status::RpcError("Send stats failed");
+ } catch (std::exception& e) {
+ LOG_WARNING("Failed to send statistics to {}, reason: {}, you can see fe log for details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ return Status::RpcError("Send stats failed");
+ }
+
+ return Status::OK();
+}
+
+TReportExecStatusParams RuntimeQueryStatiticsMgr::create_report_exec_status_params(
+ const TUniqueId& query_id, int32 fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile,
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile) {
+ TReportExecStatusParams req;
+ req.__set_query_id(query_id);
+ req.__set_backend_id(ExecEnv::GetInstance()->master_info()->backend_id);
+ req.__set_fragment_id(fragment_id);
+
+ std::vector<TDetailedReportParams> detailed_params;
+ for (const auto& p_profile : f_profile) {
+ TDetailedReportParams tmp;
+ tmp.__set_profile(*p_profile);
+ // tmp.fragment_instance_id is not needed for pipeline x
+ detailed_params.push_back(tmp);
+ }
+ req.__set_detailed_report(detailed_params);
+
+ if (load_channel_profile) {
+ req.__set_loadChannelProfile(*load_channel_profile);
+ }
+
+ return req;
+}
+
+void RuntimeQueryStatiticsMgr::start_report_thread() {
+ if (started.load()) {
+ DCHECK(false) << "report thread has been started";
+ LOG_ERROR("report thread has been started");
+ return;
+ }
+
+ started.store(true);
+
+ for (size_t i = 0; i < config::report_exec_status_thread_num; ++i) {
+ this->_report_profile_threads.emplace_back(std::make_unique<std::thread>(
+ &RuntimeQueryStatiticsMgr::report_query_profiles_thread, this));
+ }
+}
+
+void RuntimeQueryStatiticsMgr::report_query_profiles_thread() {
+ while (true) {
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+
+ while (_query_profile_map.empty() && _profile_map_x.empty() &&
+ !_report_profile_thread_stop) {
+ _report_profile_cv.wait(lock);
+ }
+
+ if (_report_profile_thread_stop) {
+ LOG_INFO("Report profile thread stopped");
+ return;
+ }
+ }
+
+ _report_query_profiles_function();
+ }
+}
+
+void RuntimeQueryStatiticsMgr::force_report_profile() {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+ _report_profile_cv.notify_one();
+}
+
+void RuntimeQueryStatiticsMgr::stop_report_thread() {
+ if (!started) {
+ return;
+ }
+
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+ _report_profile_thread_stop = true;
+ LOG_INFO("All report threads are going to stop");
+ _report_profile_cv.notify_all();
+ }
+
+ for (const auto& thread : _report_profile_threads) {
+ thread->join();
+ }
+
+ LOG_INFO("All report threads stopped");
+}
+
+void RuntimeQueryStatiticsMgr::register_instance_profile(
+ const TUniqueId& query_id, const TNetworkAddress& coor_addr, const TUniqueId& instance_id,
+ std::shared_ptr<TRuntimeProfileTree> instance_profile,
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile) {
+ if (instance_profile == nullptr) {
+ auto msg = fmt::format("Register instance profile {} {} failed, profile is null",
+ print_id(query_id), print_id(instance_id));
+ DCHECK(false) << msg;
+ LOG_ERROR(msg);
+ return;
+ }
+
+ std::lock_guard<std::shared_mutex> lg(_query_profile_map_lock);
+
+ if (!_query_profile_map.contains(query_id)) {
+ _query_profile_map[query_id] = std::make_tuple(
+ coor_addr, std::unordered_map<TUniqueId, std::shared_ptr<TRuntimeProfileTree>>());
+ }
+
+ std::get<1>(_query_profile_map[query_id]).insert(std::make_pair(instance_id, instance_profile));
+
+ if (load_channel_profile != nullptr) {
+ _load_channel_profile_map[instance_id] = load_channel_profile;
+ }
+
+ LOG_INFO("Register instance profile {} {}", print_id(query_id), print_id(instance_id));
+}
+
+void RuntimeQueryStatiticsMgr::register_fragment_profile_x(
+ const TUniqueId& query_id, const TNetworkAddress& coor_addr, int32_t fragment_id,
+ std::vector<std::shared_ptr<TRuntimeProfileTree>> p_profiles,
+ std::shared_ptr<TRuntimeProfileTree> load_channel_profile_x) {
+ for (const auto& p : p_profiles) {
+ if (p == nullptr) {
+ auto msg = fmt::format("Register fragment profile {} {} failed, profile is null",
+ print_id(query_id), fragment_id);
+ DCHECK(false) << msg;
+ LOG_ERROR(msg);
+ return;
+ }
+ }
+
+ std::lock_guard<std::shared_mutex> lg(_query_profile_map_lock);
+
+ if (!_profile_map_x.contains(query_id)) {
+ _profile_map_x[query_id] = std::make_tuple(
+ coor_addr,
+ std::unordered_map<int, std::vector<std::shared_ptr<TRuntimeProfileTree>>>());
+ }
+
+ std::unordered_map<int, std::vector<std::shared_ptr<TRuntimeProfileTree>>>&
+ fragment_profile_map = std::get<1>(_profile_map_x[query_id]);
+ fragment_profile_map.insert(std::make_pair(fragment_id, p_profiles));
+
+ if (load_channel_profile_x != nullptr) {
+ _load_channel_profile_map_x[std::make_pair(query_id, fragment_id)] = load_channel_profile_x;
+ }
+
+ LOG_INFO("register x profile done {}, fragment {}, profiles {}", print_id(query_id),
+ fragment_id, p_profiles.size());
+}
+
+void RuntimeQueryStatiticsMgr::_report_query_profiles_non_pipeline() {
+ // query_id -> {coordinator_addr, {instance_id -> instance_profile}}
+ decltype(_query_profile_map) profile_copy;
+ decltype(_load_channel_profile_map) load_channel_profile_copy;
+
+ {
+ std::lock_guard<std::shared_mutex> lg(_query_profile_map_lock);
+ profile_copy = _query_profile_map;
+ load_channel_profile_copy = _load_channel_profile_map;
+ _query_profile_map.clear();
+ _load_channel_profile_map.clear();
+ }
+
+ // query_id -> {coordinator_addr, {instance_id -> instance_profile}}
+ for (const auto& entry : profile_copy) {
+ const auto& query_id = entry.first;
+ const auto& coor_addr = std::get<0>(entry.second);
+ const std::unordered_map<TUniqueId, std::shared_ptr<TRuntimeProfileTree>>&
+ instance_id_to_profile = std::get<1>(entry.second);
+
+ for (const auto& profile_entry : instance_id_to_profile) {
+ const auto& instance_id = profile_entry.first;
+ const auto& instance_profile = profile_entry.second;
+
+ if (instance_profile == nullptr) {
+ auto msg = fmt::format("Query {} instance {} profile is null", print_id(query_id),
+ print_id(instance_id));
+ DCHECK(false) << msg;
+ LOG_ERROR(msg);
+ continue;
+ }
+
+ TReportExecStatusParams req;
+ req.__set_query_id(query_id);
+ req.__set_fragment_instance_id(instance_id);
+ // instance profile will not be added to RuntimeQueryStatiticsMgr unless it is closed
+ req.__set_done(true);
+ req.__set_profile(*instance_profile);
+
+ if (load_channel_profile_copy.contains(instance_id)) {
+ req.__set_loadChannelProfile(*load_channel_profile_copy[instance_id]);
+ }
+
+ TReportExecStatusResult res;
+ auto rpc_status = _doReportExecStatsRpc(coor_addr, req, res);
+
+ if (res.status.status_code != TStatusCode::OK) {
+ std::stringstream ss;
+ res.status.printTo(ss);
+ LOG_WARNING("Query {} send profile to {} failed, msg: {}", print_id(query_id),
+ PrintThriftNetworkAddress(coor_addr), ss.str());
+ } else {
+ LOG_INFO("Send {} profile finished", print_id(query_id));
+ }
+ }
+ }
+}
+
+void RuntimeQueryStatiticsMgr::_report_query_profiles_x() {
Review Comment:
warning: method '_report_query_profiles_x' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/runtime_query_statistics_mgr.h:123:
```diff
- void _report_query_profiles_x();
+ static void _report_query_profiles_x();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](profile) WIP [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #33015:
URL: https://github.com/apache/doris/pull/33015#discussion_r1547114067
##########
be/src/service/backend_service.cpp:
##########
@@ -1129,4 +1133,51 @@ void BaseBackendService::warm_up_tablets(TWarmUpTabletsResponse& response,
response.__set_status(Status::NotSupported("warm_up_tablets is not implemented").to_thrift());
}
+void BaseBackendService::get_realtime_query_exec_status(
Review Comment:
warning: method 'get_realtime_query_exec_status' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void BaseBackendService::get_realtime_query_exec_status(
```
##########
be/src/runtime/runtime_query_statistics_mgr.h:
##########
@@ -75,9 +82,65 @@ class RuntimeQueryStatiticsMgr {
// used for backend_active_tasks
void get_active_be_tasks_block(vectorized::Block* block);
+ void start_report_thread();
+ void report_query_profiles_thread();
+ void force_report_profile();
+ void stop_report_thread();
+
+ void submit_report_status_task(const TUniqueId& q_id, const TNetworkAddress& coor_addr,
+ int32 f_id, const TUniqueId& i_id, int be_num, bool done,
+ Status exec_status);
+
+ void register_instance_profile(const TUniqueId& query_id, const TNetworkAddress& coor_addr,
+ const TUniqueId& instance_id,
+ std::shared_ptr<TRuntimeProfileTree> i_profile);
+
+ void register_fragment_profile_x(const TUniqueId& query_id, const TNetworkAddress& const_addr,
+ int32_t fragment_id,
+ std::vector<std::shared_ptr<TRuntimeProfileTree>> p_profiles);
+ using CoorAddrWithFragmentProfileMap =
+ std::tuple<TNetworkAddress,
+ std::unordered_map<int, std::vector<std::shared_ptr<TRuntimeProfileTree>>>>;
+
+ using CoorAddrWithInstanceProfileMap =
+ std::tuple<TNetworkAddress,
+ std::unordered_map<TUniqueId, std::shared_ptr<TRuntimeProfileTree>>>;
+
+ static TReportExecStatusParams create_report_exec_status_params(
+ const TUniqueId& q_id, int32 f_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile);
+
+ static void report_query_profile(
+ const TUniqueId&, const TNetworkAddress&,
+ const std::unordered_map<int, std::vector<std::shared_ptr<TRuntimeProfileTree>>>&);
Review Comment:
warning: parameter 3 is const-qualified in the function declaration; const-qualification of parameters only has an effect in function definitions [readability-avoid-const-params-in-decls]
```suggestion
std::unordered_map<int, std::vector<std::shared_ptr<TRuntimeProfileTree>>>&);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [refactor](profile) WIP [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #33015:
URL: https://github.com/apache/doris/pull/33015#discussion_r1547513280
##########
be/src/runtime/plan_fragment_executor.h:
##########
@@ -21,6 +21,7 @@
#pragma once
#include <gen_cpp/PaloInternalService_types.h>
Review Comment:
warning: 'gen_cpp/PaloInternalService_types.h' file not found [clang-diagnostic-error]
```cpp
#include <gen_cpp/PaloInternalService_types.h>
^
```
##########
be/src/runtime/fragment_mgr.cpp:
##########
@@ -1609,4 +1616,27 @@ void FragmentMgr::get_runtime_query_info(std::vector<WorkloadQueryInfo>* query_i
}
}
+Status FragmentMgr::collect_realtime_profile_x(
Review Comment:
warning: method 'collect_realtime_profile_x' can be made static [readability-convert-member-functions-to-static]
```suggestion
static Status FragmentMgr::collect_realtime_profile_x(
```
##########
be/src/runtime/query_context.cpp:
##########
@@ -293,4 +310,81 @@
return Status::OK();
}
+void QueryContext::add_fragment_profile_x(
+ int fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& pipeline_profiles) {
+#ifndef NDEBUG
+ for (const auto& p : pipeline_profiles) {
+ DCHECK(p != nullptr) << fmt::format("Add pipeline profile failed, query {}, fragment {}",
+ print_id(this->_query_id), fragment_id);
+ }
+#endif
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ LOG_INFO("Query X add fragment profile, query {}, fragment {}, pipeline profile count {} ",
+ print_id(this->_query_id), fragment_id, pipeline_profiles.size());
+ _profile_map_x.insert(std::make_pair(fragment_id, pipeline_profiles));
+}
+
+void QueryContext::add_instance_profile(const TUniqueId& instance_id,
+ std::shared_ptr<TRuntimeProfileTree> profile) {
+ DCHECK(profile != nullptr) << print_id(instance_id);
+
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ _profile_map.insert(std::make_pair(instance_id, profile));
+}
+
+void QueryContext::_async_report_profile() {
+ if (!enable_profile()) {
+ return;
+ }
+
+ _async_report_profile_x();
+ _async_report_profile_non_pipeline();
+}
+
+void QueryContext::_async_report_profile_non_pipeline() {
+ if (enable_pipeline_exec()) {
+ return;
+ }
+
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ LOG_INFO("Query {}, register query profile, instance profile count {}", print_id(_query_id),
+ _profile_map.size());
+
+ for (auto& [instance_id, instance_profile] : _profile_map) {
+ ExecEnv::GetInstance()->runtime_query_statistics_mgr()->register_instance_profile(
+ _query_id, this->coord_addr, instance_id, instance_profile);
+ }
+}
+
+void QueryContext::_async_report_profile_x() {
Review Comment:
warning: method '_async_report_profile_x' can be made const [readability-make-member-function-const]
```suggestion
void QueryContext::_async_report_profile_x() const {
```
be/src/runtime/query_context.h:383:
```diff
- void _async_report_profile_x();
+ void _async_report_profile_x() const;
```
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -17,14 +17,300 @@
#include "runtime/runtime_query_statistics_mgr.h"
+#include <gen_cpp/FrontendService_types.h>
+#include <gen_cpp/RuntimeProfile_types.h>
+#include <gen_cpp/Status_types.h>
+#include <gen_cpp/Types_types.h>
+#include <thrift/TApplicationException.h>
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <random>
+#include <shared_mutex>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "common/logging.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "service/backend_options.h"
#include "util/debug_util.h"
#include "util/time.h"
+#include "util/uid_util.h"
#include "vec/core/block.h"
namespace doris {
+static Status _doReportExecStatsRpc(const TNetworkAddress& coor_addr,
+ const TReportExecStatusParams& req,
+ TReportExecStatusResult& res) {
+ Status client_status;
+ FrontendServiceConnection rpc_client(ExecEnv::GetInstance()->frontend_client_cache(), coor_addr,
+ &client_status);
+ if (!client_status.ok()) {
+ LOG_WARNING(
+ "could not get client rpc client of {} when reporting profiles, reason is {}, "
+ "not reporting, profile will be lost",
+ PrintThriftNetworkAddress(coor_addr), client_status.to_string());
+ return Status::RpcError("Client rpc client failed");
+ }
+
+ try {
+ try {
+ rpc_client->reportExecStatus(res, req);
+ } catch (const apache::thrift::transport::TTransportException& e) {
+ LOG_WARNING("Transport exception from {}, reason: {}, reopening",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ client_status = rpc_client.reopen(config::thrift_rpc_timeout_ms);
+ if (!client_status.ok()) {
+ LOG_WARNING("Reopen failed, reason: {}", client_status.to_string());
+ return Status::RpcError("Open rpc client failed");
+ }
+
+ rpc_client->reportExecStatus(res, req);
+ }
+ } catch (apache::thrift::TApplicationException& e) {
+ if (e.getType() == e.UNKNOWN_METHOD) {
+ LOG_WARNING(
+ "Failed to send statistics to {} due to {}, usually because the frontend "
+ "is not upgraded, check the version",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ } else {
+ LOG_WARNING(
+ "Failed to send statistics to {}, reason: {}, you can see fe log for "
+ "details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ }
+ return Status::RpcError("Send stats failed");
+ } catch (std::exception& e) {
+ LOG_WARNING("Failed to send statistics to {}, reason: {}, you can see fe log for details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ return Status::RpcError("Send stats failed");
+ }
+
+ return Status::OK();
+}
+
+TReportExecStatusParams RuntimeQueryStatiticsMgr::create_report_exec_status_params(
+ const TUniqueId& query_id, int32 fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile) {
+ TReportExecStatusParams req;
+ req.__set_query_id(query_id);
+ req.__set_backend_id(ExecEnv::GetInstance()->master_info()->backend_id);
+ req.__set_fragment_id(fragment_id);
+ std::vector<TDetailedReportParams> detailed_params;
+ for (const auto& p_profile : f_profile) {
+ TDetailedReportParams tmp;
+ tmp.__set_profile(*p_profile);
+ // tmp.fragment_instance_id is not needed for pipeline x
+ detailed_params.push_back(tmp);
+ }
+
+ req.__set_detailed_report(detailed_params);
+ return req;
+}
+
+void RuntimeQueryStatiticsMgr::start_report_thread() {
+ if (_report_profile_thread == nullptr) {
+ _report_profile_thread = std::make_unique<std::thread>(
+ &RuntimeQueryStatiticsMgr::report_query_profiles_thread, this);
+ }
+}
+
+void RuntimeQueryStatiticsMgr::report_query_profiles_thread() {
+ while (true) {
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+
+ while (!_force_report_profile) {
+ _report_profile_cv.wait_for(lock, std::chrono::milliseconds(5000));
+ break;
+ }
+ }
+
+ _report_query_profiles_function();
+ _force_report_profile = false;
+
+ if (_report_profile_thread_stop) {
+ return;
+ }
+ }
+}
+
+void RuntimeQueryStatiticsMgr::force_report_profile() {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+ _force_report_profile = true;
+ _report_profile_cv.notify_one();
+}
+
+void RuntimeQueryStatiticsMgr::stop_report_thread() {
Review Comment:
warning: method 'stop_report_thread' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/runtime_query_statistics_mgr.h:87:
```diff
- void stop_report_thread();
+ static void stop_report_thread();
```
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -17,14 +17,300 @@
#include "runtime/runtime_query_statistics_mgr.h"
+#include <gen_cpp/FrontendService_types.h>
+#include <gen_cpp/RuntimeProfile_types.h>
+#include <gen_cpp/Status_types.h>
+#include <gen_cpp/Types_types.h>
+#include <thrift/TApplicationException.h>
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <random>
+#include <shared_mutex>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "common/logging.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "service/backend_options.h"
#include "util/debug_util.h"
#include "util/time.h"
+#include "util/uid_util.h"
#include "vec/core/block.h"
namespace doris {
+static Status _doReportExecStatsRpc(const TNetworkAddress& coor_addr,
+ const TReportExecStatusParams& req,
+ TReportExecStatusResult& res) {
+ Status client_status;
+ FrontendServiceConnection rpc_client(ExecEnv::GetInstance()->frontend_client_cache(), coor_addr,
+ &client_status);
+ if (!client_status.ok()) {
+ LOG_WARNING(
+ "could not get client rpc client of {} when reporting profiles, reason is {}, "
+ "not reporting, profile will be lost",
+ PrintThriftNetworkAddress(coor_addr), client_status.to_string());
+ return Status::RpcError("Client rpc client failed");
+ }
+
+ try {
+ try {
+ rpc_client->reportExecStatus(res, req);
+ } catch (const apache::thrift::transport::TTransportException& e) {
+ LOG_WARNING("Transport exception from {}, reason: {}, reopening",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ client_status = rpc_client.reopen(config::thrift_rpc_timeout_ms);
+ if (!client_status.ok()) {
+ LOG_WARNING("Reopen failed, reason: {}", client_status.to_string());
+ return Status::RpcError("Open rpc client failed");
+ }
+
+ rpc_client->reportExecStatus(res, req);
+ }
+ } catch (apache::thrift::TApplicationException& e) {
+ if (e.getType() == e.UNKNOWN_METHOD) {
+ LOG_WARNING(
+ "Failed to send statistics to {} due to {}, usually because the frontend "
+ "is not upgraded, check the version",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ } else {
+ LOG_WARNING(
+ "Failed to send statistics to {}, reason: {}, you can see fe log for "
+ "details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ }
+ return Status::RpcError("Send stats failed");
+ } catch (std::exception& e) {
+ LOG_WARNING("Failed to send statistics to {}, reason: {}, you can see fe log for details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ return Status::RpcError("Send stats failed");
+ }
+
+ return Status::OK();
+}
+
+TReportExecStatusParams RuntimeQueryStatiticsMgr::create_report_exec_status_params(
+ const TUniqueId& query_id, int32 fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile) {
+ TReportExecStatusParams req;
+ req.__set_query_id(query_id);
+ req.__set_backend_id(ExecEnv::GetInstance()->master_info()->backend_id);
+ req.__set_fragment_id(fragment_id);
+ std::vector<TDetailedReportParams> detailed_params;
+ for (const auto& p_profile : f_profile) {
+ TDetailedReportParams tmp;
+ tmp.__set_profile(*p_profile);
+ // tmp.fragment_instance_id is not needed for pipeline x
+ detailed_params.push_back(tmp);
+ }
+
+ req.__set_detailed_report(detailed_params);
+ return req;
+}
+
+void RuntimeQueryStatiticsMgr::start_report_thread() {
Review Comment:
warning: method 'start_report_thread' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/runtime_query_statistics_mgr.h:84:
```diff
- void start_report_thread();
+ static void start_report_thread();
```
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -17,14 +17,300 @@
#include "runtime/runtime_query_statistics_mgr.h"
+#include <gen_cpp/FrontendService_types.h>
+#include <gen_cpp/RuntimeProfile_types.h>
+#include <gen_cpp/Status_types.h>
+#include <gen_cpp/Types_types.h>
+#include <thrift/TApplicationException.h>
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <random>
+#include <shared_mutex>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "common/logging.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "service/backend_options.h"
#include "util/debug_util.h"
#include "util/time.h"
+#include "util/uid_util.h"
#include "vec/core/block.h"
namespace doris {
+static Status _doReportExecStatsRpc(const TNetworkAddress& coor_addr,
+ const TReportExecStatusParams& req,
+ TReportExecStatusResult& res) {
+ Status client_status;
+ FrontendServiceConnection rpc_client(ExecEnv::GetInstance()->frontend_client_cache(), coor_addr,
+ &client_status);
+ if (!client_status.ok()) {
+ LOG_WARNING(
+ "could not get client rpc client of {} when reporting profiles, reason is {}, "
+ "not reporting, profile will be lost",
+ PrintThriftNetworkAddress(coor_addr), client_status.to_string());
+ return Status::RpcError("Client rpc client failed");
+ }
+
+ try {
+ try {
+ rpc_client->reportExecStatus(res, req);
+ } catch (const apache::thrift::transport::TTransportException& e) {
+ LOG_WARNING("Transport exception from {}, reason: {}, reopening",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ client_status = rpc_client.reopen(config::thrift_rpc_timeout_ms);
+ if (!client_status.ok()) {
+ LOG_WARNING("Reopen failed, reason: {}", client_status.to_string());
+ return Status::RpcError("Open rpc client failed");
+ }
+
+ rpc_client->reportExecStatus(res, req);
+ }
+ } catch (apache::thrift::TApplicationException& e) {
+ if (e.getType() == e.UNKNOWN_METHOD) {
+ LOG_WARNING(
+ "Failed to send statistics to {} due to {}, usually because the frontend "
+ "is not upgraded, check the version",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ } else {
+ LOG_WARNING(
+ "Failed to send statistics to {}, reason: {}, you can see fe log for "
+ "details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ }
+ return Status::RpcError("Send stats failed");
+ } catch (std::exception& e) {
+ LOG_WARNING("Failed to send statistics to {}, reason: {}, you can see fe log for details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ return Status::RpcError("Send stats failed");
+ }
+
+ return Status::OK();
+}
+
+TReportExecStatusParams RuntimeQueryStatiticsMgr::create_report_exec_status_params(
+ const TUniqueId& query_id, int32 fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile) {
+ TReportExecStatusParams req;
+ req.__set_query_id(query_id);
+ req.__set_backend_id(ExecEnv::GetInstance()->master_info()->backend_id);
+ req.__set_fragment_id(fragment_id);
+ std::vector<TDetailedReportParams> detailed_params;
+ for (const auto& p_profile : f_profile) {
+ TDetailedReportParams tmp;
+ tmp.__set_profile(*p_profile);
+ // tmp.fragment_instance_id is not needed for pipeline x
+ detailed_params.push_back(tmp);
+ }
+
+ req.__set_detailed_report(detailed_params);
+ return req;
+}
+
+void RuntimeQueryStatiticsMgr::start_report_thread() {
+ if (_report_profile_thread == nullptr) {
+ _report_profile_thread = std::make_unique<std::thread>(
+ &RuntimeQueryStatiticsMgr::report_query_profiles_thread, this);
+ }
+}
+
+void RuntimeQueryStatiticsMgr::report_query_profiles_thread() {
+ while (true) {
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+
+ while (!_force_report_profile) {
+ _report_profile_cv.wait_for(lock, std::chrono::milliseconds(5000));
+ break;
+ }
+ }
+
+ _report_query_profiles_function();
+ _force_report_profile = false;
+
+ if (_report_profile_thread_stop) {
+ return;
+ }
+ }
+}
+
+void RuntimeQueryStatiticsMgr::force_report_profile() {
Review Comment:
warning: method 'force_report_profile' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/runtime_query_statistics_mgr.h:86:
```diff
- void force_report_profile();
+ static void force_report_profile();
```
##########
be/src/runtime/plan_fragment_executor.cpp:
##########
@@ -644,4 +646,11 @@ void PlanFragmentExecutor::close() {
_closed = true;
}
+std::shared_ptr<TRuntimeProfileTree> PlanFragmentExecutor::collect_realtime_profile() const {
Review Comment:
warning: method 'collect_realtime_profile' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/plan_fragment_executor.h:149:
```diff
- std::shared_ptr<TRuntimeProfileTree> collect_realtime_profile() const;
+ static std::shared_ptr<TRuntimeProfileTree> collect_realtime_profile() ;
```
```suggestion
std::shared_ptr<TRuntimeProfileTree> PlanFragmentExecutor::collect_realtime_profile() {
```
##########
be/src/runtime/query_context.cpp:
##########
@@ -293,4 +310,81 @@ Status QueryContext::set_workload_group(WorkloadGroupPtr& tg) {
return Status::OK();
}
+void QueryContext::add_fragment_profile_x(
+ int fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& pipeline_profiles) {
+#ifndef NDEBUG
+ for (const auto& p : pipeline_profiles) {
+ DCHECK(p != nullptr) << fmt::format("Add pipeline profile failed, query {}, fragment {}",
+ print_id(this->_query_id), fragment_id);
+ }
+#endif
+ std::lock_guard<std::mutex> l(_profile_mutex);
+ LOG_INFO("Query X add fragment profile, query {}, fragment {}, pipeline profile count {} ",
+ print_id(this->_query_id), fragment_id, pipeline_profiles.size());
+ _profile_map_x.insert(std::make_pair(fragment_id, pipeline_profiles));
+}
+
+void QueryContext::add_instance_profile(const TUniqueId& instance_id,
+ std::shared_ptr<TRuntimeProfileTree> profile) {
+ DCHECK(profile != nullptr) << print_id(instance_id);
+
+ std::lock_guard<std::mutex> lg(_profile_mutex);
+ _profile_map.insert(std::make_pair(instance_id, profile));
+}
+
+void QueryContext::_async_report_profile() {
+ if (!enable_profile()) {
+ return;
+ }
+
+ _async_report_profile_x();
+ _async_report_profile_non_pipeline();
+}
+
+void QueryContext::_async_report_profile_non_pipeline() {
Review Comment:
warning: method '_async_report_profile_non_pipeline' can be made const [readability-make-member-function-const]
```suggestion
void QueryContext::_async_report_profile_non_pipeline() const {
```
be/src/runtime/query_context.h:382:
```diff
- void _async_report_profile_non_pipeline();
+ void _async_report_profile_non_pipeline() const;
```
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -17,14 +17,300 @@
#include "runtime/runtime_query_statistics_mgr.h"
+#include <gen_cpp/FrontendService_types.h>
+#include <gen_cpp/RuntimeProfile_types.h>
+#include <gen_cpp/Status_types.h>
+#include <gen_cpp/Types_types.h>
+#include <thrift/TApplicationException.h>
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <random>
+#include <shared_mutex>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "common/logging.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "service/backend_options.h"
#include "util/debug_util.h"
#include "util/time.h"
+#include "util/uid_util.h"
#include "vec/core/block.h"
namespace doris {
+static Status _doReportExecStatsRpc(const TNetworkAddress& coor_addr,
+ const TReportExecStatusParams& req,
+ TReportExecStatusResult& res) {
+ Status client_status;
+ FrontendServiceConnection rpc_client(ExecEnv::GetInstance()->frontend_client_cache(), coor_addr,
+ &client_status);
+ if (!client_status.ok()) {
+ LOG_WARNING(
+ "could not get client rpc client of {} when reporting profiles, reason is {}, "
+ "not reporting, profile will be lost",
+ PrintThriftNetworkAddress(coor_addr), client_status.to_string());
+ return Status::RpcError("Client rpc client failed");
+ }
+
+ try {
+ try {
+ rpc_client->reportExecStatus(res, req);
+ } catch (const apache::thrift::transport::TTransportException& e) {
+ LOG_WARNING("Transport exception from {}, reason: {}, reopening",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ client_status = rpc_client.reopen(config::thrift_rpc_timeout_ms);
+ if (!client_status.ok()) {
+ LOG_WARNING("Reopen failed, reason: {}", client_status.to_string());
+ return Status::RpcError("Open rpc client failed");
+ }
+
+ rpc_client->reportExecStatus(res, req);
+ }
+ } catch (apache::thrift::TApplicationException& e) {
+ if (e.getType() == e.UNKNOWN_METHOD) {
+ LOG_WARNING(
+ "Failed to send statistics to {} due to {}, usually because the frontend "
+ "is not upgraded, check the version",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ } else {
+ LOG_WARNING(
+ "Failed to send statistics to {}, reason: {}, you can see fe log for "
+ "details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ }
+ return Status::RpcError("Send stats failed");
+ } catch (std::exception& e) {
+ LOG_WARNING("Failed to send statistics to {}, reason: {}, you can see fe log for details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ return Status::RpcError("Send stats failed");
+ }
+
+ return Status::OK();
+}
+
+TReportExecStatusParams RuntimeQueryStatiticsMgr::create_report_exec_status_params(
+ const TUniqueId& query_id, int32 fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile) {
+ TReportExecStatusParams req;
+ req.__set_query_id(query_id);
+ req.__set_backend_id(ExecEnv::GetInstance()->master_info()->backend_id);
+ req.__set_fragment_id(fragment_id);
+ std::vector<TDetailedReportParams> detailed_params;
+ for (const auto& p_profile : f_profile) {
+ TDetailedReportParams tmp;
+ tmp.__set_profile(*p_profile);
+ // tmp.fragment_instance_id is not needed for pipeline x
+ detailed_params.push_back(tmp);
+ }
+
+ req.__set_detailed_report(detailed_params);
+ return req;
+}
+
+void RuntimeQueryStatiticsMgr::start_report_thread() {
+ if (_report_profile_thread == nullptr) {
+ _report_profile_thread = std::make_unique<std::thread>(
+ &RuntimeQueryStatiticsMgr::report_query_profiles_thread, this);
+ }
+}
+
+void RuntimeQueryStatiticsMgr::report_query_profiles_thread() {
+ while (true) {
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+
+ while (!_force_report_profile) {
+ _report_profile_cv.wait_for(lock, std::chrono::milliseconds(5000));
+ break;
+ }
+ }
+
+ _report_query_profiles_function();
+ _force_report_profile = false;
+
+ if (_report_profile_thread_stop) {
+ return;
+ }
+ }
+}
+
+void RuntimeQueryStatiticsMgr::force_report_profile() {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+ _force_report_profile = true;
+ _report_profile_cv.notify_one();
+}
+
+void RuntimeQueryStatiticsMgr::stop_report_thread() {
+ if (_report_profile_thread == nullptr) {
+ return;
+ }
+
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+ _force_report_profile = true;
+ _report_profile_thread_stop = true;
+ _report_profile_cv.notify_one();
+ }
+
+ _report_profile_thread->join();
+ _report_profile_thread.reset();
+}
+
+void RuntimeQueryStatiticsMgr::register_instance_profile(
Review Comment:
warning: method 'register_instance_profile' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void RuntimeQueryStatiticsMgr::register_instance_profile(
```
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -17,14 +17,300 @@
#include "runtime/runtime_query_statistics_mgr.h"
+#include <gen_cpp/FrontendService_types.h>
+#include <gen_cpp/RuntimeProfile_types.h>
+#include <gen_cpp/Status_types.h>
+#include <gen_cpp/Types_types.h>
+#include <thrift/TApplicationException.h>
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <random>
+#include <shared_mutex>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "common/logging.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "service/backend_options.h"
#include "util/debug_util.h"
#include "util/time.h"
+#include "util/uid_util.h"
#include "vec/core/block.h"
namespace doris {
+static Status _doReportExecStatsRpc(const TNetworkAddress& coor_addr,
+ const TReportExecStatusParams& req,
+ TReportExecStatusResult& res) {
+ Status client_status;
+ FrontendServiceConnection rpc_client(ExecEnv::GetInstance()->frontend_client_cache(), coor_addr,
+ &client_status);
+ if (!client_status.ok()) {
+ LOG_WARNING(
+ "could not get client rpc client of {} when reporting profiles, reason is {}, "
+ "not reporting, profile will be lost",
+ PrintThriftNetworkAddress(coor_addr), client_status.to_string());
+ return Status::RpcError("Client rpc client failed");
+ }
+
+ try {
+ try {
+ rpc_client->reportExecStatus(res, req);
+ } catch (const apache::thrift::transport::TTransportException& e) {
+ LOG_WARNING("Transport exception from {}, reason: {}, reopening",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ client_status = rpc_client.reopen(config::thrift_rpc_timeout_ms);
+ if (!client_status.ok()) {
+ LOG_WARNING("Reopen failed, reason: {}", client_status.to_string());
+ return Status::RpcError("Open rpc client failed");
+ }
+
+ rpc_client->reportExecStatus(res, req);
+ }
+ } catch (apache::thrift::TApplicationException& e) {
+ if (e.getType() == e.UNKNOWN_METHOD) {
+ LOG_WARNING(
+ "Failed to send statistics to {} due to {}, usually because the frontend "
+ "is not upgraded, check the version",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ } else {
+ LOG_WARNING(
+ "Failed to send statistics to {}, reason: {}, you can see fe log for "
+ "details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ }
+ return Status::RpcError("Send stats failed");
+ } catch (std::exception& e) {
+ LOG_WARNING("Failed to send statistics to {}, reason: {}, you can see fe log for details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ return Status::RpcError("Send stats failed");
+ }
+
+ return Status::OK();
+}
+
+TReportExecStatusParams RuntimeQueryStatiticsMgr::create_report_exec_status_params(
+ const TUniqueId& query_id, int32 fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile) {
+ TReportExecStatusParams req;
+ req.__set_query_id(query_id);
+ req.__set_backend_id(ExecEnv::GetInstance()->master_info()->backend_id);
+ req.__set_fragment_id(fragment_id);
+ std::vector<TDetailedReportParams> detailed_params;
+ for (const auto& p_profile : f_profile) {
+ TDetailedReportParams tmp;
+ tmp.__set_profile(*p_profile);
+ // tmp.fragment_instance_id is not needed for pipeline x
+ detailed_params.push_back(tmp);
+ }
+
+ req.__set_detailed_report(detailed_params);
+ return req;
+}
+
+void RuntimeQueryStatiticsMgr::start_report_thread() {
+ if (_report_profile_thread == nullptr) {
+ _report_profile_thread = std::make_unique<std::thread>(
+ &RuntimeQueryStatiticsMgr::report_query_profiles_thread, this);
+ }
+}
+
+void RuntimeQueryStatiticsMgr::report_query_profiles_thread() {
+ while (true) {
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+
+ while (!_force_report_profile) {
+ _report_profile_cv.wait_for(lock, std::chrono::milliseconds(5000));
+ break;
+ }
+ }
+
+ _report_query_profiles_function();
+ _force_report_profile = false;
+
+ if (_report_profile_thread_stop) {
+ return;
+ }
+ }
+}
+
+void RuntimeQueryStatiticsMgr::force_report_profile() {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+ _force_report_profile = true;
+ _report_profile_cv.notify_one();
+}
+
+void RuntimeQueryStatiticsMgr::stop_report_thread() {
+ if (_report_profile_thread == nullptr) {
+ return;
+ }
+
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+ _force_report_profile = true;
+ _report_profile_thread_stop = true;
+ _report_profile_cv.notify_one();
+ }
+
+ _report_profile_thread->join();
+ _report_profile_thread.reset();
+}
+
+void RuntimeQueryStatiticsMgr::register_instance_profile(
+ const TUniqueId& query_id, const TNetworkAddress& coor_addr, const TUniqueId& instance_id,
+ std::shared_ptr<TRuntimeProfileTree> instance_profile) {
+ if (instance_profile == nullptr) {
+ auto msg = fmt::format("Register instance profile {} {} failed, profile is null",
+ print_id(query_id), print_id(instance_id));
+ DCHECK(false) << msg;
+ LOG_ERROR(msg);
+ return;
+ }
+
+ std::lock_guard<std::shared_mutex> lg(_query_profile_map_lock);
+
+ if (!_profile_map.contains(query_id)) {
+ _profile_map[query_id] = std::make_tuple(
+ coor_addr, std::unordered_map<TUniqueId, std::shared_ptr<TRuntimeProfileTree>>());
+ }
+
+ std::get<1>(_profile_map[query_id]).insert(std::make_pair(instance_id, instance_profile));
+
+ LOG_INFO("Register instance profile {} {}", print_id(query_id), print_id(instance_id));
+}
+
+void RuntimeQueryStatiticsMgr::register_fragment_profile_x(
+ const TUniqueId& query_id, const TNetworkAddress& coor_addr, int32_t fragment_id,
+ std::vector<std::shared_ptr<TRuntimeProfileTree>> p_profiles) {
+ for (const auto& p : p_profiles) {
+ if (p == nullptr) {
+ auto msg = fmt::format("Register fragment profile {} {} failed, profile is null",
+ print_id(query_id), fragment_id);
+ DCHECK(false) << msg;
+ LOG_ERROR(msg);
+ return;
+ }
+ }
+
+ std::lock_guard<std::shared_mutex> lg(_query_profile_map_lock);
+
+ if (!_profile_map_x.contains(query_id)) {
+ _profile_map_x[query_id] = std::make_tuple(
+ coor_addr,
+ std::unordered_map<int, std::vector<std::shared_ptr<TRuntimeProfileTree>>>());
+ }
+
+ std::get<1>(_profile_map_x[query_id]).insert(std::make_pair(fragment_id, p_profiles));
+
+ LOG_INFO("register x profile done {}, fragment {}, profiles {}", print_id(query_id),
+ fragment_id, p_profiles.size());
+}
+
+void RuntimeQueryStatiticsMgr::_report_query_profiles_non_pipeline() {
Review Comment:
warning: method '_report_query_profiles_non_pipeline' can be made static [readability-convert-member-functions-to-static]
be/src/runtime/runtime_query_statistics_mgr.h:128:
```diff
- void _report_query_profiles_non_pipeline();
+ static void _report_query_profiles_non_pipeline();
```
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -17,14 +17,300 @@
#include "runtime/runtime_query_statistics_mgr.h"
+#include <gen_cpp/FrontendService_types.h>
+#include <gen_cpp/RuntimeProfile_types.h>
+#include <gen_cpp/Status_types.h>
+#include <gen_cpp/Types_types.h>
+#include <thrift/TApplicationException.h>
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <random>
+#include <shared_mutex>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "common/logging.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
+#include "service/backend_options.h"
#include "util/debug_util.h"
#include "util/time.h"
+#include "util/uid_util.h"
#include "vec/core/block.h"
namespace doris {
+static Status _doReportExecStatsRpc(const TNetworkAddress& coor_addr,
+ const TReportExecStatusParams& req,
+ TReportExecStatusResult& res) {
+ Status client_status;
+ FrontendServiceConnection rpc_client(ExecEnv::GetInstance()->frontend_client_cache(), coor_addr,
+ &client_status);
+ if (!client_status.ok()) {
+ LOG_WARNING(
+ "could not get client rpc client of {} when reporting profiles, reason is {}, "
+ "not reporting, profile will be lost",
+ PrintThriftNetworkAddress(coor_addr), client_status.to_string());
+ return Status::RpcError("Client rpc client failed");
+ }
+
+ try {
+ try {
+ rpc_client->reportExecStatus(res, req);
+ } catch (const apache::thrift::transport::TTransportException& e) {
+ LOG_WARNING("Transport exception from {}, reason: {}, reopening",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ client_status = rpc_client.reopen(config::thrift_rpc_timeout_ms);
+ if (!client_status.ok()) {
+ LOG_WARNING("Reopen failed, reason: {}", client_status.to_string());
+ return Status::RpcError("Open rpc client failed");
+ }
+
+ rpc_client->reportExecStatus(res, req);
+ }
+ } catch (apache::thrift::TApplicationException& e) {
+ if (e.getType() == e.UNKNOWN_METHOD) {
+ LOG_WARNING(
+ "Failed to send statistics to {} due to {}, usually because the frontend "
+ "is not upgraded, check the version",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ } else {
+ LOG_WARNING(
+ "Failed to send statistics to {}, reason: {}, you can see fe log for "
+ "details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ }
+ return Status::RpcError("Send stats failed");
+ } catch (std::exception& e) {
+ LOG_WARNING("Failed to send statistics to {}, reason: {}, you can see fe log for details.",
+ PrintThriftNetworkAddress(coor_addr), e.what());
+ return Status::RpcError("Send stats failed");
+ }
+
+ return Status::OK();
+}
+
+TReportExecStatusParams RuntimeQueryStatiticsMgr::create_report_exec_status_params(
+ const TUniqueId& query_id, int32 fragment_id,
+ const std::vector<std::shared_ptr<TRuntimeProfileTree>>& f_profile) {
+ TReportExecStatusParams req;
+ req.__set_query_id(query_id);
+ req.__set_backend_id(ExecEnv::GetInstance()->master_info()->backend_id);
+ req.__set_fragment_id(fragment_id);
+ std::vector<TDetailedReportParams> detailed_params;
+ for (const auto& p_profile : f_profile) {
+ TDetailedReportParams tmp;
+ tmp.__set_profile(*p_profile);
+ // tmp.fragment_instance_id is not needed for pipeline x
+ detailed_params.push_back(tmp);
+ }
+
+ req.__set_detailed_report(detailed_params);
+ return req;
+}
+
+void RuntimeQueryStatiticsMgr::start_report_thread() {
+ if (_report_profile_thread == nullptr) {
+ _report_profile_thread = std::make_unique<std::thread>(
+ &RuntimeQueryStatiticsMgr::report_query_profiles_thread, this);
+ }
+}
+
+void RuntimeQueryStatiticsMgr::report_query_profiles_thread() {
+ while (true) {
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+
+ while (!_force_report_profile) {
+ _report_profile_cv.wait_for(lock, std::chrono::milliseconds(5000));
+ break;
+ }
+ }
+
+ _report_query_profiles_function();
+ _force_report_profile = false;
+
+ if (_report_profile_thread_stop) {
+ return;
+ }
+ }
+}
+
+void RuntimeQueryStatiticsMgr::force_report_profile() {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+ _force_report_profile = true;
+ _report_profile_cv.notify_one();
+}
+
+void RuntimeQueryStatiticsMgr::stop_report_thread() {
+ if (_report_profile_thread == nullptr) {
+ return;
+ }
+
+ {
+ std::unique_lock<std::mutex> lock(_report_profile_mutex);
+ _force_report_profile = true;
+ _report_profile_thread_stop = true;
+ _report_profile_cv.notify_one();
+ }
+
+ _report_profile_thread->join();
+ _report_profile_thread.reset();
+}
+
+void RuntimeQueryStatiticsMgr::register_instance_profile(
+ const TUniqueId& query_id, const TNetworkAddress& coor_addr, const TUniqueId& instance_id,
+ std::shared_ptr<TRuntimeProfileTree> instance_profile) {
+ if (instance_profile == nullptr) {
+ auto msg = fmt::format("Register instance profile {} {} failed, profile is null",
+ print_id(query_id), print_id(instance_id));
+ DCHECK(false) << msg;
+ LOG_ERROR(msg);
+ return;
+ }
+
+ std::lock_guard<std::shared_mutex> lg(_query_profile_map_lock);
+
+ if (!_profile_map.contains(query_id)) {
+ _profile_map[query_id] = std::make_tuple(
+ coor_addr, std::unordered_map<TUniqueId, std::shared_ptr<TRuntimeProfileTree>>());
+ }
+
+ std::get<1>(_profile_map[query_id]).insert(std::make_pair(instance_id, instance_profile));
+
+ LOG_INFO("Register instance profile {} {}", print_id(query_id), print_id(instance_id));
+}
+
+void RuntimeQueryStatiticsMgr::register_fragment_profile_x(
Review Comment:
warning: method 'register_fragment_profile_x' can be made static [readability-convert-member-functions-to-static]
```suggestion
static void RuntimeQueryStatiticsMgr::register_fragment_profile_x(
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org