[fix](pipelinex) fix fragment instance progress reports (part 2) (#40694

) (#41641) backport #40694
apache · Oct 10, 2024 · 3120bfb · 3120bfb
1 parent 30492a2
commit 3120bfb
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 1 deletion.
diff --git 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
@@ -2818,7 +2818,7 @@ public void updateFragmentExecStatus(TReportExecStatusParams params) {
                 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
@@ -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