github-actions[bot] commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3859104078
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -143,96 +244,455 @@ public void submitFinishQueryToAudit(AuditEvent event) {
queryAuditEventList.size(), event.queryId);
}
Env.getCurrentAuditEventProcessor().handleAuditEvent(event);
+ if (transfersInFlightOwnership) {
+ releaseAuditStatisticsOwnership(statisticsQueryId);
+ }
} else {
// put the event to queryAuditEventList and let the worker
thread to handle it.
// the worker thread will try best to wait for the statistic
info before logging this event.
event.pushToAuditLogQueueTime = System.currentTimeMillis();
- queryAuditEventList.add(event);
+ PendingAuditEvent pending = new PendingAuditEvent(
+ event, expectedBackendIds, statisticsQueryId);
+ // Replace the execution owner with participant-scoped owners
under one lock so an
+ // accepted final snapshot cannot disappear between
coordinator teardown and audit.
+ registerPendingAuditEvent(pending, transfersInFlightOwnership);
+ queryAuditEventList.add(pending);
}
} finally {
queryAuditEventLogWriteUnlock();
}
}
- private List<AuditEvent> getQueryNeedAudit() {
- List<AuditEvent> ret = new ArrayList<>();
+ @VisibleForTesting
+ List<AuditEvent> getQueryNeedAudit() {
+ RuntimeStatisticsSnapshot runtimeSnapshot =
buildRuntimeStatisticsSnapshot();
+ queryStatisticsSnapshot =
ImmutableMap.copyOf(runtimeSnapshot.queryStatistics);
+
long currentTime = System.currentTimeMillis();
+ int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
+ long maximumWaitMs = Math.max(queryAuditLogTimeout,
+ Config.be_report_query_statistics_timeout_ms);
+ List<PendingAuditEvent> dueEvents = new ArrayList<>();
+ // Keep only identity/timestamp traversal under the producer-facing
queue lock; participant
+ // lookup and statistics merging may touch every expected backend and
must run lock-free.
queryAuditEventLogWriteLock();
try {
- int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
- Iterator<AuditEvent> iter = queryAuditEventList.iterator();
- while (iter.hasNext()) {
- AuditEvent ae = iter.next();
- if (currentTime - ae.pushToAuditLogQueueTime >
queryAuditLogTimeout) {
- ret.add(ae);
- iter.remove();
- } else {
- break;
+ for (PendingAuditEvent pending : queryAuditEventList) {
+ long elapsed = currentTime -
pending.event.pushToAuditLogQueueTime;
+ if (elapsed <= queryAuditLogTimeout) {
+ // Wall-clock corrections can make insertion order
disagree with deadlines.
+ continue;
}
+ dueEvents.add(pending);
}
} finally {
queryAuditEventLogWriteUnlock();
}
+
+ Map<PendingAuditEvent, TQueryStatistics> readyEvents = new
IdentityHashMap<>();
+ for (PendingAuditEvent pending : dueEvents) {
+ beforeHydrateAuditEvent(pending.event.queryId);
+ boolean allFinalSnapshotsReceived = true;
+ TQueryStatistics auditStatistics =
pending.expectedBackendIds.length == 0
+ ?
runtimeSnapshot.queryStatistics.get(pending.statisticsQueryId)
+ : new TQueryStatistics();
+ for (int i = 0; i < pending.expectedBackendIds.length; i++) {
+ bindParticipantFromExistingReport(pending, i);
+ long backendStartTime =
pending.expectedBackendStartTimes.get(i);
+ TQueryStatisticsResult statistics = findStatisticsForBackend(
+ pending.statisticsQueryId,
pending.expectedBackendIds[i], backendStartTime);
+ if (statistics == null) {
+ allFinalSnapshotsReceived = false;
+ continue;
+ }
+ mergeQueryStatistics(auditStatistics, statistics);
+ if (!statistics.isSetQueryFinished() ||
!statistics.isQueryFinished()) {
+ allFinalSnapshotsReceived = false;
+ }
+ }
+ long elapsed = currentTime - pending.event.pushToAuditLogQueueTime;
+ if (allFinalSnapshotsReceived || elapsed > maximumWaitMs) {
+ readyEvents.put(pending, auditStatistics);
+ }
+ }
+
+ List<PendingAuditEvent> removedEvents = new ArrayList<>();
+ queryAuditEventLogWriteLock();
+ try {
+ for (PendingAuditEvent pending : dueEvents) {
+ if (readyEvents.containsKey(pending) &&
queryAuditEventList.remove(pending)) {
+ removedEvents.add(pending);
+ }
+ }
+ } finally {
+ queryAuditEventLogWriteUnlock();
+ }
+
+ List<AuditEvent> ret = new ArrayList<>(removedEvents.size());
+ for (PendingAuditEvent pending : removedEvents) {
+ // A DML audit must not pass the queue until every scheduled BE
has published its
+ // query-level final snapshot. The upper bound only protects the
queue if a BE dies.
+ applyQueryStatisticsToAuditEvent(pending.event,
readyEvents.get(pending));
+ ret.add(pending.event);
+ unregisterPendingAuditEvent(pending);
+ }
return ret;
}
- public void updateBeQueryStats(TReportWorkloadRuntimeStatusParams params) {
+ public boolean updateBeQueryStats(TReportWorkloadRuntimeStatusParams
params) {
if (!params.isSetBackendId()) {
LOG.warn("be report workload runtime status but without beid");
- return;
+ return false;
}
if (!params.isSetQueryStatisticsResultMap()) {
LOG.warn("be report workload runtime status but without query
stats map");
- return;
+ return false;
+ }
+ if (!params.isSetBackendStartTime()) {
+ LOG.warn("be report workload runtime status without backend start
time");
+ return false;
}
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);
+ long backendStartTime = params.backend_start_time;
+ long currentBackendStartTime = getBackendStartTime(beId);
+ if (currentBackendStartTime > 0 && currentBackendStartTime !=
backendStartTime) {
+ LOG.info("ignore stale workload runtime status from backend {},
report start time {}, current {}",
+ beId, backendStartTime, currentBackendStartTime);
+ return false;
+ }
+ // Newly replayed backends use -1 as well as 0 for an unknown
heartbeat epoch. In either
+ // state the first accepted report must latch the only process allowed
to publish data.
+ if (currentBackendStartTime <= 0) {
+ Long acceptedStartTime =
lastAcceptedBackendStartTimes.putIfAbsent(beId, backendStartTime);
+ if (acceptedStartTime != null && acceptedStartTime !=
backendStartTime) {
+ return false;
+ }
} else {
- beReportInfo.beLastReportTime = currentTime;
+ lastAcceptedBackendStartTimes.put(beId, backendStartTime);
+ }
+ BackendIncarnation incarnation = new BackendIncarnation(beId,
backendStartTime);
+ beToQueryStatsMap.computeIfAbsent(incarnation, ignored -> new
BeReportInfo(backendStartTime));
+ long currentTime = System.currentTimeMillis();
+ for (Map.Entry<String, TQueryStatisticsResult> entry
+ : params.query_statistics_result_map.entrySet()) {
+ beforeUpdateQueryStatistics(entry.getKey());
+ while (true) {
+ BeReportInfo beReportInfo =
beToQueryStatsMap.computeIfAbsent(incarnation,
+ ignored -> new BeReportInfo(backendStartTime));
+ beReportInfo.lifecycleLock.readLock().lock();
+ try {
+ // Cleanup removes only an empty shell under the write
lock. If this insertion
+ // raced with removal, retry the single entry against the
replacement shell.
+ if (beToQueryStatsMap.get(incarnation) != beReportInfo) {
+ continue;
+ }
+ 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);
+ break;
+ } finally {
+ beReportInfo.lifecycleLock.readLock().unlock();
+ }
+ }
+ }
+ // Heartbeat can change while a large report is being merged. Refuse
the acknowledgement
+ // so the BE retries against the FE state that is authoritative after
the merge.
+ long latestBackendStartTime = getBackendStartTime(beId);
+ boolean accepted = latestBackendStartTime <= 0 ||
latestBackendStartTime == backendStartTime;
+ if (accepted) {
+ beforeBindPendingAuditEvents();
+ bindPendingAuditEvents(beId, backendStartTime,
+ params.query_statistics_result_map.keySet());
+ }
+ return accepted;
+ }
+
+ private TQueryStatisticsResult findStatisticsForBackend(String queryId,
+ long backendId, long backendStartTime) {
+ if (backendStartTime <= 0) {
+ return null;
+ }
+ BeReportInfo reportInfo = beToQueryStatsMap.get(
+ new BackendIncarnation(backendId, backendStartTime));
+ if (reportInfo != null) {
+ Pair<Long, TQueryStatisticsResult> pair =
reportInfo.queryStatsMap.get(queryId);
+ return pair == null ? null : pair.second;
+ }
+ return null;
+ }
+
+ public void beginAuditStatisticsOwnership(String queryId) {
+ if (queryId == null) {
+ return;
+ }
+ pendingAuditBindingLock.lock();
+ try {
+ inFlightAuditQueryReferences.compute(queryId,
+ (ignored, count) -> count == null ? 1 : count + 1);
+ } finally {
+ pendingAuditBindingLock.unlock();
+ }
+ }
+
+ public void transferAuditStatisticsOwnership(String previousQueryId,
String nextQueryId) {
+ if (previousQueryId == null || nextQueryId == null ||
previousQueryId.equals(nextQueryId)) {
+ return;
+ }
+ pendingAuditBindingLock.lock();
+ try {
+ releaseAuditStatisticsOwnershipLocked(previousQueryId);
+ inFlightAuditQueryReferences.compute(nextQueryId,
+ (ignored, count) -> count == null ? 1 : count + 1);
+ } finally {
+ pendingAuditBindingLock.unlock();
}
- for (Map.Entry<String, TQueryStatisticsResult> entry :
params.query_statistics_result_map.entrySet()) {
- beReportInfo.queryStatsMap.put(entry.getKey(),
Pair.of(currentTime, entry.getValue()));
+ }
+
+ public void releaseAuditStatisticsOwnership(String queryId) {
+ if (queryId == null) {
+ return;
+ }
+ pendingAuditBindingLock.lock();
+ try {
+ releaseAuditStatisticsOwnershipLocked(queryId);
+ } finally {
+ pendingAuditBindingLock.unlock();
+ }
+ }
+
+ private void releaseAuditStatisticsOwnershipLocked(String queryId) {
+ inFlightAuditQueryReferences.computeIfPresent(queryId,
+ (ignored, count) -> count == 1 ? null : count - 1);
+ }
+
+ private void registerPendingAuditEvent(PendingAuditEvent pending,
+ boolean transfersInFlightOwnership) {
+ pendingAuditBindingLock.lock();
+ try {
+ if (!pending.active.get()) {
+ return;
+ }
+ for (int i = 0; i < pending.expectedBackendIds.length; i++) {
+ if (bindParticipantFromExistingReportLocked(pending, i)) {
+ continue;
+ }
+ BackendQuery key = new BackendQuery(
+ pending.expectedBackendIds[i],
pending.statisticsQueryId);
+ unboundAuditParticipants.computeIfAbsent(key, ignored -> new
ArrayList<>())
+ .add(new PendingAuditBinding(pending, i));
+ }
+ if (transfersInFlightOwnership) {
+
releaseAuditStatisticsOwnershipLocked(pending.statisticsQueryId);
+ }
+ } finally {
+ pendingAuditBindingLock.unlock();
+ }
+ }
+
+ private void bindPendingAuditEvents(long backendId, long backendStartTime,
+ Set<String> reportedQueryIds) {
+ for (String queryId : reportedQueryIds) {
+ pendingAuditBindingLock.lock();
+ try {
+ BackendQuery key = new BackendQuery(backendId, queryId);
+ List<PendingAuditBinding> bindings =
unboundAuditParticipants.remove(key);
+ if (bindings == null) {
+ continue;
+ }
+ for (PendingAuditBinding binding : bindings) {
+ bindParticipantLocked(binding.pending,
binding.participantIndex,
+ backendStartTime);
+ }
+ } finally {
+ pendingAuditBindingLock.unlock();
+ }
+ }
+ }
+
+ private void bindParticipantFromExistingReport(PendingAuditEvent pending,
int participantIndex) {
+ if (pending.expectedBackendStartTimes.get(participantIndex) > 0) {
+ return;
+ }
+ pendingAuditBindingLock.lock();
+ try {
+ if (bindParticipantFromExistingReportLocked(pending,
participantIndex)) {
+ removeUnboundParticipantLocked(pending, participantIndex);
+ }
+ } finally {
+ pendingAuditBindingLock.unlock();
+ }
+ }
+
+ private boolean bindParticipantFromExistingReportLocked(
+ PendingAuditEvent pending, int participantIndex) {
+ if (!pending.active.get()) {
+ return false;
+ }
+ if (pending.expectedBackendStartTimes.get(participantIndex) > 0) {
+ return true;
+ }
+ long backendId = pending.expectedBackendIds[participantIndex];
+ long currentBackendStartTime = getBackendStartTime(backendId);
+ long reportStartTime = currentBackendStartTime > 0
+ ? currentBackendStartTime
+ : lastAcceptedBackendStartTimes.getOrDefault(backendId, 0L);
+ if (findStatisticsForBackend(pending.statisticsQueryId, backendId,
reportStartTime) == null) {
+ return false;
+ }
+ bindParticipantLocked(pending, participantIndex, reportStartTime);
+ return true;
+ }
+
+ private void bindParticipantLocked(PendingAuditEvent pending, int
participantIndex,
+ long backendStartTime) {
+ if (!pending.active.get()
+ || !pending.expectedBackendStartTimes.compareAndSet(
+ participantIndex, 0, backendStartTime)) {
+ return;
+ }
+ // Retention is query-scoped so one queued audit cannot pin unrelated
completed queries
+ // from the same long-lived backend process.
+ RetainedQuery retainedQuery = new RetainedQuery(
+ new
BackendIncarnation(pending.expectedBackendIds[participantIndex],
+ backendStartTime), pending.statisticsQueryId);
+ pendingQueryReferences.compute(retainedQuery,
+ (ignored, count) -> count == null ? 1 : count + 1);
+ }
+
+ private void unregisterPendingAuditEvent(PendingAuditEvent pending) {
+ pendingAuditBindingLock.lock();
+ try {
+ if (!pending.active.compareAndSet(true, false)) {
+ return;
+ }
+ for (int i = 0; i < pending.expectedBackendIds.length; i++) {
+ long backendStartTime =
pending.expectedBackendStartTimes.get(i);
+ if (backendStartTime > 0) {
+ RetainedQuery retainedQuery = new RetainedQuery(
+ new
BackendIncarnation(pending.expectedBackendIds[i], backendStartTime),
+ pending.statisticsQueryId);
+ pendingQueryReferences.computeIfPresent(retainedQuery,
(ignored, count) ->
+ count == 1 ? null : count - 1);
+ } else {
+ removeUnboundParticipantLocked(pending, i);
+ }
+ }
+ } finally {
+ pendingAuditBindingLock.unlock();
+ }
+ }
+
+ private void removeUnboundParticipantLocked(PendingAuditEvent pending, int
participantIndex) {
+ BackendQuery key = new BackendQuery(
+ pending.expectedBackendIds[participantIndex],
pending.statisticsQueryId);
+ List<PendingAuditBinding> bindings = unboundAuditParticipants.get(key);
+ if (bindings == null) {
+ return;
+ }
+ bindings.removeIf(binding -> binding.pending == pending
Review Comment:
[P2] Avoid quadratic cleanup for audits sharing one identity. Cached
full-prepare group commit can queue many events under the same (backendId,
loadId). If no matching report reaches this FE, maximum-wait expiry calls this
removeIf once per event, scanning N, N-1, ... remaining bindings under
pendingAuditBindingLock. At the 250,000-event configured scale, that can delay
audit draining and contend with submission, report binding, ownership, and
retention operations. Use identity-addressable unlinking or purge expired
bindings once per key, and add a high-cardinality same-loadId no-report timeout
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]
