This is an automated email from the ASF dual-hosted git repository.
gabriellee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 8470c2f4151 [feature](info) Append RunningTasks into BE info (#56277)
8470c2f4151 is described below
commit 8470c2f41511f84f8eb49da2cf69ddc6a502cbb8
Author: Gabriel <[email protected]>
AuthorDate: Tue Sep 23 12:29:32 2025 +0800
[feature](info) Append RunningTasks into BE info (#56277)
To help us determine whether any query/loading tasks are running,
`RunningTasks ` metrics is appended in this PR and users could retrieve
it by `show backends` command.
<img width="3800" height="668" alt="image"
src="https://github.com/user-attachments/assets/50124384-3427-4621-91b3-b165bfda8d93";
/>
---
be/src/agent/task_worker_pool.cpp | 1 +
be/src/runtime/fragment_mgr.cpp | 8 +++++++-
be/src/runtime/fragment_mgr.h | 1 +
be/src/runtime/query_context.cpp | 2 ++
.../java/org/apache/doris/common/proc/BackendsProcDir.java | 5 ++++-
.../main/java/org/apache/doris/master/ReportHandler.java | 13 +++++++++----
.../src/main/java/org/apache/doris/system/Backend.java | 10 ++++++++++
.../org/apache/doris/utframe/DemoMultiBackendsTest.java | 8 ++++----
gensrc/thrift/MasterService.thrift | 2 ++
9 files changed, 40 insertions(+), 10 deletions(-)
diff --git a/be/src/agent/task_worker_pool.cpp
b/be/src/agent/task_worker_pool.cpp
index 677a166d89f..e3ca1a4b473 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -1093,6 +1093,7 @@ void report_task_callback(const ClusterInfo*
cluster_info) {
}
}
request.__set_backend(BackendOptions::get_local_backend());
+
request.__set_running_tasks(ExecEnv::GetInstance()->fragment_mgr()->running_query_num());
bool succ = handle_report(request, cluster_info, "task");
report_task_total << 1;
if (!succ) [[unlikely]] {
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index a25334d670c..dc9c38ea084 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -662,6 +662,12 @@ void
FragmentMgr::remove_pipeline_context(std::pair<TUniqueId, int> key) {
_pipeline_map.erase(key);
}
+void FragmentMgr::remove_query_context(const TUniqueId& key) {
+#ifndef BE_TEST
+ _query_ctx_map.erase(key);
+#endif
+}
+
std::shared_ptr<QueryContext> FragmentMgr::get_query_ctx(const TUniqueId&
query_id) {
auto val = _query_ctx_map.find(query_id);
if (auto q_ctx = val.lock()) {
@@ -893,7 +899,7 @@ void FragmentMgr::cancel_query(const TUniqueId query_id,
const Status reason) {
}
}
query_ctx->cancel(reason);
- _query_ctx_map.erase(query_id);
+ remove_query_context(query_id);
LOG(INFO) << "Query " << print_id(query_id)
<< " is cancelled and removed. Reason: " << reason.to_string();
}
diff --git a/be/src/runtime/fragment_mgr.h b/be/src/runtime/fragment_mgr.h
index aca118dedaf..b0c1a3ad592 100644
--- a/be/src/runtime/fragment_mgr.h
+++ b/be/src/runtime/fragment_mgr.h
@@ -124,6 +124,7 @@ public:
const TPipelineFragmentParamsList& parent);
void remove_pipeline_context(std::pair<TUniqueId, int> key);
+ void remove_query_context(const TUniqueId& key);
Status exec_plan_fragment(const TPipelineFragmentParams& params, const
QuerySource query_type,
const FinishCallback& cb, const
TPipelineFragmentParamsList& parent);
diff --git a/be/src/runtime/query_context.cpp b/be/src/runtime/query_context.cpp
index a84e5483ad0..c19a27e2d39 100644
--- a/be/src/runtime/query_context.cpp
+++ b/be/src/runtime/query_context.cpp
@@ -234,6 +234,8 @@ QueryContext::~QueryContext() {
_merge_controller_handler.reset();
DorisMetrics::instance()->query_ctx_cnt->increment(-1);
+ // TODO(gabriel): we need to clear outdated query contexts on time
+ //
ExecEnv::GetInstance()->fragment_mgr()->remove_query_context(this->_query_id);
// 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);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/proc/BackendsProcDir.java
b/fe/fe-core/src/main/java/org/apache/doris/common/proc/BackendsProcDir.java
index b8eb8d1fcb2..92ef9f09e03 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/proc/BackendsProcDir.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/proc/BackendsProcDir.java
@@ -51,7 +51,7 @@ public class BackendsProcDir implements ProcDirInterface {
.add("DataUsedCapacity").add("TrashUsedCapacity").add("AvailCapacity").add("TotalCapacity").add("UsedPct")
.add("MaxDiskUsedPct").add("RemoteUsedCapacity").add("Tag").add("ErrMsg").add("Version").add("Status")
.add("HeartbeatFailureCounter").add("NodeRole").add("CpuCores").add("Memory").add("LiveSince")
- .build();
+ .add("RunningTasks").build();
public static final ImmutableList<String> DISK_TITLE_NAMES = new
ImmutableList.Builder<String>()
.add("BackendId").add("Host").add("RootPath").add("DirType").add("DiskState")
@@ -177,6 +177,9 @@ public class BackendsProcDir implements ProcDirInterface {
// liveSince
backendInfo.add(TimeUtils.longToTimeString(backend.getLiveSince()));
+ // runningFragments
+ backendInfo.add(String.valueOf(backend.getRunningTasks()));
+
comparableBackendInfos.add(backendInfo);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java
index f91913e5a89..536b1fc25a9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/master/ReportHandler.java
@@ -224,7 +224,8 @@ public class ReportHandler extends Daemon {
ReportTask reportTask = new ReportTask(beId, reportType, tasks, disks,
tablets, partitionsVersion,
reportVersion, request.getStoragePolicy(),
request.getResource(), request.getNumCores(),
- request.getPipelineExecutorSize(), numTablets,
request.getIndexPolicy());
+ request.getPipelineExecutorSize(), numTablets,
request.getIndexPolicy(),
+ request.isSetRunningTasks() ? request.getRunningTasks() : -1);
try {
putToQueue(reportTask);
} catch (Exception e) {
@@ -314,6 +315,7 @@ public class ReportHandler extends Daemon {
private List<TStorageResource> storageResources;
private int cpuCores;
private int pipelineExecutorSize;
+ private long runningTasks;
private long numTablets;
private List<TIndexPolicy> indexPolicys;
@@ -321,7 +323,7 @@ public class ReportHandler extends Daemon {
Map<String, TDisk> disks, Map<Long, TTablet> tablets,
Map<Long, Long> partitionsVersion, long reportVersion,
List<TStoragePolicy> storagePolicies, List<TStorageResource>
storageResources, int cpuCores,
- int pipelineExecutorSize, long numTablets, List<TIndexPolicy>
indexPolicys) {
+ int pipelineExecutorSize, long numTablets, List<TIndexPolicy>
indexPolicys, long runningTasks) {
this.beId = beId;
this.reportType = reportType;
this.tasks = tasks;
@@ -335,12 +337,13 @@ public class ReportHandler extends Daemon {
this.pipelineExecutorSize = pipelineExecutorSize;
this.numTablets = numTablets;
this.indexPolicys = indexPolicys;
+ this.runningTasks = runningTasks;
}
@Override
protected void exec() {
if (tasks != null) {
- ReportHandler.taskReport(beId, tasks);
+ ReportHandler.taskReport(beId, tasks, runningTasks);
}
if (disks != null) {
ReportHandler.diskReport(beId, disks);
@@ -704,7 +707,8 @@ public class ReportHandler extends Daemon {
}
}
- private static void taskReport(long backendId, Map<TTaskType, Set<Long>>
runningTasks) {
+ private static void taskReport(long backendId, Map<TTaskType, Set<Long>>
runningTasks,
+ long numRunningTasks) {
debugBlock();
if (LOG.isDebugEnabled()) {
LOG.debug("begin to handle task report from backend {}",
backendId);
@@ -728,6 +732,7 @@ public class ReportHandler extends Daemon {
? runningTasks.get(TTaskType.PUBLISH_VERSION).size() : 0;
if (be != null) {
be.setPublishTaskLastTimeAccumulated((long) publishTaskSize);
+ be.setRunningTasks(numRunningTasks);
}
List<AgentTask> diffTasks = AgentTaskQueue.getDiffTasks(backendId,
runningTasks);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java
b/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java
index de438f676bd..0f3ac2e2226 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java
@@ -103,6 +103,8 @@ public class Backend implements Writable {
private Long lastPublishTaskAccumulatedNum = 0L;
+ private Long runningTasks = 0L;
+
private String heartbeatErrMsg = "";
// This is used for the first time we init pathHashToDishInfo in
SystemInfoService.
@@ -961,6 +963,14 @@ public class Backend implements Writable {
return disksRef.size();
}
+ public Long getRunningTasks() {
+ return runningTasks;
+ }
+
+ public void setRunningTasks(Long runningTasks) {
+ this.runningTasks = runningTasks;
+ }
+
/**
* Note: This class must be a POJO in order to display in JSON format
* Add additional information in the class to show in `show backends`
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/utframe/DemoMultiBackendsTest.java
b/fe/fe-core/src/test/java/org/apache/doris/utframe/DemoMultiBackendsTest.java
index b69bbcb9b85..62c7d140bf1 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/utframe/DemoMultiBackendsTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/utframe/DemoMultiBackendsTest.java
@@ -209,13 +209,13 @@ public class DemoMultiBackendsTest {
ProcResult result = dir.fetchResult();
Assert.assertEquals(BackendsProcDir.TITLE_NAMES.size(),
result.getColumnNames().size());
Assert.assertEquals("{\"location\" : \"default\"}",
- result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size()
- 9));
+ result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size()
- 10));
Assert.assertEquals(
"{\"lastSuccessReportTabletsTime\":\"N/A\",\"lastStreamLoadTime\":-1,\"isQueryDisabled\":false,"
+
"\"isLoadDisabled\":false,\"isActive\":true,\"currentFragmentNum\":0,\"lastFragmentUpdateTime\":0}",
- result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size()
- 6));
- Assert.assertEquals("0",
result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 5));
- Assert.assertEquals(Tag.VALUE_MIX,
result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 4));
+ result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size()
- 7));
+ Assert.assertEquals("0",
result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 6));
+ Assert.assertEquals(Tag.VALUE_MIX,
result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 5));
}
protected void alterTable(String sql, ConnectContext connectContext)
throws Exception {
diff --git a/gensrc/thrift/MasterService.thrift
b/gensrc/thrift/MasterService.thrift
index 50693a789e8..2caf3c76965 100644
--- a/gensrc/thrift/MasterService.thrift
+++ b/gensrc/thrift/MasterService.thrift
@@ -121,6 +121,8 @@ struct TReportRequest {
// tablet num in be, in cloud num_tablets may not eq tablet_list.size()
14: optional i64 num_tablets
15: optional list<AgentService.TIndexPolicy> index_policy
+ // Running query/loading tasks
+ 16: optional i64 running_tasks
}
struct TMasterResult {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
