This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.1 by this push:
new 3120bfb6e37 [fix](pipelinex) fix fragment instance progress reports
(part 2) (#40694) (#41641)
3120bfb6e37 is described below
commit 3120bfb6e37dbadf1b154668ad7c874868fe8604
Author: Kaijie Chen <c...@apache.org>
AuthorDate: Thu Oct 10 17:49:41 2024 +0800
[fix](pipelinex) fix fragment instance progress reports (part 2) (#40694)
(#41641)
backport #40694
---
be/src/runtime/fragment_mgr.cpp | 4 ++++
fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java | 2 +-
gensrc/thrift/FrontendService.thrift | 2 ++
3 files changed, 7 insertions(+), 1 deletion(-)
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index e5cd7c7cb8d..acf622b4196 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -414,6 +414,8 @@ void FragmentMgr::coordinator_callback(const
ReportStatusRequest& req) {
TFragmentInstanceReport t;
t.__set_fragment_instance_id(req.runtime_state->fragment_instance_id());
t.__set_num_finished_range(req.runtime_state->num_finished_range());
+ t.__set_loaded_rows(req.runtime_state->num_rows_load_total());
+ t.__set_loaded_bytes(req.runtime_state->num_bytes_load_total());
params.fragment_instance_reports.push_back(t);
} else if (!req.runtime_states.empty()) {
for (auto* rs : req.runtime_states) {
@@ -427,6 +429,8 @@ void FragmentMgr::coordinator_callback(const
ReportStatusRequest& req) {
TFragmentInstanceReport t;
t.__set_fragment_instance_id(rs->fragment_instance_id());
t.__set_num_finished_range(rs->num_finished_range());
+ t.__set_loaded_rows(rs->num_rows_load_total());
+ t.__set_loaded_bytes(rs->num_bytes_load_total());
params.fragment_instance_reports.push_back(t);
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
index 28c45608c5f..b0d1cbaba05 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
@@ -2818,7 +2818,7 @@ public class Coordinator implements CoordInterface {
for (TFragmentInstanceReport report :
params.getFragmentInstanceReports()) {
Env.getCurrentEnv().getLoadManager().updateJobProgress(
jobId, params.getBackendId(), params.getQueryId(),
report.getFragmentInstanceId(),
- params.getLoadedRows(), params.getLoadedBytes(),
params.isDone());
+ report.getLoadedRows(), report.getLoadedBytes(),
params.isDone());
Env.getCurrentEnv().getProgressManager().updateProgress(String.valueOf(jobId),
params.getQueryId(),
report.getFragmentInstanceId(), report.getNumFinishedRange());
}
diff --git a/gensrc/thrift/FrontendService.thrift
b/gensrc/thrift/FrontendService.thrift
index c53f25f7660..2258fff8fdf 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -420,6 +420,8 @@ struct TReportWorkloadRuntimeStatusParams {
struct TFragmentInstanceReport {
1: optional Types.TUniqueId fragment_instance_id;
2: optional i32 num_finished_range;
+ 3: optional i64 loaded_rows
+ 4: optional i64 loaded_bytes
}
// The results of an INSERT query, sent to the coordinator as part of
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
