Gabriel39 commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3841718859
##########
be/src/runtime/query_context.cpp:
##########
@@ -248,6 +248,13 @@ QueryContext::~QueryContext() {
PrettyPrinter::print_bytes(query_mem_tracker()->peak_consumption()));
}
_resource_ctx->task_controller()->finish();
+#ifndef BE_TEST
+ if (_resource_ctx->task_controller()->query_type() == TQueryType::LOAD) {
Review Comment:
Fixed in aed8597f4d. FINAL_CLOSE now only arms the final report under the
task mutex. notify_close() or the last task close submits it exactly once after
all target tasks are closed. The new FINAL_CLOSE_WAITS_FOR_LAST_RECURSIVE_TASK
BE test verifies that the query is not marked finished early.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2604,18 +2619,23 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
LOG_INFO("Going to cancel query {} since report exec status got rpc
failed: {}",
print_id(req.query_id), rpc_status.to_string());
Review Comment:
Fixed in aed8597f4d. Both direct and periodic builders now reserve the
sequence before observing completion or mutable counters, making the sequence
the snapshot linearization point. A deterministic SyncPoint test pauses an
older snapshot, publishes a newer final snapshot, and verifies the older
snapshot retains the lower sequence.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -186,20 +186,40 @@ public void
updateBeQueryStats(TReportWorkloadRuntimeStatusParams params) {
return;
}
long beId = params.backend_id;
- // NOTE(wb) one be sends update request one by one,
- // so there is no need a global lock for beToQueryStatsMap here,
- // just keep one be's put/remove/get is atomic operation is enough
long currentTime = System.currentTimeMillis();
- BeReportInfo beReportInfo = beToQueryStatsMap.get(beId);
- if (beReportInfo == null) {
- beReportInfo = new BeReportInfo(currentTime);
- beToQueryStatsMap.put(beId, beReportInfo);
- } else {
- beReportInfo.beLastReportTime = currentTime;
- }
+ BeReportInfo beReportInfo = beToQueryStatsMap.computeIfAbsent(beId,
+ ignored -> new BeReportInfo(currentTime));
+ beReportInfo.beLastReportTime = currentTime;
for (Map.Entry<String, TQueryStatisticsResult> entry :
params.query_statistics_result_map.entrySet()) {
- beReportInfo.queryStatsMap.put(entry.getKey(),
Pair.of(currentTime, entry.getValue()));
+ Pair<Long, TQueryStatisticsResult> incoming = Pair.of(currentTime,
entry.getValue());
+ beReportInfo.queryStatsMap.compute(entry.getKey(), (queryId,
previous) ->
+ previous == null ||
shouldReplaceQueryStatistics(previous.second, incoming.second)
+ ? incoming : previous);
+ }
+ }
+
+ private boolean shouldReplaceQueryStatistics(TQueryStatisticsResult
previous,
+ TQueryStatisticsResult incoming) {
+ long previousGeneration = previous.isSetQueryStatisticsGeneration()
Review Comment:
Per the clarified scope, rolling-upgrade and mixed-version compatibility are
not required for this change, so no legacy-FE negotiation path is added. The
current-version ordering and cleanup races are addressed in aed8597f4d.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2544,6 +2553,22 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
return;
}
+ if (req.report_query_statistics) {
+ // The client-facing FE owns the audit event and may differ from the
coordinator FE.
+ // Keep statistics in a separate RPC so they cannot invalidate the
terminal status payload.
+ Status statistics_status =
Review Comment:
Fixed in aed8597f4d. The essential coordinator ReportExecStatus RPC and
external-file ownership outcome are now completed before the best-effort
statistics RPC to the audit FE. A slow or unavailable audit FE can no longer
delay delivery of terminal status to the coordinator.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -186,20 +186,40 @@ public void
updateBeQueryStats(TReportWorkloadRuntimeStatusParams params) {
return;
}
long beId = params.backend_id;
- // NOTE(wb) one be sends update request one by one,
- // so there is no need a global lock for beToQueryStatsMap here,
- // just keep one be's put/remove/get is atomic operation is enough
long currentTime = System.currentTimeMillis();
- BeReportInfo beReportInfo = beToQueryStatsMap.get(beId);
- if (beReportInfo == null) {
- beReportInfo = new BeReportInfo(currentTime);
- beToQueryStatsMap.put(beId, beReportInfo);
- } else {
- beReportInfo.beLastReportTime = currentTime;
- }
+ BeReportInfo beReportInfo = beToQueryStatsMap.computeIfAbsent(beId,
Review Comment:
Fixed in aed8597f4d. Statistics updates and timeout cleanup now use atomic
ConcurrentMap.compute operations per BE. A latch-controlled FE test reproduces
the stale cleanup decision racing with a final update and verifies the
acknowledged final snapshot survives.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
