Gabriel39 commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3843859692
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2604,18 +2633,36 @@ 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());
req.cancel_fn(rpc_status);
- } else if (req.done && req.status.ok()) {
+ }
+
+ if (rpc_status.ok() && req.done && req.status.ok()) {
// Files remain rollback-owned until the coordinator has acknowledged
the final metadata report.
req.runtime_state->finalize_external_file_report_cleanup(
ExternalFileReportOutcome::ACKNOWLEDGED);
- } else if (req.done) {
+ } else if (rpc_status.ok() && req.done) {
// An acknowledged error report confirms that FE will not publish this
write's files.
req.runtime_state->finalize_external_file_report_cleanup(
ExternalFileReportOutcome::REJECTED);
}
+
+ if (rpc_status.ok() && req.report_query_statistics) {
+ // Essential completion and external-file ownership must reach the
coordinator before a
+ // slow audit FE can block this best-effort statistics RPC.
+ Status statistics_status =
Review Comment:
Fixed in e81012f040 by deleting the second audit RPC rather than moving it
to another executor. The immutable cumulative snapshot is embedded in the
existing essential ReportExecStatus request, so there is no additional blocked
worker, retry loop, PFC retention, or audit-destination queue. The completion
RPC already owns the PFC lifetime required for terminal status delivery.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -186,53 +181,75 @@ 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 {
+ // Serialize updates and expiry only per BE. A final update
acknowledged here must not be
+ // removed by a cleaner that made its timeout decision on an older
value.
+ beToQueryStatsMap.compute(beId, (ignored, previousInfo) -> {
Review Comment:
Fixed in e81012f040. FE applies embedded workload statistics before
updateFragmentExecStatus can expose completion to the coordinator. Audit
dequeue now hydrates each event directly from the live per-BE maps instead of
the previously published immutable snapshot. The regression test publishes S0,
applies S1, enqueues the audit event, and verifies that dequeue emits S1.
--
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]
