Gabriel39 commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3848929709
##########
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:
Fixed both bypasses. Forwarded execution now returns an immutable
auditBackendIds set through TMasterOpResult, and InsertIntoTVFCommand retains
its locally created Coordinator on StmtExecutor for the later audit. Added
forwarding accessor and TVF coordinator-retention tests.
##########
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:
Reworked FE storage around backend incarnations instead of replacing the
backend shell. A pending audit binds to an observed process and retains that
exact old incarnation until drain, while old data is excluded from current
snapshots and cannot satisfy a new query. Added
report-enqueue-restart-cleanup-drain and cross-FE heartbeat-catch-up tests.
--
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]
