[fix](pipelinex) fix fragment instance progress reports (part 2) (apa…

…che#40694)

This PR is a followup of apache#40325. Because PipelineX has deprecated the
old report.
This PR fixed the `ScannedRows` and `LoadBytes` in the progress of the
`SHOW LOAD` result.

Currently the progress will only be updated when a fragment instance is
done.
Timely progress updates will be supported in a later PR.

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);
                 }
             }

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());
                 }

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