github-actions[bot] commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3849497419
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -466,25 +498,79 @@ void
RuntimeQueryStatisticsMgr::report_runtime_query_statistics() {
}
}
- // 3 when query is finished and (last rpc is send success), remove
finished query statistics
+ // 3 release accepted/expired compact snapshots and shed finished
ResourceContexts.
if (fe_qs_map.empty()) {
return;
}
{
std::lock_guard<std::shared_mutex>
write_lock(_resource_contexts_map_lock);
- for (auto& [addr, qs_map] : fe_qs_map) {
- bool is_rpc_success = rpc_result[addr];
- for (auto& [query_id, qs] : qs_map) {
- auto& qs_status_pair = qs_status[query_id];
- bool is_query_finished = qs_status_pair.first;
- bool is_timeout_after_finish = qs_status_pair.second;
- if ((is_rpc_success && is_query_finished) ||
is_timeout_after_finish) {
- _resource_contexts_map.erase(query_id);
- }
+ for (const auto& [query_id, reported_statistics] :
reported_pending_statistics) {
+ auto iter = _pending_final_statistics.find(query_id);
+ if (iter == _pending_final_statistics.end() ||
+ iter->second.statistics != reported_statistics) {
+ continue;
+ }
+ const auto& report_result = rpc_results[iter->second.fe_addr];
+ int64_t elapsed_after_finish_ms = MonotonicMillis() -
iter->second.finish_time_ms;
+ if (should_remove_finished_query(report_result.response_received,
+ report_result.accepted,
elapsed_after_finish_ms,
+
config::query_statistics_reserve_timeout_ms,
+ report_result.retention_ms)) {
+ _pending_final_statistics.erase(iter);
}
}
+
+ for (const auto& [query_id, final_statistics] :
active_final_statistics) {
+ auto iter = _resource_contexts_map.find(query_id);
+ // A later group-commit generation may reuse the query ID while
this RPC is in flight;
+ // only compact or erase the exact ResourceContext that produced
this snapshot.
+ if (iter == _resource_contexts_map.end() ||
+ iter->second != qs_resource_contexts[query_id]) {
+ continue;
+ }
+ auto resource_ctx = iter->second;
+ const auto& fe_addr = resource_ctx->task_controller()->fe_addr();
+ const auto& report_result = rpc_results[fe_addr];
+ int64_t finish_time_ms =
resource_ctx->task_controller()->finish_time();
+ int64_t elapsed_after_finish_ms = MonotonicMillis() -
finish_time_ms;
+ FinalStatisticsAction action = final_statistics_action(
+ report_result.response_received, report_result.accepted,
+ elapsed_after_finish_ms,
config::query_statistics_reserve_timeout_ms,
+ report_result.retention_ms);
+ if (action == FinalStatisticsAction::COMPACT_RESOURCE_CONTEXT) {
+ _pending_final_statistics[query_id] = {fe_addr,
final_statistics, finish_time_ms};
+ }
+ if (action != FinalStatisticsAction::KEEP_RESOURCE_CONTEXT) {
+ _resource_contexts_map.erase(iter);
+ }
+ }
+ }
+}
+
+bool RuntimeQueryStatisticsMgr::should_remove_finished_query(bool
response_received, bool accepted,
+ int64_t
elapsed_after_finish_ms,
+ int64_t
local_retention_ms,
+ int64_t
fe_retention_ms) {
+ if (accepted) {
+ return true;
+ }
+ // Without an FE response the BE does not know the audit fallback window,
so timing out here
+ // could discard the only final snapshot before a lagging FE can accept a
retry.
+ return response_received &&
Review Comment:
[P1] Bound final snapshots when an FE never responds. A transport/client
failure leaves `response_received` false, so this condition keeps every compact
entry forever; meanwhile each completed DML for that destination is copied into
every later retry payload. A permanently removed or partitioned FE therefore
grows `_pending_final_statistics` and the Thrift request without limit
(eventually making recovery harder once the batch becomes too large). This is
distinct from an explicit rejection, where FE supplies a retention window.
Please add a finite no-response/dead-destination policy and a reporter
failure/recovery test.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java:
##########
@@ -831,6 +831,7 @@ public TMasterOpResult proxyExecute(TMasterOpRequest
request) throws TException
}
}
if (executor != null) {
+ result.setAuditBackendIds(executor.getAuditBackendIds());
Review Comment:
[P1] Return participants from the executor that ran the prepared DML. In the
`prepareExecuteBuffer` branch, this local variable still refers to the
COM_STMT_PREPARE executor. `MysqlConnectProcessor.handleExecute(...)` creates
the actual COM_STMT_EXECUTE executor in the inherited instance field/context,
so `getAuditBackendIds()` here serializes an empty set. The follower can then
release the DML audit after the short timeout before periodic final counters
arrive. Read the execution executor after `handleExecute()` and cover a
forwarded prepared INSERT.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -147,92 +240,398 @@ public void submitFinishQueryToAudit(AuditEvent event) {
// 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);
+ // Register before publishing to the queue so cleanup cannot
remove a first report
+ // in the gap between queue insertion and participant
retention.
+ registerPendingAuditEvent(pending);
+ 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 {
+ for (PendingAuditEvent pending : queryAuditEventList) {
+ long elapsed = currentTime -
pending.event.pushToAuditLogQueueTime;
+ if (elapsed <= queryAuditLogTimeout) {
+ // Events are appended in timestamp order, so later
entries cannot be due yet.
break;
}
+ 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.event.queryId)
+ : new TQueryStatistics();
+ for (int i = 0; i < pending.expectedBackendIds.length; i++) {
+ bindParticipantFromExistingReport(pending, i);
+ long backendStartTime =
pending.expectedBackendStartTimes.get(i);
+ TQueryStatisticsResult statistics = findStatisticsForBackend(
+ pending.event.queryId, 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();
+ }
+ }
}
- for (Map.Entry<String, TQueryStatisticsResult> entry :
params.query_statistics_result_map.entrySet()) {
- beReportInfo.queryStatsMap.put(entry.getKey(),
Pair.of(currentTime, entry.getValue()));
+ // 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 void clearReportTimeoutBeStatistics() {
- // 1 clear report timeout be
- Set<Long> currentBeIdSet = beToQueryStatsMap.keySet();
+ 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;
+ }
+
+ private void registerPendingAuditEvent(PendingAuditEvent pending) {
+ 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.event.queryId);
+ unboundAuditParticipants.computeIfAbsent(key, ignored -> new
ArrayList<>())
+ .add(new PendingAuditBinding(pending, i));
+ }
+ } 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.event.queryId, 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.event.queryId);
+ 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.event.queryId);
+ 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.event.queryId);
+ List<PendingAuditBinding> bindings = unboundAuditParticipants.get(key);
+ if (bindings == null) {
+ return;
+ }
+ bindings.removeIf(binding -> binding.pending == pending
+ && binding.participantIndex == participantIndex);
+ if (bindings.isEmpty()) {
+ unboundAuditParticipants.remove(key);
+ }
+ }
+
+ @VisibleForTesting
+ void beforeHydrateAuditEvent(String queryId) {
+ }
+
+ @VisibleForTesting
+ void beforeBindPendingAuditEvents() {
+ }
+
+ private boolean isQueryRetainedForAudit(BackendIncarnation incarnation,
String queryId) {
+ RetainedQuery retainedQuery = new RetainedQuery(incarnation, queryId);
+ if (pendingQueryReferences.containsKey(retainedQuery)) {
+ return true;
+ }
+ pendingAuditBindingLock.lock();
+ try {
+ // Serialize cleanup with first-report binding. Until the
participant has an epoch,
+ // retaining only this backend/query candidate closes the
insert-to-bind race.
+ return pendingQueryReferences.containsKey(retainedQuery)
+ || unboundAuditParticipants.containsKey(
+ new BackendQuery(incarnation.backendId, queryId));
+ } finally {
+ pendingAuditBindingLock.unlock();
+ }
+ }
+
+ @VisibleForTesting
+ long getBackendStartTime(long backendId) {
+ Backend backend = Env.getCurrentSystemInfo().getBackend(backendId);
+ return backend == null ? 0 : backend.getLastStartTime();
+ }
+
+ @VisibleForTesting
+ void beforeUpdateQueryStatistics(String queryId) {
+ }
+
+ 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;
+ }
+
+ long previousSequence = previous.isSetQueryStatisticsSequence()
+ ? previous.getQueryStatisticsSequence() : 0;
+ long incomingSequence = incoming.isSetQueryStatisticsSequence()
+ ? incoming.getQueryStatisticsSequence() : 0;
+ if (previousSequence != incomingSequence) {
+ return incomingSequence > previousSequence;
+ }
+
+ return false;
+ }
+
+ void clearReportTimeoutBeStatistics() {
+ Set<BackendIncarnation> 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);
+ for (BackendIncarnation incarnation : currentBeIdSet) {
+ BeReportInfo beReportInfo = beToQueryStatsMap.get(incarnation);
+ if (beReportInfo == null) {
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);
+ beforeExpireQueryStatistics(queryId);
+ beReportInfo.queryStatsMap.computeIfPresent(queryId,
(ignoredQueryId, pair) -> {
+ long queryLastReportTime = pair.first;
+ boolean timeout = currentTime - queryLastReportTime
+ > Config.be_report_query_statistics_timeout_ms;
+ // Conditional removal and updates serialize only for the
same query. Work on
+ // a high-cardinality report cannot delay another query's
terminal update.
+ return timeout && !isQueryRetainedForAudit(incarnation,
queryId)
Review Comment:
[P1] Retain accepted finals until audit ownership is registered. This
cleaner can expire S1 whenever the query is absent from the local QeProcessor
and no pending reference exists. On a follower forwarding a long external
INSERT, the follower never owns the master's coordinator: BEs report to the
follower, so an early participant's accepted S1 can age out while the master is
still finishing, before `TMasterOpResult` returns the participant set. Local
inserts also unregister before `auditAfterExec()` submits the event, and this
check is not atomic with concurrent registration. BE has already dropped S1
after acceptance, so the audit cannot recover it. Make coordinator-to-audit
ownership transfer atomic and test an aged forwarded final plus cleanup racing
audit registration.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -147,92 +240,398 @@ public void submitFinishQueryToAudit(AuditEvent event) {
// 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);
+ // Register before publishing to the queue so cleanup cannot
remove a first report
+ // in the gap between queue insertion and participant
retention.
+ registerPendingAuditEvent(pending);
+ 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 {
+ for (PendingAuditEvent pending : queryAuditEventList) {
+ long elapsed = currentTime -
pending.event.pushToAuditLogQueueTime;
+ if (elapsed <= queryAuditLogTimeout) {
Review Comment:
[P2] Avoid wall-clock head-of-line blocking here. Events are insertion
ordered, but their enqueue time comes from `System.currentTimeMillis()`. After
a backward clock correction, head event A can have a later timestamp than
subsequently inserted B; A is not due, this `break` skips B even after B
exceeds its audit deadline, and the same head can retain the whole later queue
until wall time catches up. Use a monotonic deadline/timestamp in the pending
entry, or continue scanning past non-due events, with a reversed-timestamp
regression test.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java:
##########
@@ -2284,6 +2285,14 @@ public Coordinator getCoord() {
return coord;
}
+ public Set<Long> getAuditBackendIds() {
+ if (coord != null) {
+ return ImmutableSet.copyOf(coord.getInvolvedBackendIds());
+ }
+ return masterOpExecutor == null
Review Comment:
[P1] Include full-prepare group-commit executions in audit binding. This
fast path dispatches a real fragment through `GroupCommitPlanner`, but it
installs neither a coordinator nor a master executor, so this fallback returns
no participant. On reuse, the cached planner also keeps the first execution's
`loadId` as the BE query ID while the audit event uses the new COM_STMT_EXECUTE
query ID, making lookup impossible even if the backend were known. Propagate
the selected backend and actual statistics identity to the audit event, and
test two prepared group-commit executions with auditing enabled.
--
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]
