github-actions[bot] commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3842218818
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2246,8 +2261,9 @@ bool PipelineFragmentContext::_close_fragment_instance() {
}
Defer defer_op {[&]() { _is_fragment_instance_closed = true; }};
_fragment_level_profile->total_time_counter()->update(_fragment_watcher.elapsed_time());
- if (!_need_notify_close) {
- auto st = send_report(true);
+ if (!_need_notify_close && !_final_report_submitted) {
+ _final_report_submitted = true;
+ auto st = send_report(true, _finish_load_fragment());
Review Comment:
[P1] Do not count intermediate recursive generations as finished
`WAIT_FOR_DESTROY` calls `notify_close()` without arming
`_send_report_on_close`. If that RPC arrives before the old target's last task
closes, it clears `_need_notify_close`; the last task then enters this generic
branch and calls `_finish_load_fragment()` even though this is only an
intermediate recursion round. `REBUILD` creates a fresh PFC/one-shot flag with
the same QueryContext and `_num_instances`, so each disposable generation can
advance the counter again while FE's expected total still counts the original
local instances once. This can mark the query finished before the recursive
source closes or overshoot exact equality and suppress the real final report.
Track terminal completion across the logical recursive fragment rather than
each rebuilt PFC, and test WAIT_FOR_DESTROY-before-last-task over multiple
rounds.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2290,6 +2306,20 @@ bool PipelineFragmentContext::_close_fragment_instance()
{
return !_need_notify_close;
}
+bool PipelineFragmentContext::_finish_load_fragment() {
+ // FINAL_CLOSE can also release a context whose preparation never
installed RuntimeState;
+ // cleanup in that state must not attempt to publish load statistics.
+ if (_runtime_state == nullptr) {
+ return false;
+ }
+ if (_runtime_state->query_type() != TQueryType::LOAD) {
+ return false;
+ }
+ // A fragment closes only after its tasks and sink writers close. Let only
the last group of
+ // local LOAD instances trigger the final snapshot before the coordinator
can start auditing.
+ return _query_ctx->finish_fragment(_num_instances);
Review Comment:
[P1] Count terminal PFCs when FINAL_CLOSE never arrives
This helper only runs from the two report-submission branches. A recursive
target can finish all tasks while `_need_notify_close` is still true; if
another fragment cancels the query before `FINAL_CLOSE`,
`cancel()->notify_close()` removes that already-closed PFC with
`_send_report_on_close == false`, so its `_num_instances` never advances the
exact `_fragment_num_on_host` gate. Initial or recursive-`REBUILD` prepare
failure has the same accounting hole because cancellation returns before the
planned PFC can reach this helper. The remaining PFCs can then never hit
equality, so the direct final-statistics report is suppressed and audit falls
back to the periodic/destructor race. Make every terminal LOAD-PFC outcome
contribute its planned group exactly once, and test cancellation before
FINAL_CLOSE plus initial and rebuilt PFC prepare failure.
##########
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:
[P1] Isolate the audit RPC from the shared fragment pool
Reordering protects this callback's own coordinator report, but this
best-effort call still runs synchronously on `FragmentMgrAsyncWorkThreadPool`
while `send_report()`'s captured `ctx` retains the full PFC, QueryContext,
RuntimeStates, and their memory. The helper uses the general 60-second Thrift
timeout and can reopen/retry, so an unavailable audit FE creates one blocked
worker per completion. This pool grows to 2048 workers, queues 4096, and is
also used for later final callbacks and parallel pipeline preparation; under an
outage those later callbacks are delayed or rejected before their healthy
coordinators receive essential terminal status. Put statistics on a dedicated
bounded path with a deadline below the audit budget, retaining registry state
for periodic reconciliation, and test blocked audit destinations followed by a
healthy query.
##########
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:
[P1] Include acknowledged S1 in the audit snapshot
The statistics-only RPC returns OK as soon as this live map is updated, but
the audit daemon rebuilds its immutable snapshot before it selects expired
events. If an event is already over the five-second threshold, the daemon can
publish S0, this compute can then accept final S1 and acknowledge the BE, which
unregisters its retry state, and the same daemon cycle can remove the event and
fill it from the already-published S0. S1 survives in the live map but that
audit is never retried. Establish an ordering boundary so a final update
acknowledged before an event is selected is reflected in the snapshot used for
that event, and add a latch-controlled snapshot/update/dequeue test.
##########
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) -> {
+ BeReportInfo beReportInfo = previousInfo == null
+ ? new BeReportInfo(currentTime) : previousInfo;
beReportInfo.beLastReportTime = currentTime;
+ for (Map.Entry<String, TQueryStatisticsResult> entry
+ : params.query_statistics_result_map.entrySet()) {
+ 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);
+ }
+ return beReportInfo;
+ });
+ }
+
+ private boolean shouldReplaceQueryStatistics(TQueryStatisticsResult
previous,
+ TQueryStatisticsResult incoming) {
+ long previousGeneration = previous.isSetQueryStatisticsGeneration()
+ ? previous.getQueryStatisticsGeneration() : 0;
+ long incomingGeneration = incoming.isSetQueryStatisticsGeneration()
+ ? incoming.getQueryStatisticsGeneration() : 0;
+ if (previousGeneration != incomingGeneration) {
+ return incomingGeneration > previousGeneration;
}
- for (Map.Entry<String, TQueryStatisticsResult> entry :
params.query_statistics_result_map.entrySet()) {
- beReportInfo.queryStatsMap.put(entry.getKey(),
Pair.of(currentTime, entry.getValue()));
+
+ long previousSequence = previous.isSetQueryStatisticsSequence()
+ ? previous.getQueryStatisticsSequence() : 0;
+ long incomingSequence = incoming.isSetQueryStatisticsSequence()
+ ? incoming.getQueryStatisticsSequence() : 0;
+ if (previousSequence != incomingSequence) {
+ return incomingSequence > previousSequence;
}
+
+ // Legacy BEs have no ordering fields; once their terminal snapshot
arrives, an older
+ // in-flight periodic snapshot must not make the same query look
unfinished again.
+ return !previous.isSetQueryFinished() || !previous.isQueryFinished()
+ || (incoming.isSetQueryFinished() &&
incoming.isQueryFinished());
}
- private void clearReportTimeoutBeStatistics() {
+ void clearReportTimeoutBeStatistics() {
// 1 clear report timeout be
Set<Long> currentBeIdSet = beToQueryStatsMap.keySet();
Long currentTime = System.currentTimeMillis();
for (Long beId : currentBeIdSet) {
- BeReportInfo beReportInfo = beToQueryStatsMap.get(beId);
- if (currentTime - beReportInfo.beLastReportTime >
Config.be_report_query_statistics_timeout_ms) {
- beToQueryStatsMap.remove(beId);
- continue;
- }
- Set<String> queryIdSet = beReportInfo.queryStatsMap.keySet();
- for (String queryId : queryIdSet) {
- Pair<Long, TQueryStatisticsResult> pair =
beReportInfo.queryStatsMap.get(queryId);
- long queryLastReportTime = pair.first;
- boolean timeout = currentTime - queryLastReportTime
- > Config.be_report_query_statistics_timeout_ms;
- // Remove query statistics only when both conditions are
satisfied:
- // 1) this query statistics is timeout, and
- // 2) FE no longer has this query in QeProcessorImpl.
- // Example timeline:
- // - t0: query q1 is still running, but one periodic BE report
is delayed for > timeout.
- // - t1: clear thread runs. timeout condition is true, but q1
still exists in FE.
- // - t2: we keep q1 statistics instead of removing it; later
reports can update it again.
- if (timeout && isQueryNotExistInFe(queryId)) {
- beReportInfo.queryStatsMap.remove(queryId);
+ beToQueryStatsMap.computeIfPresent(beId, (ignored, beReportInfo)
-> {
Review Comment:
[P2] Keep the per-BE compute bounded
`updateBeQueryStats()` holds this outer backend-key compute while processing
every query in the incoming map, and timeout cleanup uses the same key while
traversing the full 60-second retained set and doing FE ownership lookups. A
latency-sensitive one-query final S1 is therefore serialized behind an
unbounded periodic bulk report from that BE; the audit daemon can publish S0
and expire the event while S1 is still waiting, even though the destination FE
is healthy. Keep top-level entry creation/removal short and make update/expiry
conditional per query, or compare/remove the exact observed value, with a
high-cardinality periodic-batch/cleanup versus final-update test.
--
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]
