You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2025/10/11 03:15:15 UTC
(doris) branch branch-3.1 updated: branch-3.1: [feature](info) Append RunningTasks into BE info #56277 (#56781)
This is an automated email from the ASF dual-hosted git repository.
morrysnow pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new d4cffec8729 branch-3.1: [feature](info) Append RunningTasks into BE info #56277 (#56781)
d4cffec8729 is described below
commit d4cffec8729add4e8a130eea575803c08e362fb8
Author: Gabriel <li...@selectdb.com>
AuthorDate: Sat Oct 11 11:15:09 2025 +0800
branch-3.1: [feature](info) Append RunningTasks into BE info #56277 (#56781)
picked from #56277
---
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/analysis/ShowBackendsStmtTest.java | 2 +-
.../org/apache/doris/utframe/DemoMultiBackendsTest.java | 8 ++++----
gensrc/thrift/MasterService.thrift | 2 ++
.../suites/nereids_syntax_p0/information_schema.groovy | 1 -
11 files changed, 41 insertions(+), 12 deletions(-)
diff --git a/be/src/agent/task_worker_pool.cpp b/be/src/agent/task_worker_pool.cpp
index 64c56072f95..8dcad387552 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -1061,6 +1061,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 9ba558be76d..95dbed9ae38 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -653,6 +653,12 @@ void FragmentMgr::remove_pipeline_context(
_pipeline_map.erase({query_id, f_context->get_fragment_id()});
}
+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()) {
@@ -891,7 +897,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 fb01c899104..5aafc929f37 100644
--- a/be/src/runtime/fragment_mgr.h
+++ b/be/src/runtime/fragment_mgr.h
@@ -134,6 +134,7 @@ public:
Status exec_plan_fragment(const TPipelineFragmentParams& params, const QuerySource query_type,
const FinishCallback& cb);
+ void remove_query_context(const TUniqueId& key);
Status start_query_execution(const PExecPlanFragmentStartRequest* request);
Status trigger_pipeline_context_report(const ReportStatusRequest,
diff --git a/be/src/runtime/query_context.cpp b/be/src/runtime/query_context.cpp
index c2a4cfca565..bcd99fec381 100644
--- a/be/src/runtime/query_context.cpp
+++ b/be/src/runtime/query_context.cpp
@@ -196,6 +196,8 @@ QueryContext::~QueryContext() {
_exec_env->spill_stream_mgr()->async_cleanup_query(_query_id);
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 e3db2a3a716..44f45b200fc 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")
- .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")
@@ -173,6 +173,9 @@ public class BackendsProcDir implements ProcDirInterface {
// memory
backendInfo.add(RuntimeProfile.printCounter(backend.getBeMemory(), TUnit.BYTES));
+
+ // 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 4d1017d0c47..53883e79da6 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 f0a7990c9bd..c42456acb63 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
@@ -106,6 +106,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.
@@ -959,6 +961,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/analysis/ShowBackendsStmtTest.java b/fe/fe-core/src/test/java/org/apache/doris/analysis/ShowBackendsStmtTest.java
index c46c6428b96..c14769489ec 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/analysis/ShowBackendsStmtTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/analysis/ShowBackendsStmtTest.java
@@ -71,7 +71,7 @@ public class ShowBackendsStmtTest {
public void getMetaData() {
ShowBackendsStmt stmt = new ShowBackendsStmt();
ShowResultSetMetaData result = stmt.getMetaData();
- Assertions.assertEquals(result.getColumnCount(), 27);
+ Assertions.assertEquals(result.getColumnCount(), 28);
result.getColumns().forEach(col -> Assertions.assertEquals(col.getType(), ScalarType.createVarchar(30)));
}
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 d3238328814..208fb2afaa1 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
@@ -200,13 +200,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() - 8));
+ result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 9));
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() - 5));
- Assert.assertEquals("0", result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 4));
- Assert.assertEquals(Tag.VALUE_MIX, result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 3));
+ 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));
}
private static void updateReplicaPathHash() {
diff --git a/gensrc/thrift/MasterService.thrift b/gensrc/thrift/MasterService.thrift
index 8c31a13e28e..d0794b70445 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 {
diff --git a/regression-test/suites/nereids_syntax_p0/information_schema.groovy b/regression-test/suites/nereids_syntax_p0/information_schema.groovy
index a86146b559e..5cd30bc89ce 100644
--- a/regression-test/suites/nereids_syntax_p0/information_schema.groovy
+++ b/regression-test/suites/nereids_syntax_p0/information_schema.groovy
@@ -18,7 +18,6 @@
suite("information_schema") {
List<List<Object>> table = sql """ select * from backends(); """
assertTrue(table.size() > 0)
- assertTrue(table[0].size == 25)
sql "SELECT DATABASE();"
sql "select USER();"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org