github-actions[bot] commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3850723759
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1256,7 +1265,14 @@ private void forwardToMaster() throws Exception {
LOG.debug("need to transfer to Master. stmt: {}",
context.getStmtId());
}
masterOpExecutor.setMoreStmtExists(moreStmtExists);
+ // A follower has no local coordinator, so ownership must start before
the remote execution
+ // can produce an early final snapshot on one of its backends.
+ ensureAuditStatisticsOwnership();
masterOpExecutor.execute();
Review Comment:
[P1] Do not make the audit identity depend on receiving this non-idempotent
response. The master may complete a forwarded prepared group-commit and route
BE statistics `(loadId, backend)` to this follower, then the Thrift receive can
fail; DML is deliberately not retried, `FEOpExecutor.result` remains null, and
the follower's error audit uses the client query ID with no participants. Those
final counters can never be matched. Establish the statistics
identity/participant mapping before dispatch (or via an idempotent side
channel), and test successful master execution followed by response loss.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -24,50 +24,162 @@
import org.apache.doris.common.util.MasterDaemon;
import org.apache.doris.plugin.AuditEvent;
import org.apache.doris.qe.QeProcessorImpl;
+import org.apache.doris.system.Backend;
import org.apache.doris.thrift.TQueryStatistics;
import org.apache.doris.thrift.TQueryStatisticsResult;
import org.apache.doris.thrift.TReportWorkloadRuntimeStatusParams;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Lists;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
-import java.util.Iterator;
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLongArray;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
-// NOTE: not using a lock for beToQueryStatsMap's update because it should
void global lock for all be
-// this may cause in some corner case missing statistics update,for example:
-// time1: clear logic judge query 1 is timeout
-// time2: query 1 is update by report
-// time3: clear logic remove query 1
-// in this case, lost query stats is allowed. because query report time out is
60s by default,
-// when this case happens, we should find why be not report for so long first.
public class WorkloadRuntimeStatusMgr extends MasterDaemon {
private static final Logger LOG =
LogManager.getLogger(WorkloadRuntimeStatusMgr.class);
- // backend id --> {query id --> (query last report time, query stats)}
- private Map<Long, BeReportInfo> beToQueryStatsMap =
Maps.newConcurrentMap();
+ // backend process incarnation --> {query id --> (query last report time,
query stats)}
+ private final ConcurrentMap<BackendIncarnation, BeReportInfo>
beToQueryStatsMap = Maps.newConcurrentMap();
+ private final ConcurrentMap<Long, Long> lastAcceptedBackendStartTimes =
Maps.newConcurrentMap();
+ private final ConcurrentMap<RetainedQuery, Integer> pendingQueryReferences
+ = Maps.newConcurrentMap();
// Publish an immutable snapshot for synchronous proc/REST readers.
private volatile Map<String, TQueryStatistics> queryStatisticsSnapshot =
ImmutableMap.of();
private final ReentrantLock queryAuditEventLock = new ReentrantLock();
- private List<AuditEvent> queryAuditEventList = Lists.newLinkedList();
+ private final Set<PendingAuditEvent> queryAuditEventList = new
LinkedHashSet<>();
+ private final ReentrantLock pendingAuditBindingLock = new ReentrantLock();
+ private final Map<BackendQuery, List<PendingAuditBinding>>
unboundAuditParticipants = new HashMap<>();
+ private final Map<String, Integer> inFlightAuditQueryReferences = new
HashMap<>();
private volatile long lastWarnTime;
- private class BeReportInfo {
- volatile long beLastReportTime;
+ private static class BackendIncarnation {
+ final long backendId;
+ final long backendStartTime;
- BeReportInfo(long beLastReportTime) {
- this.beLastReportTime = beLastReportTime;
+ BackendIncarnation(long backendId, long backendStartTime) {
+ this.backendId = backendId;
+ this.backendStartTime = backendStartTime;
}
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof BackendIncarnation)) {
+ return false;
+ }
+ BackendIncarnation that = (BackendIncarnation) other;
+ return backendId == that.backendId && backendStartTime ==
that.backendStartTime;
+ }
+
+ @Override
+ public int hashCode() {
+ return Long.hashCode(backendId) * 31 +
Long.hashCode(backendStartTime);
+ }
+ }
+
+ private static class BeReportInfo {
+ final long backendStartTime;
+ final ReentrantReadWriteLock lifecycleLock = new
ReentrantReadWriteLock();
// query id --> (query last report time, query stats)
- Map<String, Pair<Long, TQueryStatisticsResult>> queryStatsMap =
Maps.newConcurrentMap();
+ final ConcurrentMap<String, Pair<Long, TQueryStatisticsResult>>
queryStatsMap
+ = Maps.newConcurrentMap();
+
+ BeReportInfo(long backendStartTime) {
+ this.backendStartTime = backendStartTime;
+ }
+ }
+
+ private static class BackendQuery {
+ final long backendId;
+ final String queryId;
+
+ BackendQuery(long backendId, String queryId) {
+ this.backendId = backendId;
+ this.queryId = queryId;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof BackendQuery)) {
+ return false;
+ }
+ BackendQuery that = (BackendQuery) other;
+ return backendId == that.backendId && queryId.equals(that.queryId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Long.hashCode(backendId) * 31 + queryId.hashCode();
+ }
+ }
+
+ private static class RetainedQuery {
+ final BackendIncarnation incarnation;
+ final String queryId;
+
+ RetainedQuery(BackendIncarnation incarnation, String queryId) {
+ this.incarnation = incarnation;
+ this.queryId = queryId;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (!(other instanceof RetainedQuery)) {
+ return false;
+ }
+ RetainedQuery that = (RetainedQuery) other;
+ return incarnation.equals(that.incarnation) &&
queryId.equals(that.queryId);
+ }
+
+ @Override
+ public int hashCode() {
+ return incarnation.hashCode() * 31 + queryId.hashCode();
+ }
+ }
+
+ private static class PendingAuditEvent {
+ final AuditEvent event;
+ final String statisticsQueryId;
Review Comment:
[P1] Bind each audit to its statistics generation, not just the cached load
ID. Full-prepare group commit deliberately reuses one `loadId` for successive
executions, and each call queues a different audit event. If E2 reports before
E1's audit drain, the higher generation replaces `(incarnation, loadId)`; E1
then hydrates from E2 (or waits for E2 to finish), so both records can log E2's
counters. Preserving both BE snapshots alone does not solve this because the
event carries no generation. Propagate a generation or unique execution
identity and test two rapid cached-plan executions with different counters.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -1256,7 +1265,14 @@ private void forwardToMaster() throws Exception {
LOG.debug("need to transfer to Master. stmt: {}",
context.getStmtId());
}
masterOpExecutor.setMoreStmtExists(moreStmtExists);
+ // A follower has no local coordinator, so ownership must start before
the remote execution
+ // can produce an early final snapshot on one of its backends.
+ ensureAuditStatisticsOwnership();
masterOpExecutor.execute();
+ if (masterOpExecutor.getAuditStatisticsQueryId() != null) {
Review Comment:
[P1] Preserve the returned audit identity when the journal wait fails.
`MasterOpExecutor.execute()` assigns its result before calling
`waitOnReplaying()`; a replay timeout therefore throws before this copy even
though the result already contains the group-commit load ID and backends.
Backend IDs later survive through the master-executor fallback, but
`getAuditStatisticsQueryId()` falls back only to the client ID, so the error
audit watches those backends under the wrong key and misses their counters.
Copy the fields in an exception-safe path whenever a result exists, and test a
successful forward response followed by a replay-wait failure.
##########
fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java:
##########
@@ -151,6 +151,7 @@ public static QueryState execUpdate(String sql) throws
Exception {
AutoCloseConnectContext r = StatisticsUtil.buildConnectContext(false);
try {
stmtExecutor = new StmtExecutor(r.connectContext, sql);
+ stmtExecutor.beginAuditStatisticsOwnership();
Review Comment:
[P1] Keep this owner reachable until the audit handoff. `r.close()` below
calls `ConnectContext.clear()` and nulls `executor` before `AuditLogHelper`
runs, so the helper sees no executor and can neither transfer nor release this
new reference. Every recurring internal statistics update therefore leaves an
`inFlightAuditQueryReferences` entry—and any accepted BE snapshot for that
query—permanently retained. Audit before closing the context (with
exception-safe release), and add repeated success/failure coverage proving
ownership returns to zero.
##########
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
Review Comment:
[P1] Bind the retained incarnation, not only the current heartbeat. A final
`(BE,A,q)` can be accepted while q has an in-flight owner, then heartbeat can
advance to B before audit submission. This lookup probes only B, so transfer
leaves q unbound even though A is retained; every drain retries B, and the
audit falls back without A's counters (or can consume a reused q from B).
Existing tests cover report -> enqueue -> restart and enqueue -> report ->
restart, but not report -> restart -> enqueue. Record and consume the accepted
incarnation during in-flight ownership and add that ordering 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]
