This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new d2d9256b77d [fix](audit) preserve final statistics for external writes
(#67053)
d2d9256b77d is described below
commit d2d9256b77d0a9c986820d044da629b2106ef330
Author: Gabriel <[email protected]>
AuthorDate: Wed Aug 26 19:29:38 2026 +0800
[fix](audit) preserve final statistics for external writes (#67053)
## What problem does this PR solve?
External writes can enqueue their audit event before every BE reports
its final cumulative query statistics. This causes scan rows and bytes,
local and remote scan bytes, CPU time, and peak memory to remain zero or
incomplete.
## What is changed?
- Classify the target from the resolved insert executor instead of the
mutable logical-plan root. This includes plugin-driven and Remote Doris
external tables while excluding internal OLAP, CTE-wrapped internal
inserts, dictionary, and blackhole/WARM UP execution.
- Record only BEs whose fragment RPC dispatch was attempted,
conservatively including uncertain RPC outcomes and excluding planned
but never-dispatched workers.
- Apply the final-report barrier to successful, failed, and cancelled
external writes after dispatch.
- Cover external `INSERT INTO`, external `INSERT OVERWRITE`, and `FILES`
sinks.
- Return the participant BE IDs to a forwarding FE through one optional
FE-to-FE result field, so the FE receiving BE statistics owns the audit
barrier.
- Rebuild the immutable statistics snapshot after all available final
reports arrive. Expensive participant lookups run outside the audit
queue lock.
- Keep the existing BE statistics report timeout as a bounded fallback
when a final report is unavailable.
## Scope boundary
In scope:
- Nereids external-table `INSERT INTO` and `INSERT OVERWRITE`, including
plugin-driven and Remote Doris targets
- `INSERT INTO FILES`
- Direct execution and normal `COM_QUERY` forwarding to the master
- Success, execution failure, and cancellation after fragment dispatch
- Classic and Nereids coordinators, including multi-BE execution
Explicitly unchanged:
- Internal OLAP, dictionary, blackhole/WARM UP, SELECT, UPDATE, and
DELETE audit behavior
- Analysis failures, empty writes, and other paths with no dispatched
fragment; these keep the normal audit timeout because no final BE
statistics exist
- Group commit and forwarded binary prepared-statement execution
- BE reporting, retention, retry, configuration, and FE-to-BE protocols
- Audit queue overflow behavior and the bounded fallback when a BE final
report is unavailable
- A transport failure before the master response is received follows the
existing forwarded error-audit path because no participant result is
available
The only protocol addition is an optional FE-to-FE participant-ID field;
write transactions, commit behavior, and BE execution are not changed.
## Testing
- 43 focused FE unit tests passed across
`AuditLogHelperBackendSelectionTest`,
`MasterOpExecutorBackendSelectionTest`, `PipelineExecutionTaskTest`,
`WorkloadRuntimeStatusMgrTest`, and `InsertIntoTableCommandTest`
- FE Checkstyle: 0 errors
- `git diff --check`: passed
- Automatic merge with current `origin/master`: no conflicts
- The same 43 focused tests passed on the merged tree
---
.../commands/insert/InsertIntoTVFCommand.java | 3 +
.../commands/insert/InsertIntoTableCommand.java | 14 ++-
.../java/org/apache/doris/qe/AuditLogHelper.java | 13 +-
.../java/org/apache/doris/qe/ConnectProcessor.java | 5 +
.../main/java/org/apache/doris/qe/Coordinator.java | 9 ++
.../java/org/apache/doris/qe/FEOpExecutor.java | 9 ++
.../org/apache/doris/qe/NereidsCoordinator.java | 8 ++
.../java/org/apache/doris/qe/StmtExecutor.java | 14 +++
.../doris/qe/runtime/PipelineExecutionTask.java | 11 ++
.../WorkloadRuntimeStatusMgr.java | 78 ++++++++++--
.../insert/InsertIntoTableCommandTest.java | 12 ++
.../qe/AuditLogHelperBackendSelectionTest.java | 108 ++++++++++++++++
.../qe/MasterOpExecutorBackendSelectionTest.java | 26 ++++
.../qe/runtime/PipelineExecutionTaskTest.java | 44 +++++++
.../WorkloadRuntimeStatusMgrTest.java | 138 +++++++++++++++++++++
gensrc/thrift/FrontendService.thrift | 2 +
16 files changed, 477 insertions(+), 17 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTVFCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTVFCommand.java
index 612c7fe33c0..6c8826c6dae 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTVFCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTVFCommand.java
@@ -126,6 +126,9 @@ public class InsertIntoTVFCommand extends Command
implements ForwardWithSync, Ex
new QueryInfo(ctx, "INSERT INTO TVF", coordinator));
try {
+ // Audit ownership follows the executor that resolved the external
sink, not a logical
+ // plan shape that can also represent internal or blackhole writes.
+ executor.setExternalDmlAuditCoordinator(coordinator);
coordinator.exec();
// Wait for completion
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
index 56deaa27a2f..4a1440872f0 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
@@ -357,12 +357,22 @@ public class InsertIntoTableCommand extends Command
implements NeedAuditEncrypti
// so we need to set this here
insertExecutor.getCoordinator().setTxnId(insertExecutor.getTxnId());
stmtExecutor.setCoord(insertExecutor.getCoordinator());
+ if (needsExternalDmlAuditBarrier(insertExecutor)) {
+ // The resolved executor is the invariant that distinguishes
an external write;
+ // logical sink roots are rewritten and are not a stable audit
classification.
+
stmtExecutor.setExternalDmlAuditCoordinator(insertExecutor.getCoordinator());
+ }
return insertExecutor;
}
LOG.warn("insert plan failed {} times. query id is {}.", retryTimes,
DebugUtil.printId(ctx.queryId()));
throw new AnalysisException("Insert plan failed. Could not get target
table lock.");
}
+ static boolean needsExternalDmlAuditBarrier(AbstractInsertExecutor
insertExecutor) {
+ return insertExecutor instanceof BaseExternalTableInsertExecutor
+ || insertExecutor instanceof RemoteOlapInsertExecutor;
+ }
+
/**
* Hook method to determine if auth check is needed.
* Subclasses can override this to skip auth check, e.g.,
WarmupSelectCommand.
@@ -689,10 +699,6 @@ public class InsertIntoTableCommand extends Command
implements NeedAuditEncrypti
LineageUtils.submitLineageEventIfNeeded(executor, lineagePlan,
getLogicalQuery(), getClass());
}
- public boolean isExternalTableSink() {
- return !(getLogicalQuery() instanceof UnboundTableSink);
- }
-
/**
* get the target table of the insert command
*/
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/AuditLogHelper.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/AuditLogHelper.java
index 57b84d2594e..8afa17a49bc 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/AuditLogHelper.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/AuditLogHelper.java
@@ -51,6 +51,7 @@ import org.apache.doris.resource.workloadgroup.QueueToken;
import org.apache.doris.service.FrontendOptions;
import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.logging.log4j.LogManager;
@@ -408,12 +409,22 @@ public class AuditLogHelper {
auditEventBuilder.setState(String.valueOf(MysqlStateType.OK));
}
AuditEvent event = auditEventBuilder.build();
-
Env.getCurrentEnv().getWorkloadRuntimeStatusMgr().submitFinishQueryToAudit(event);
+ Set<Long> externalDmlBackendIds =
getExternalDmlAuditBackendIds(ctx.getExecutor());
+ if (externalDmlBackendIds.isEmpty()) {
+
Env.getCurrentEnv().getWorkloadRuntimeStatusMgr().submitFinishQueryToAudit(event);
+ } else {
+ Env.getCurrentEnv().getWorkloadRuntimeStatusMgr()
+ .submitFinishQueryToAudit(event, externalDmlBackendIds);
+ }
if (LOG.isDebugEnabled()) {
LOG.debug("submit audit event: {}", event.queryId);
}
}
+ static Set<Long> getExternalDmlAuditBackendIds(StmtExecutor executor) {
+ return executor == null ? ImmutableSet.of() :
executor.getExternalDmlAuditBackendIds();
+ }
+
private static long getQueueTimeMs(ConnectContext ctx) {
QueueToken queueToken = null;
if (ctx.getExecutor() != null && ctx.getExecutor().getCoord() != null)
{
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
index c83397f58dc..ed0cda0ab1f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
@@ -831,6 +831,11 @@ public abstract class ConnectProcessor {
}
}
if (executor != null) {
+ List<Long> auditStatisticsBackendIds = Lists.newArrayList(
+ AuditLogHelper.getExternalDmlAuditBackendIds(executor));
+ if (!auditStatisticsBackendIds.isEmpty()) {
+ result.setAuditStatisticsBackendIds(auditStatisticsBackendIds);
+ }
if (executor.getProxyShowResultSet() != null) {
result.setResultSet(executor.getProxyShowResultSet().tothrift());
} else if (!executor.getProxyQueryResultBufList().isEmpty()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
index ca489b86258..4a9e58949ca 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
@@ -138,6 +138,7 @@ import com.google.common.base.Strings;
import com.google.common.collect.HashMultiset;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multiset;
@@ -236,6 +237,7 @@ public class Coordinator implements CoordInterface {
private final List<PlanFragment> fragments;
private Map<Long, PipelineExecContexts> beToPipelineExecCtxs =
Maps.newHashMap();
+ private final Set<Long> dispatchedBackendIdsForAudit =
Sets.newConcurrentHashSet();
private final Map<Pair<Integer, Long>, PipelineExecContext>
pipelineExecContexts = new HashMap<>();
private final List<PipelineExecContext> needCheckPipelineExecContexts =
Lists.newArrayList();
@@ -922,6 +924,7 @@ public class Coordinator implements CoordInterface {
int backendIdx = 0;
int profileFragmentId = 0;
beToPipelineExecCtxs.clear();
+ dispatchedBackendIdsForAudit.clear();
// fragment:backend
List<Pair<PlanFragmentId, Long>> backendFragments =
Lists.newArrayList();
// If #fragments >=2, use twoPhaseExecution with
exec_plan_fragments_prepare and exec_plan_fragments_start,
@@ -1050,6 +1053,8 @@ public class Coordinator implements CoordInterface {
if (LOG.isDebugEnabled()) {
LOG.debug(ctxs.debugInfo());
}
+ // Include uncertain RPC outcomes, but never a planned backend
whose dispatch was not attempted.
+ dispatchedBackendIdsForAudit.add(ctxs.getBackend().getId());
futures.add(Pair.of(DateTime.now().getMillis(),
ImmutableTriple.of(ctxs, proxy,
ctxs.execRemoteFragmentsAsync(proxy))));
}
@@ -3842,6 +3847,10 @@ public class Coordinator implements CoordInterface {
return backendAddresses;
}
+ public Set<Long> getDispatchedBackendIdsForAudit() {
+ return ImmutableSet.copyOf(dispatchedBackendIdsForAudit);
+ }
+
/**
* Returns the IDs of backends that have scan ranges assigned, collected
from each ScanNode's
* scanBackendIds (populated during plan phase).
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java
index e0b27673727..f0e31c4aa5f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java
@@ -37,6 +37,7 @@ import org.apache.doris.thrift.TUniqueId;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -47,6 +48,7 @@ import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/**
* FEOpExecutor is used to send request to specific FE
@@ -264,6 +266,13 @@ public class FEOpExecutor {
}
}
+ public Set<Long> getAuditStatisticsBackendIds() {
+ if (result == null || !result.isSetAuditStatisticsBackendIds()) {
+ return Collections.emptySet();
+ }
+ return ImmutableSet.copyOf(result.getAuditStatisticsBackendIds());
+ }
+
public String getProxyStatus() {
if (result == null) {
return QueryState.MysqlStateType.UNKNOWN.name();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
index 3e991e7755c..74671eac0bf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
@@ -68,6 +68,7 @@ import org.apache.doris.thrift.TTabletCommitInfo;
import org.apache.doris.thrift.TUniqueId;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.logging.log4j.LogManager;
@@ -76,6 +77,7 @@ import org.apache.logging.log4j.Logger;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
/** NereidsCoordinator */
@@ -373,6 +375,12 @@ public class NereidsCoordinator extends Coordinator {
return
Utils.fastToImmutableList(coordinatorContext.backends.get().keySet());
}
+ @Override
+ public Set<Long> getDispatchedBackendIdsForAudit() {
+ return executionTask == null
+ ? ImmutableSet.of() :
executionTask.getDispatchedBackendIdsForAudit();
+ }
+
@Override
public List<FragmentInstanceInfo> getFragmentInstanceInfos() {
List<QueryStatisticsItem.FragmentInstanceInfo> infos =
Lists.newArrayList();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
index a12b458708e..76456330951 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
@@ -201,6 +201,7 @@ public class StmtExecutor {
@Setter
private volatile Coordinator coord = null;
+ private volatile Coordinator externalDmlAuditCoordinator = null;
// Arrow Flight SQL: when true, this query's coordinator is kept alive
past GetFlightInfo and
// is finalized later by ConnectContext (see #62259), so the eager close
in executeAndSendResult
// is skipped.
@@ -506,6 +507,18 @@ public class StmtExecutor {
return masterOpExecutor != null;
}
+ public void setExternalDmlAuditCoordinator(Coordinator coordinator) {
+ externalDmlAuditCoordinator = coordinator;
+ }
+
+ public Set<Long> getExternalDmlAuditBackendIds() {
+ if (masterOpExecutor != null) {
+ return masterOpExecutor.getAuditStatisticsBackendIds();
+ }
+ return externalDmlAuditCoordinator == null
+ ? Collections.emptySet() :
externalDmlAuditCoordinator.getDispatchedBackendIdsForAudit();
+ }
+
public ShowResultSet getProxyShowResultSet() {
return proxyShowResultSet;
}
@@ -684,6 +697,7 @@ public class StmtExecutor {
public void execute(TUniqueId queryId) throws Exception {
SessionVariable sessionVariable = context.getSessionVariable();
context.setEffectiveCloudCluster(null);
+ externalDmlAuditCoordinator = null;
if (context.getConnectType() == ConnectType.ARROW_FLIGHT_SQL) {
context.setReturnResultFromLocal(true);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/PipelineExecutionTask.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/PipelineExecutionTask.java
index 59670ccbbf1..6ea7842562c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/PipelineExecutionTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/PipelineExecutionTask.java
@@ -36,6 +36,7 @@ import org.apache.doris.thrift.TStatusCode;
import org.apache.doris.thrift.TUniqueId;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -46,6 +47,8 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -64,6 +67,7 @@ public class PipelineExecutionTask extends
AbstractRuntimeTask<BackendWorker, Mu
private final long timeoutDeadline;
private final CoordinatorContext coordinatorContext;
private final BackendServiceProxy backendServiceProxy;
+ private final Set<Long> dispatchedBackendIdsForAudit =
ConcurrentHashMap.newKeySet();
// mutable states
public PipelineExecutionTask(
@@ -93,6 +97,7 @@ public class PipelineExecutionTask extends
AbstractRuntimeTask<BackendWorker, Mu
@Override
public void execute() throws Exception {
coordinatorContext.withLock(() -> {
+ dispatchedBackendIdsForAudit.clear();
sendAndWaitPhaseOneRpc();
if (coordinatorContext.twoPhaseExecution()) {
sendAndWaitPhaseTwoRpc();
@@ -115,6 +120,8 @@ public class PipelineExecutionTask extends
AbstractRuntimeTask<BackendWorker, Mu
private void sendAndWaitPhaseOneRpc() throws UserException, RpcException {
List<RpcInfo> rpcs = Lists.newArrayList();
for (MultiFragmentsPipelineTask fragmentsTask :
childrenTasks.allTasks()) {
+ // Include uncertain RPC outcomes, but never a planned backend
whose dispatch was not attempted.
+
dispatchedBackendIdsForAudit.add(fragmentsTask.getBackend().getId());
rpcs.add(new RpcInfo(
fragmentsTask,
DateTime.now().getMillis(),
@@ -129,6 +136,10 @@ public class PipelineExecutionTask extends
AbstractRuntimeTask<BackendWorker, Mu
coordinatorContext.updateProfileIfPresent(profile ->
profile.setRpcPhase1Latency(rpcPhase1Latency));
}
+ public Set<Long> getDispatchedBackendIdsForAudit() {
+ return ImmutableSet.copyOf(dispatchedBackendIdsForAudit);
+ }
+
private void sendAndWaitPhaseTwoRpc() throws RpcException, UserException {
List<RpcInfo> rpcs = Lists.newArrayList();
for (MultiFragmentsPipelineTask fragmentTask :
childrenTasks.allTasks()) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java
index 7920b1e59a2..812294b5148 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java
@@ -29,12 +29,15 @@ import org.apache.doris.thrift.TQueryStatisticsResult;
import org.apache.doris.thrift.TReportWorkloadRuntimeStatusParams;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
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.Collections;
+import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -57,6 +60,7 @@ public class WorkloadRuntimeStatusMgr extends MasterDaemon {
private volatile Map<String, TQueryStatistics> queryStatisticsSnapshot =
ImmutableMap.of();
private final ReentrantLock queryAuditEventLock = new ReentrantLock();
private List<AuditEvent> queryAuditEventList = Lists.newLinkedList();
+ private final Map<AuditEvent, Set<Long>> externalDmlAuditBackendIds = new
IdentityHashMap<>();
private volatile long lastWarnTime;
private class BeReportInfo {
@@ -76,14 +80,12 @@ public class WorkloadRuntimeStatusMgr extends MasterDaemon {
@Override
protected void runAfterCatalogReady() {
- // 1 rebuild and publish query statistics snapshot
- rebuildQueryStatisticsSnapshot();
- // 2 read the latest immutable snapshot for downstream processing
- Map<String, TQueryStatistics> queryStatisticsMap =
getQueryStatisticsMap();
-
- // 3 log query audit
try {
List<AuditEvent> auditEventList = getQueryNeedAudit();
+ // Once an external write is ready, rebuild after the readiness
check so the audit
+ // event observes the same final BE reports that satisfied its
completion barrier.
+ rebuildQueryStatisticsSnapshot();
+ Map<String, TQueryStatistics> queryStatisticsMap =
getQueryStatisticsMap();
int missedLogCount = 0;
int succLogCount = 0;
for (AuditEvent auditEvent : auditEventList) {
@@ -115,7 +117,7 @@ public class WorkloadRuntimeStatusMgr extends MasterDaemon {
LOG.warn("exception happens when handleAuditEvent, ", t);
}
- // 4 clear beToQueryStatsMap when be report timeout
+ // clear beToQueryStatsMap when be report timeout
clearReportTimeoutBeStatistics();
}
@@ -127,6 +129,10 @@ public class WorkloadRuntimeStatusMgr extends MasterDaemon
{
// And the worker thread will get an event from the queue and get the
statistic info for this
// event from queryStatisticsMap.
public void submitFinishQueryToAudit(AuditEvent event) {
+ submitFinishQueryToAudit(event, ImmutableSet.of());
+ }
+
+ public void submitFinishQueryToAudit(AuditEvent event, Set<Long>
expectedBackendIds) {
queryAuditEventLogWriteLock();
try {
if (queryAuditEventList.size() >
Config.audit_event_log_queue_size) {
@@ -148,6 +154,9 @@ public class WorkloadRuntimeStatusMgr extends MasterDaemon {
// the worker thread will try best to wait for the statistic
info before logging this event.
event.pushToAuditLogQueueTime = System.currentTimeMillis();
queryAuditEventList.add(event);
+ if (expectedBackendIds != null &&
!expectedBackendIds.isEmpty()) {
+ externalDmlAuditBackendIds.put(event,
ImmutableSet.copyOf(expectedBackendIds));
+ }
}
} finally {
queryAuditEventLogWriteUnlock();
@@ -155,19 +164,49 @@ public class WorkloadRuntimeStatusMgr extends
MasterDaemon {
}
private List<AuditEvent> getQueryNeedAudit() {
- List<AuditEvent> ret = new ArrayList<>();
long currentTime = System.currentTimeMillis();
+ int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
+ long maximumWaitMs = Math.max(queryAuditLogTimeout,
+ Config.be_report_query_statistics_timeout_ms);
+ Map<AuditEvent, Set<Long>> dueExternalDmlEvents = new
IdentityHashMap<>();
+ Set<AuditEvent> readyEvents = Collections.newSetFromMap(new
IdentityHashMap<>());
+
+ queryAuditEventLogWriteLock();
+ try {
+ for (AuditEvent ae : queryAuditEventList) {
+ long waitTimeMs = currentTime - ae.pushToAuditLogQueueTime;
+ if (waitTimeMs <= queryAuditLogTimeout) {
+ continue;
+ }
+ Set<Long> expectedBackendIds =
externalDmlAuditBackendIds.get(ae);
+ if (expectedBackendIds == null || waitTimeMs > maximumWaitMs) {
+ readyEvents.add(ae);
+ } else {
+ dueExternalDmlEvents.put(ae, expectedBackendIds);
+ }
+ }
+ } finally {
+ queryAuditEventLogWriteUnlock();
+ }
+
+ // BE lookups are O(events * participants), so keep them outside the
queue lock that
+ // statement threads need in order to submit unrelated audit events.
+ for (Map.Entry<AuditEvent, Set<Long>> entry :
dueExternalDmlEvents.entrySet()) {
+ if (haveAllBackendsReportedFinalStatistics(entry.getKey().queryId,
entry.getValue())) {
+ readyEvents.add(entry.getKey());
+ }
+ }
+
+ List<AuditEvent> ret = new ArrayList<>();
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) {
+ if (readyEvents.contains(ae)) {
ret.add(ae);
iter.remove();
- } else {
- break;
+ externalDmlAuditBackendIds.remove(ae);
}
}
} finally {
@@ -176,6 +215,21 @@ public class WorkloadRuntimeStatusMgr extends MasterDaemon
{
return ret;
}
+ private boolean haveAllBackendsReportedFinalStatistics(String queryId,
Set<Long> expectedBackendIds) {
+ for (Long backendId : expectedBackendIds) {
+ BeReportInfo reportInfo = beToQueryStatsMap.get(backendId);
+ if (reportInfo == null) {
+ return false;
+ }
+ Pair<Long, TQueryStatisticsResult> queryStatistics =
reportInfo.queryStatsMap.get(queryId);
+ if (queryStatistics == null || queryStatistics.second == null
+ || !queryStatistics.second.isQueryFinished()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
public void updateBeQueryStats(TReportWorkloadRuntimeStatusParams params) {
if (!params.isSetBackendId()) {
LOG.warn("be report workload runtime status but without beid");
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommandTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommandTest.java
index 8a8ebc835b0..32cac86e019 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommandTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommandTest.java
@@ -214,4 +214,16 @@ class InsertIntoTableCommandTest {
Assertions.assertFalse(nonPlugin,
"a non-plugin table type must NOT be treated as write-branch
capable");
}
+
+ @Test
+ void testExternalDmlAuditClassificationUsesResolvedExecutor() {
+
Assertions.assertTrue(InsertIntoTableCommand.needsExternalDmlAuditBarrier(
+ Mockito.mock(BaseExternalTableInsertExecutor.class)));
+
Assertions.assertTrue(InsertIntoTableCommand.needsExternalDmlAuditBarrier(
+ Mockito.mock(RemoteOlapInsertExecutor.class)));
+
Assertions.assertFalse(InsertIntoTableCommand.needsExternalDmlAuditBarrier(
+ Mockito.mock(OlapInsertExecutor.class)));
+
Assertions.assertFalse(InsertIntoTableCommand.needsExternalDmlAuditBarrier(
+ Mockito.mock(BlackholeInsertExecutor.class)));
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/AuditLogHelperBackendSelectionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/AuditLogHelperBackendSelectionTest.java
index fbf545fe2da..ad1bef6ec08 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/qe/AuditLogHelperBackendSelectionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/AuditLogHelperBackendSelectionTest.java
@@ -17,11 +17,21 @@
package org.apache.doris.qe;
+import org.apache.doris.analysis.StmtType;
import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.Config;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.common.profile.SummaryProfile;
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.glue.LogicalPlanAdapter;
+import
org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTVFCommand;
+import
org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand;
+import
org.apache.doris.nereids.trees.plans.commands.insert.InsertOverwriteTableCommand;
+import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.plugin.AuditEvent;
import org.apache.doris.resource.BackendSelection;
import org.apache.doris.resource.BackendSelectionManager;
@@ -35,6 +45,8 @@ import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.util.Set;
+
public class AuditLogHelperBackendSelectionTest {
@After
@@ -59,6 +71,102 @@ public class AuditLogHelperBackendSelectionTest {
"load_group", BackendSelection.Mode.PREFER, "test"),
"cloud_group", "cloud", "cloud_group");
}
+ @Test
+ public void testSuccessfulExternalInsertWaitsForItsCoordinatorBackends()
throws Exception {
+ InsertIntoTableCommand command =
Mockito.mock(InsertIntoTableCommand.class);
+ Mockito.when(command.stmtType()).thenReturn(StmtType.INSERT);
+ assertExternalDmlWaitsForCoordinatorBackends(command, true);
+ }
+
+ @Test
+ public void testFailedExternalInsertWaitsForItsCoordinatorBackends()
throws Exception {
+ InsertIntoTableCommand command =
Mockito.mock(InsertIntoTableCommand.class);
+ Mockito.when(command.stmtType()).thenReturn(StmtType.INSERT);
+ assertExternalDmlWaitsForCoordinatorBackends(command, false);
+ }
+
+ @Test
+ public void testExternalInsertOverwriteWaitsForItsCoordinatorBackends()
throws Exception {
+ InsertOverwriteTableCommand command =
Mockito.mock(InsertOverwriteTableCommand.class);
+ Mockito.when(command.stmtType()).thenReturn(StmtType.INSERT);
+ assertExternalDmlWaitsForCoordinatorBackends(command, true);
+ }
+
+ @Test
+ public void testFilesInsertWaitsForItsCoordinatorBackends() throws
Exception {
+ InsertIntoTVFCommand command =
Mockito.mock(InsertIntoTVFCommand.class);
+ Mockito.when(command.stmtType()).thenReturn(StmtType.INSERT);
+ assertExternalDmlWaitsForCoordinatorBackends(command, true);
+ }
+
+ @Test
+ public void testInternalInsertDoesNotUseExternalDmlBarrier() {
+ StmtExecutor executor = Mockito.mock(StmtExecutor.class,
Mockito.CALLS_REAL_METHODS);
+
+
Assert.assertTrue(AuditLogHelper.getExternalDmlAuditBackendIds(executor).isEmpty());
+ }
+
+ @Test
+ public void testForwardedExternalDmlUsesBackendIdsReturnedByMaster() {
+ Set<Long> expectedBackendIds = Set.of(10001L, 10002L);
+ MasterOpExecutor masterExecutor = Mockito.mock(MasterOpExecutor.class);
+
Mockito.when(masterExecutor.getAuditStatisticsBackendIds()).thenReturn(expectedBackendIds);
+ StmtExecutor executor = Mockito.mock(StmtExecutor.class,
Mockito.CALLS_REAL_METHODS);
+ Deencapsulation.setField(executor, "masterOpExecutor", masterExecutor);
+
+ Assert.assertEquals(expectedBackendIds,
+ AuditLogHelper.getExternalDmlAuditBackendIds(executor));
+ }
+
+ @Test
+ public void testResolvedExternalDmlUsesOnlyDispatchedBackends() {
+ Coordinator coordinator = Mockito.mock(Coordinator.class);
+
Mockito.when(coordinator.getDispatchedBackendIdsForAudit()).thenReturn(Set.of(10001L));
+ StmtExecutor executor = Mockito.mock(StmtExecutor.class,
Mockito.CALLS_REAL_METHODS);
+
+ executor.setExternalDmlAuditCoordinator(coordinator);
+
+ Assert.assertEquals(Set.of(10001L),
executor.getExternalDmlAuditBackendIds());
+ }
+
+ private void assertExternalDmlWaitsForCoordinatorBackends(LogicalPlan
command, boolean success)
+ throws Exception {
+ ConnectContext context = Mockito.spy(new ConnectContext());
+ context.setStartTime();
+ context.setCurrentUserIdentity(UserIdentity.ROOT);
+ if (success) {
+ context.getState().setOk();
+ } else {
+ context.getState().setError(ErrorCode.ERR_UNKNOWN_ERROR, "external
write failed");
+ }
+
+ StmtExecutor executor = Mockito.mock(StmtExecutor.class);
+
Mockito.when(executor.getExternalDmlAuditBackendIds()).thenReturn(Set.of(10001L,
10002L));
+
Mockito.when(executor.getSummaryProfile()).thenReturn(Mockito.mock(SummaryProfile.class));
+ context.setExecutor(executor);
+
+ LogicalPlanAdapter statement = new LogicalPlanAdapter(command, new
StatementContext());
+
+ Env env = Mockito.mock(Env.class);
+ CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
+ CatalogIf catalog = Mockito.mock(CatalogIf.class);
+ WorkloadRuntimeStatusMgr statusMgr =
Mockito.mock(WorkloadRuntimeStatusMgr.class);
+ Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
+
Mockito.when(catalogMgr.getCatalog(Mockito.anyString())).thenReturn(catalog);
+ Mockito.when(catalog.getName()).thenReturn("internal");
+ Mockito.when(env.getWorkloadRuntimeStatusMgr()).thenReturn(statusMgr);
+
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+
+ AuditLogHelper.logAuditLog(context, "insert into external_table
select 1",
+ statement, null, true);
+
+ Mockito.verify(statusMgr).submitFinishQueryToAudit(
+ Mockito.any(AuditEvent.class), Mockito.eq(Set.of(10001L,
10002L)));
+ }
+ }
+
private void assertAuditComputeGroup(BackendSelection.SelectionHint hint,
String expectedComputeGroup,
String deployMode, String cloudComputeGroup) throws Exception {
String oldDeployMode = Config.deploy_mode;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/MasterOpExecutorBackendSelectionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/MasterOpExecutorBackendSelectionTest.java
index 40ed8aa370f..b7f125cb061 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/qe/MasterOpExecutorBackendSelectionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/MasterOpExecutorBackendSelectionTest.java
@@ -33,6 +33,9 @@ import org.junit.Test;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.util.List;
+import java.util.Set;
+
public class MasterOpExecutorBackendSelectionTest {
@After
@@ -126,6 +129,25 @@ public class MasterOpExecutorBackendSelectionTest {
}
}
+ @Test
+ public void testForwardResultExposesExternalDmlAuditBackendIds() throws
Exception {
+ TMasterOpResult result = new TMasterOpResult();
+ result.setMaxJournalId(0L);
+ result.setAuditStatisticsBackendIds(List.of(10001L, 10002L));
+ ConnectContext context = mockConnectContext();
+ Env env = context.getEnv();
+ Mockito.when(env.getSelfNode()).thenReturn(new HostInfo("127.0.0.1",
9010));
+ TestingMasterOpExecutor executor = new
TestingMasterOpExecutor(context, result);
+
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+
+ executor.installForwardResult();
+
+ Assert.assertEquals(Set.of(10001L, 10002L),
executor.getAuditStatisticsBackendIds());
+ }
+ }
+
@Test
public void testDisabledLoadSelectionDoesNotPopulateForwardedInfo() {
ConnectContext context = new ConnectContext();
@@ -212,5 +234,9 @@ public class MasterOpExecutorBackendSelectionTest {
capturedRequest = params;
return forwardResult;
}
+
+ private void installForwardResult() {
+ result = forwardResult;
+ }
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/PipelineExecutionTaskTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/PipelineExecutionTaskTest.java
index 762ffdafd18..83c61861e2c 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/PipelineExecutionTaskTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/PipelineExecutionTaskTest.java
@@ -24,6 +24,7 @@ import
org.apache.doris.nereids.trees.plans.distribute.worker.BackendWorker;
import org.apache.doris.proto.InternalService.PExecPlanFragmentResult;
import org.apache.doris.qe.CoordinatorContext;
import org.apache.doris.rpc.BackendServiceProxy;
+import org.apache.doris.system.Backend;
import org.apache.doris.thrift.TQueryOptions;
import org.apache.doris.thrift.TUniqueId;
@@ -33,6 +34,9 @@ import org.mockito.ArgumentMatchers;
import org.mockito.Mockito;
import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
@@ -52,6 +56,9 @@ class PipelineExecutionTaskTest {
Mockito.when(coordinatorContext.twoPhaseExecution()).thenReturn(false);
MultiFragmentsPipelineTask fragmentsTask =
Mockito.mock(MultiFragmentsPipelineTask.class);
+ Backend backend = Mockito.mock(Backend.class);
+ Mockito.when(backend.getId()).thenReturn(10001L);
+ Mockito.when(fragmentsTask.getBackend()).thenReturn(backend);
Mockito.when(fragmentsTask.getChildrenTasks()).thenReturn(Collections.emptyMap());
Mockito.when(fragmentsTask.sendPhaseOneRpc(false))
.thenReturn(CompletableFuture.completedFuture(PExecPlanFragmentResult.getDefaultInstance()));
@@ -70,6 +77,43 @@ class PipelineExecutionTaskTest {
status -> hasDeadlineTimeoutMessage(status)));
}
+ @Test
+ void auditParticipantsIncludeOnlyAttemptedDispatches() throws Exception {
+ CoordinatorContext coordinatorContext =
Mockito.mock(CoordinatorContext.class);
+ Deencapsulation.setField(coordinatorContext, "timeoutDeadline",
(Supplier<Long>) () -> Long.MAX_VALUE);
+
Mockito.when(coordinatorContext.withLock(ArgumentMatchers.<Callable<Object>>any()))
+ .thenAnswer(invocation ->
invocation.<Callable<Object>>getArgument(0).call());
+ Mockito.when(coordinatorContext.twoPhaseExecution()).thenReturn(false);
+
+ MultiFragmentsPipelineTask first = mockFragmentTask(10001L);
+ MultiFragmentsPipelineTask uncertain = mockFragmentTask(10002L);
+ MultiFragmentsPipelineTask neverAttempted = mockFragmentTask(10003L);
+ Mockito.when(first.sendPhaseOneRpc(false))
+
.thenReturn(CompletableFuture.completedFuture(PExecPlanFragmentResult.getDefaultInstance()));
+ Mockito.when(uncertain.sendPhaseOneRpc(false)).thenThrow(new
IllegalStateException("dispatch failed"));
+ Map<BackendWorker, MultiFragmentsPipelineTask> tasks = new
LinkedHashMap<>();
+ tasks.put(Mockito.mock(BackendWorker.class), first);
+ tasks.put(Mockito.mock(BackendWorker.class), uncertain);
+ tasks.put(Mockito.mock(BackendWorker.class), neverAttempted);
+ PipelineExecutionTask executionTask = new PipelineExecutionTask(
+ coordinatorContext, Mockito.mock(BackendServiceProxy.class),
tasks);
+
+ Assertions.assertThrows(IllegalStateException.class,
executionTask::execute);
+
+ Assertions.assertEquals(Set.of(10001L, 10002L),
+ executionTask.getDispatchedBackendIdsForAudit());
+ Mockito.verify(neverAttempted, Mockito.never()).sendPhaseOneRpc(false);
+ }
+
+ private static MultiFragmentsPipelineTask mockFragmentTask(long backendId)
{
+ MultiFragmentsPipelineTask task =
Mockito.mock(MultiFragmentsPipelineTask.class);
+ Backend backend = Mockito.mock(Backend.class);
+ Mockito.when(backend.getId()).thenReturn(backendId);
+ Mockito.when(task.getBackend()).thenReturn(backend);
+
Mockito.when(task.getChildrenTasks()).thenReturn(Collections.emptyMap());
+ return task;
+ }
+
private static boolean hasDeadlineTimeoutMessage(Status status) {
return status.getErrorMsg().contains("timeout before waiting send
fragments rpc");
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgrTest.java
b/fe/fe-core/src/test/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgrTest.java
index ae17dcc430a..6b58bf996a3 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgrTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgrTest.java
@@ -17,6 +17,11 @@
package org.apache.doris.resource.workloadschedpolicy;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.plugin.AuditEvent;
+import org.apache.doris.qe.AuditEventProcessor;
import org.apache.doris.thrift.TQueryStatistics;
import org.apache.doris.thrift.TQueryStatisticsResult;
import org.apache.doris.thrift.TReportWorkloadRuntimeStatusParams;
@@ -25,8 +30,12 @@ import com.google.common.collect.Maps;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import java.util.List;
import java.util.Map;
+import java.util.Set;
/**
* Unit test for WorkloadRuntimeStatusMgr.
@@ -256,6 +265,129 @@ public class WorkloadRuntimeStatusMgrTest {
Assert.assertEquals(2, merged.get("q1").getFinishedTasksNum());
}
+ @Test
+ public void testExternalDmlAuditWaitsForEveryBackendFinalSnapshot() {
+ int originalAuditTimeout = Config.query_audit_log_timeout_ms;
+ try {
+ Config.query_audit_log_timeout_ms = 10;
+ AuditEvent event = new
AuditEvent.AuditEventBuilder().setQueryId("q1").build();
+ Deencapsulation.invoke(mgr, "submitFinishQueryToAudit", event,
+ Set.of(10001L, 10002L));
+ event.pushToAuditLogQueueTime = System.currentTimeMillis() - 20;
+
+ mgr.updateBeQueryStats(buildParams(10001L, "q1", buildStats(10,
3), true));
+ mgr.updateBeQueryStats(buildParams(10002L, "q1", buildStats(20,
4), false));
+
+ List<AuditEvent> events = Deencapsulation.invoke(mgr,
"getQueryNeedAudit");
+ Assert.assertTrue("external DML audit must wait for every
participating BE", events.isEmpty());
+
+ mgr.updateBeQueryStats(buildParams(10002L, "q1", buildStats(20,
4), true));
+ events = Deencapsulation.invoke(mgr, "getQueryNeedAudit");
+ Assert.assertEquals(1, events.size());
+ Assert.assertSame(event, events.get(0));
+ } finally {
+ Config.query_audit_log_timeout_ms = originalAuditTimeout;
+ }
+ }
+
+ @Test
+ public void testExternalDmlAuditUsesBoundedFallback() {
+ int originalAuditTimeout = Config.query_audit_log_timeout_ms;
+ int originalReportTimeout =
Config.be_report_query_statistics_timeout_ms;
+ try {
+ Config.query_audit_log_timeout_ms = 10;
+ Config.be_report_query_statistics_timeout_ms = 30;
+ AuditEvent event = new
AuditEvent.AuditEventBuilder().setQueryId("q1").build();
+ mgr.submitFinishQueryToAudit(event, Set.of(10001L));
+ event.pushToAuditLogQueueTime = System.currentTimeMillis() - 40;
+
+ List<AuditEvent> events = Deencapsulation.invoke(mgr,
"getQueryNeedAudit");
+ Assert.assertEquals(1, events.size());
+ Assert.assertSame(event, events.get(0));
+ } finally {
+ Config.query_audit_log_timeout_ms = originalAuditTimeout;
+ Config.be_report_query_statistics_timeout_ms =
originalReportTimeout;
+ }
+ }
+
+ @Test
+ public void testExternalDmlAuditUsesFinalCumulativeStatistics() {
+ int originalAuditTimeout = Config.query_audit_log_timeout_ms;
+ try {
+ Config.query_audit_log_timeout_ms = 10;
+ AuditEvent event = new
AuditEvent.AuditEventBuilder().setQueryId("q1").build();
+ mgr.submitFinishQueryToAudit(event, Set.of(10001L));
+ event.pushToAuditLogQueueTime = System.currentTimeMillis() - 20;
+
+ TQueryStatistics finalStatistics = buildStats(10, 10);
+ finalStatistics.setScanRows(100);
+ finalStatistics.setScanBytes(110);
+ finalStatistics.setScanBytesFromLocalStorage(120);
+ finalStatistics.setScanBytesFromRemoteStorage(130);
+ finalStatistics.setCpuMs(20);
+ finalStatistics.setMaxPeakMemoryBytes(30);
+ mgr.updateBeQueryStats(buildParams(10001L, "q1", finalStatistics,
true));
+
+ Env env = Mockito.mock(Env.class);
+ AuditEventProcessor processor =
Mockito.mock(AuditEventProcessor.class);
+ Mockito.when(env.getAuditEventProcessor()).thenReturn(processor);
+ Mockito.when(processor.handleAuditEvent(event)).thenReturn(true);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+
mockedEnv.when(Env::getCurrentAuditEventProcessor).thenReturn(processor);
+ Deencapsulation.invoke(mgr, "runAfterCatalogReady");
+ }
+
+ Mockito.verify(processor).handleAuditEvent(event);
+ Assert.assertEquals(100, event.scanRows);
+ Assert.assertEquals(110, event.scanBytes);
+ Assert.assertEquals(120, event.scanBytesFromLocalStorage);
+ Assert.assertEquals(130, event.scanBytesFromRemoteStorage);
+ Assert.assertEquals(20, event.cpuTimeMs);
+ Assert.assertEquals(30, event.peakMemoryBytes);
+ } finally {
+ Config.query_audit_log_timeout_ms = originalAuditTimeout;
+ }
+ }
+
+ @Test
+ public void testRegularAuditKeepsExistingTimeoutBehavior() {
+ int originalAuditTimeout = Config.query_audit_log_timeout_ms;
+ try {
+ Config.query_audit_log_timeout_ms = 10;
+ AuditEvent event = new
AuditEvent.AuditEventBuilder().setQueryId("q1").build();
+ mgr.submitFinishQueryToAudit(event);
+ event.pushToAuditLogQueueTime = System.currentTimeMillis() - 20;
+
+ List<AuditEvent> events = Deencapsulation.invoke(mgr,
"getQueryNeedAudit");
+ Assert.assertEquals(1, events.size());
+ Assert.assertSame(event, events.get(0));
+ } finally {
+ Config.query_audit_log_timeout_ms = originalAuditTimeout;
+ }
+ }
+
+ @Test
+ public void testAuditScanDoesNotAssumeWallClockInsertionOrder() {
+ int originalAuditTimeout = Config.query_audit_log_timeout_ms;
+ try {
+ Config.query_audit_log_timeout_ms = 10;
+ AuditEvent futureHead = new
AuditEvent.AuditEventBuilder().setQueryId("future").build();
+ AuditEvent dueEvent = new
AuditEvent.AuditEventBuilder().setQueryId("due").build();
+ mgr.submitFinishQueryToAudit(futureHead);
+ mgr.submitFinishQueryToAudit(dueEvent);
+ futureHead.pushToAuditLogQueueTime = System.currentTimeMillis() +
1000;
+ dueEvent.pushToAuditLogQueueTime = System.currentTimeMillis() - 20;
+
+ List<AuditEvent> events = Deencapsulation.invoke(mgr,
"getQueryNeedAudit");
+
+ Assert.assertEquals(1, events.size());
+ Assert.assertSame(dueEvent, events.get(0));
+ } finally {
+ Config.query_audit_log_timeout_ms = originalAuditTimeout;
+ }
+ }
+
// ---- helper methods ----
private TQueryStatistics buildStats(int totalTasks, int finishedTasks) {
@@ -266,8 +398,14 @@ public class WorkloadRuntimeStatusMgrTest {
}
private TReportWorkloadRuntimeStatusParams buildParams(long beId, String
queryId, TQueryStatistics stats) {
+ return buildParams(beId, queryId, stats, false);
+ }
+
+ private TReportWorkloadRuntimeStatusParams buildParams(long beId, String
queryId,
+ TQueryStatistics stats, boolean queryFinished) {
TQueryStatisticsResult result = new TQueryStatisticsResult();
result.setStatistics(stats);
+ result.setQueryFinished(queryFinished);
TReportWorkloadRuntimeStatusParams params = new
TReportWorkloadRuntimeStatusParams();
params.setBackendId(beId);
diff --git a/gensrc/thrift/FrontendService.thrift
b/gensrc/thrift/FrontendService.thrift
index 69a90df46da..9f197d691f8 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -472,6 +472,8 @@ struct TMasterOpResult {
9: optional TTxnLoadInfo txnLoadInfo;
10: optional i64 groupCommitLoadBeId;
11: optional i64 affectedRows;
+ // Lets the forwarding FE wait for the final statistics of external write
fragments.
+ 12: optional list<i64> auditStatisticsBackendIds;
}
// Certificate-based authentication info forwarded from BE to FE
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
