This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 63cdc3179f4fd10622b5838f35c69f5d5cf94269
Author: Kaijie Chen <c...@apache.org>
AuthorDate: Wed Sep 25 16:37:46 2024 +0800
[fix](pipelinex) fix fragment instance progress reports (part 2) (#40694)
## Proposed changes
This PR is a followup of #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 | 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 6c61f9fa656..353489b23f8 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -399,6 +399,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) {
@@ -412,6 +414,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 f921710c3ca..701926a0889 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
@@ -2457,7 +2457,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 64b7c07fdc6..9e439b655f2 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -437,6 +437,8 @@ struct TQueryProfile {
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
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
