This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 11a9a48cd4d [branch-3.0](log) Pick all BE execution log reduction
(#43579)
11a9a48cd4d is described below
commit 11a9a48cd4d27a513199bf70278e638920534e4f
Author: zclllhhjj <zhaochan...@selectdb.com>
AuthorDate: Wed Nov 13 19:19:27 2024 +0800
[branch-3.0](log) Pick all BE execution log reduction (#43579)
pick https://github.com/apache/doris/pull/42383
pick https://github.com/apache/doris/pull/42900
---
be/src/common/status.h | 2 +-
be/src/pipeline/exec/result_sink_operator.cpp | 11 ++++++-----
be/src/pipeline/pipeline_fragment_context.cpp | 5 -----
be/src/runtime/fragment_mgr.cpp | 24 +++++++++---------------
be/src/runtime/query_context.cpp | 5 ++---
5 files changed, 18 insertions(+), 29 deletions(-)
diff --git a/be/src/common/status.h b/be/src/common/status.h
index 234566c2b01..91386a5887e 100644
--- a/be/src/common/status.h
+++ b/be/src/common/status.h
@@ -479,7 +479,7 @@ public:
ERROR_CTOR_NOSTACK(Cancelled, CANCELLED)
ERROR_CTOR(MemoryLimitExceeded, MEM_LIMIT_EXCEEDED)
ERROR_CTOR(RpcError, THRIFT_RPC_ERROR)
- ERROR_CTOR(TimedOut, TIMEOUT)
+ ERROR_CTOR_NOSTACK(TimedOut, TIMEOUT)
ERROR_CTOR_NOSTACK(TooManyTasks, TOO_MANY_TASKS)
ERROR_CTOR(Uninitialized, UNINITIALIZED)
ERROR_CTOR(Aborted, ABORTED)
diff --git a/be/src/pipeline/exec/result_sink_operator.cpp
b/be/src/pipeline/exec/result_sink_operator.cpp
index ccd74c6d572..b8faa4f76f7 100644
--- a/be/src/pipeline/exec/result_sink_operator.cpp
+++ b/be/src/pipeline/exec/result_sink_operator.cpp
@@ -17,11 +17,11 @@
#include "result_sink_operator.h"
+#include <fmt/format.h>
+
#include <memory>
-#include <utility>
#include "common/config.h"
-#include "common/object_pool.h"
#include "exec/rowid_fetcher.h"
#include "pipeline/exec/operator.h"
#include "runtime/buffer_control_block.h"
@@ -190,9 +190,10 @@ Status ResultSinkLocalState::close(RuntimeState* state,
Status exec_status) {
final_status = st;
}
- LOG_INFO("Query {} result sink closed with status {} and has written
{} rows",
- print_id(state->query_id()),
final_status.to_string_no_stack(),
- _writer->get_written_rows());
+ VLOG_NOTICE << fmt::format(
+ "Query {} result sink closed with status {} and has written {}
rows",
+ print_id(state->query_id()), final_status.to_string_no_stack(),
+ _writer->get_written_rows());
}
// close sender, this is normal path end
diff --git a/be/src/pipeline/pipeline_fragment_context.cpp
b/be/src/pipeline/pipeline_fragment_context.cpp
index 4e05a39d77c..75623b8f3e9 100644
--- a/be/src/pipeline/pipeline_fragment_context.cpp
+++ b/be/src/pipeline/pipeline_fragment_context.cpp
@@ -252,11 +252,6 @@ Status PipelineFragmentContext::prepare(const
doris::TPipelineFragmentParams& re
auto* fragment_context = this;
- LOG_INFO("PipelineFragmentContext::prepare")
- .tag("query_id", print_id(_query_id))
- .tag("fragment_id", _fragment_id)
- .tag("pthread_id", (uintptr_t)pthread_self());
-
if (request.query_options.__isset.is_report_success) {
fragment_context->set_is_report_success(request.query_options.is_report_success);
}
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index 86a3a8e773d..71b6304d428 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -615,17 +615,19 @@ Status FragmentMgr::exec_plan_fragment(const
TPipelineFragmentParams& params,
}
}
+// Stage 2. prepare finished. then get FE instruction to execute
Status FragmentMgr::start_query_execution(const PExecPlanFragmentStartRequest*
request) {
+ TUniqueId query_id;
+ query_id.__set_hi(request->query_id().hi());
+ query_id.__set_lo(request->query_id().lo());
std::shared_ptr<QueryContext> q_ctx = nullptr;
{
std::lock_guard<std::mutex> lock(_lock);
- TUniqueId query_id;
- query_id.__set_hi(request->query_id().hi());
- query_id.__set_lo(request->query_id().lo());
q_ctx = _get_or_erase_query_ctx(query_id);
}
if (q_ctx) {
q_ctx->set_ready_to_execute(Status::OK());
+ LOG_INFO("Query {} start execution", print_id(query_id));
} else {
return Status::InternalError(
"Failed to get query fragments context. Query may be "
@@ -645,6 +647,7 @@ void FragmentMgr::remove_pipeline_context(
.count();
g_fragment_executing_count << -1;
g_fragment_last_active_time.set_value(now);
+ // this log will show when a query is really finished in BEs
LOG_INFO("Removing query {} fragment {}", print_id(query_id),
f_context->get_fragment_id());
_pipeline_map.erase({query_id, f_context->get_fragment_id()});
}
@@ -699,6 +702,7 @@ Status FragmentMgr::_get_query_ctx(const Params& params,
TUniqueId query_id, boo
return Status::OK();
}
+ // First time a fragment of a query arrived. print logs.
LOG(INFO) << "query_id: " << print_id(query_id) << ", coord_addr: " <<
params.coord
<< ", total fragment num on current host: " <<
params.fragment_num_on_host
<< ", fe process uuid: " <<
params.query_options.fe_process_uuid
@@ -727,7 +731,6 @@ Status FragmentMgr::_get_query_ctx(const Params& params,
TUniqueId query_id, boo
}
_set_scan_concurrency(params, query_ctx.get());
- const bool is_pipeline = std::is_same_v<TPipelineFragmentParams,
Params>;
if (params.__isset.workload_groups && !params.workload_groups.empty())
{
uint64_t tg_id = params.workload_groups[0].id;
@@ -738,21 +741,14 @@ Status FragmentMgr::_get_query_ctx(const Params& params,
TUniqueId query_id, boo
RETURN_IF_ERROR(query_ctx->set_workload_group(workload_group_ptr));
_exec_env->runtime_query_statistics_mgr()->set_workload_group_id(print_id(query_id),
tg_id);
-
- LOG(INFO) << "Query/load id: " <<
print_id(query_ctx->query_id())
- << ", use workload group: " <<
workload_group_ptr->debug_string()
- << ", is pipeline: " << ((int)is_pipeline);
} else {
- LOG(INFO) << "Query/load id: " <<
print_id(query_ctx->query_id())
- << " carried group info but can not find group in
be";
+ LOG(WARNING) << "Query/load id: " <<
print_id(query_ctx->query_id())
+ << "can't find its workload group " << tg_id;
}
}
// There is some logic in query ctx's dctor, we could not check if
exists and delete the
// temp query ctx now. For example, the query id maybe removed from
workload group's queryset.
_query_ctx_map.insert(std::make_pair(query_ctx->query_id(),
query_ctx));
- LOG(INFO) << "Register query/load memory tracker, query/load id: "
- << print_id(query_ctx->query_id())
- << " limit: " <<
PrettyPrinter::print(query_ctx->mem_limit(), TUnit::BYTES);
}
return Status::OK();
}
@@ -1185,8 +1181,6 @@ Status FragmentMgr::apply_filterv2(const
PPublishFilterRequestV2* request,
auto iter = _pipeline_map.find(
{UniqueId(request->query_id()).to_thrift(),
fragment_id});
if (iter == _pipeline_map.end()) {
- LOG(WARNING) << "No pipeline fragment is found: Query-ID =
"
- << request->query_id() << " fragment_id = "
<< fragment_id;
continue;
}
pip_context = iter->second;
diff --git a/be/src/runtime/query_context.cpp b/be/src/runtime/query_context.cpp
index 354de1426e9..ece4c76a17a 100644
--- a/be/src/runtime/query_context.cpp
+++ b/be/src/runtime/query_context.cpp
@@ -26,11 +26,9 @@
#include <exception>
#include <memory>
#include <mutex>
-#include <sstream>
#include <utility>
#include "common/logging.h"
-#include "olap/olap_common.h"
#include "pipeline/dependency.h"
#include "pipeline/pipeline_fragment_context.h"
#include "runtime/exec_env.h"
@@ -198,7 +196,8 @@ QueryContext::~QueryContext() {
_exec_env->spill_stream_mgr()->async_cleanup_query(_query_id);
DorisMetrics::instance()->query_ctx_cnt->increment(-1);
- LOG_INFO("Query {} deconstructed, {}", print_id(this->_query_id),
mem_tracker_msg);
+ // the only one msg shows query's end. any other msg should append to it
if need.
+ LOG_INFO("Query {} deconstructed, mem_tracker: {}",
print_id(this->_query_id), mem_tracker_msg);
}
void QueryContext::set_ready_to_execute(Status reason) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
