github-actions[bot] commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3845650634
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -347,7 +370,8 @@ void
RuntimeQueryStatisticsMgr::report_runtime_query_statistics() {
for (auto iter = _resource_contexts_map.begin(); iter !=
_resource_contexts_map.end();) {
std::string query_id = iter->first;
auto resource_ctx = iter->second;
- bool is_query_finished =
resource_ctx->task_controller()->is_finished();
+ TQueryStatisticsResult stats_result =
create_query_statistics_result(resource_ctx);
Review Comment:
[P2] Skip snapshot construction for excluded EXTERNAL tasks. This now calls
create_query_statistics_result() for every retained context while holding the
registry lock, but the following branch never reports active
TQueryType::EXTERNAL contexts and discards the serialized result. Previously
to_thrift_query_statistics() ran only inside the non-EXTERNAL branch. With many
long-lived Spark/Flink tasks this adds a full counter snapshot every three
seconds and stalls registrations for work that cannot be sent. Read only the
finished flag for the EXTERNAL erase decision and build the Thrift result after
that filter.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -252,34 +352,61 @@ public Map<String, TQueryStatistics>
getQueryStatisticsMap() {
}
// Build a merged map by traversing concurrent runtime structures.
- private Map<String, TQueryStatistics> buildQueryStatisticsMapUnsafe() {
- // 1 merge query stats in all be
- Set<Long> beIdSet = beToQueryStatsMap.keySet();
- Map<String, TQueryStatistics> resultQueryMap = Maps.newHashMap();
- for (Long beId : beIdSet) {
- BeReportInfo beReportInfo = beToQueryStatsMap.get(beId);
+ private RuntimeStatisticsSnapshot buildRuntimeStatisticsSnapshot() {
+ RuntimeStatisticsSnapshot snapshot = new RuntimeStatisticsSnapshot();
+ for (Map.Entry<Long, BeReportInfo> beEntry :
beToQueryStatsMap.entrySet()) {
+ long beId = beEntry.getKey();
+ BeReportInfo beReportInfo = beEntry.getValue();
if (beReportInfo == null) {
continue;
}
- Set<String> queryIdSet = beReportInfo.queryStatsMap.keySet();
- for (String queryId : queryIdSet) {
- Pair<Long, TQueryStatisticsResult> queryStatsPair =
- beReportInfo.queryStatsMap.get(queryId);
+ long currentBackendStartTime = getBackendStartTime(beId);
+ if (currentBackendStartTime > 0
+ && currentBackendStartTime !=
beReportInfo.backendStartTime) {
+ // Heartbeat can expose a restart before the new process sends
its first runtime
+ // report. Drop the old shell now so its final flag cannot
release a new audit.
+ beToQueryStatsMap.remove(beId, beReportInfo);
Review Comment:
[P1] Preserve old-incarnation snapshots for audits already queued. If
process A's final q1 report is stored and q1 is enqueued, then the BE restarts
before the five-second drain, this branch drops A's shell before merging it.
Process B cannot reproduce q1, so that audit waits for fallback and logs
without the accepted counters. Associate pending participants with the process
incarnation and retain per-incarnation data until those audits drain, while
preventing A from satisfying new B queries; add final-report -> enqueue ->
restart -> drain coverage.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/AuditLogHelper.java:
##########
@@ -408,7 +409,14 @@ private static void logAuditLogImpl(ConnectContext ctx,
String origStmt, Stateme
auditEventBuilder.setState(String.valueOf(MysqlStateType.OK));
}
AuditEvent event = auditEventBuilder.build();
-
Env.getCurrentEnv().getWorkloadRuntimeStatusMgr().submitFinishQueryToAudit(event);
+ Set<Long> expectedBackendIds = ImmutableSet.of();
+ if (!event.isQuery && ctx.getExecutor() != null &&
ctx.getExecutor().getCoord() != null) {
+ // Audit completion is query-scoped: keep the event pending until
every backend that
+ // received a fragment publishes its final cumulative statistics
snapshot.
+ expectedBackendIds =
ctx.getExecutor().getCoord().getInvolvedBackendIds();
+ }
+ Env.getCurrentEnv().getWorkloadRuntimeStatusMgr()
Review Comment:
[P1] Update the existing audit mock expectations.
AuditLogHelperBackendSelectionTest and StmtExecutorInternalQueryTest both
verify submitFinishQueryToAudit(AuditEvent), but this path now invokes only the
two-argument overload; a Mockito mock records no matching one-argument call.
Update both tests to capture the backend-ID argument and assert it is empty in
their no-coordinator cases, then add direct coordinator coverage.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -185,54 +211,128 @@ public void
updateBeQueryStats(TReportWorkloadRuntimeStatusParams params) {
LOG.warn("be report workload runtime status but without query
stats map");
return;
}
+ if (!params.isSetBackendStartTime()) {
+ LOG.warn("be report workload runtime status without backend start
time");
+ 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 backendStartTime = params.backend_start_time;
+ long currentBackendStartTime = getBackendStartTime(beId);
+ if (currentBackendStartTime > 0 && currentBackendStartTime !=
backendStartTime) {
Review Comment:
[P1] Do not acknowledge a report rejected before heartbeat catch-up. In a
forwarded DML, coordinator FE B can schedule a restarted BE after B sees its
new heartbeat while audit FE A still has the previous start time. The BE
reports to A via current_connect_fe; this branch drops the payload, but the
statistics-only RPC returns OK and BE erases the finished ResourceContext
without inspecting acceptance. Propagate an explicit acceptance/retry signal,
make BE inspect it before erasing, and cover the two-FE heartbeat-ordering case.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/AuditLogHelper.java:
##########
@@ -408,7 +409,14 @@ private static void logAuditLogImpl(ConnectContext ctx,
String origStmt, Stateme
auditEventBuilder.setState(String.valueOf(MysqlStateType.OK));
}
AuditEvent event = auditEventBuilder.build();
-
Env.getCurrentEnv().getWorkloadRuntimeStatusMgr().submitFinishQueryToAudit(event);
+ Set<Long> expectedBackendIds = ImmutableSet.of();
+ if (!event.isQuery && ctx.getExecutor() != null &&
ctx.getExecutor().getCoord() != null) {
Review Comment:
[P1] Capture participants outside StmtExecutor.coord. This condition leaves
the expected set empty for at least two distributed DML paths: a follower that
forwards to the master has only a MasterOpExecutor, and the proxy result
returns no backend IDs; InsertIntoTVFCommand owns a local Coordinator without
assigning executor.coord. Empty is treated as all-final after five seconds, so
a delayed periodic report still misses these audits. Publish the immutable
participant set from every coordinator owner, including through TMasterOpResult
for forwarding, and cover both paths.
--
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]
