From b38bc4b5126ad5690446ffebbe2ae26c9fe5989e Mon Sep 17 00:00:00 2001 From: Gabriel Date: Fri, 13 Nov 2026 11:36:21 +0800 Subject: [PATCH] [refactor](fe) Unify audit statistics participant tracking ### Background BE runtime statistics arrive asynchronously, so FE audit logging waits for final reports from the backends that received fragment dispatch attempts. The current implementation already supports generic participant tracking but also retains a separate coordinator and executor classification for external writes. ### What problem does this PR solve? Issue Number: None Related PR: #67053 (introduced external-write-specific audit tracking) PR Type: Refactor Problem Summary: Remove the external-DML-specific audit state, accessors, and executor-type checks. Table inserts use their existing shared execution coordinator, and FILES inserts publish their coordinator through the same statement execution state. Audit participants come from dispatched backend IDs and the existing query-scoped failure snapshot. ### Details - **Moves vs. logic changes:** Remove the duplicate external-write coordinator registration and classification helper. Rename the scheduler participant accessor to `getDispatchedBackendIds()` and update its callers and tests. - **Behavior change:** FILES inserts publish the execution coordinator through `StmtExecutor.setCoord()`, sharing the existing audit and cancellation handoff. Participant selection, final-report waiting, timeout fallback, and failure snapshots retain their existing rules. - **Compatibility:** No Thrift field IDs, types, or optionality change; the forwarding-field edit only updates its comment. No storage format changes. - **Performance impact:** Remove an extra retained coordinator reference and an extra participant source. No additional threads, locks, queues, or waits. - **Rollback:** Revert this PR; no migration is required. ### Testing - Locally ran `run-fe-ut.sh --run` for eight suites: `AuditLogHelperBackendSelectionTest`, `MasterOpExecutorBackendSelectionTest`, `PipelineExecutionTaskTest`, `WorkloadRuntimeStatusMgrTest`, `InsertIntoTableCommandTest`, `InsertIntoTVFCommandTest`, `OlapInsertExecutorTest`, and `StreamingInsertTaskAuditTest`: 76 tests passed, with no failures, errors, or skips. - The two new FILES command tests failed before the change because the standard coordinator was absent. Both now pass and cover successful execution, partial dispatch failure, retained participants, and coordinator cleanup. - FE Checkstyle: zero violations. - `git diff --check`: passed. ### Release note None ### Check List (For Author) - Behavior changed: Yes (see Details) - Does this need documentation: No - Pre-submission status: Ready ### AI Disclosure - Tool: Codex (GPT-6) - Generated: Code changes, tests, and this description. - Human review: Pending. --- .../commands/insert/InsertIntoTVFCommand.java | 5 +- .../insert/InsertIntoTableCommand.java | 10 -- .../java/org/apache/doris/qe/Coordinator.java | 4 +- .../org/apache/doris/qe/StmtExecutor.java | 23 +---- .../qe/runtime/PipelineExecutionTask.java | 10 +- .../StreamingInsertTaskAuditTest.java | 1 - .../insert/InsertIntoTVFCommandTest.java | 98 +++++++++++++++++++ .../insert/InsertIntoTableCommandTest.java | 12 --- .../AuditLogHelperBackendSelectionTest.java | 45 ++++----- .../MasterOpExecutorBackendSelectionTest.java | 2 +- .../qe/runtime/PipelineExecutionTaskTest.java | 2 +- .../WorkloadRuntimeStatusMgrTest.java | 8 +- gensrc/thrift/FrontendService.thrift | 2 +- 13 files changed, 138 insertions(+), 84 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTVFCommandTest.java 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 30cd5da46e6218..218baf69e124e0 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 @@ -121,9 +121,8 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { 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); + // Audit and cancellation must observe the same coordinator that dispatches the fragments. + executor.setCoord(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 5a82ecc0dbf8d2..5cd535122c8a1a 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 @@ -439,22 +439,12 @@ public void beforeComplete(AbstractInsertExecutor insertExecutor, StmtExecutor e // We exposed @StmtExecutor#cancel as a unified entry point for statement interruption, // so executeSingleInsert publishes the coordinator at its common execution boundary. insertExecutor.getCoordinator().setTxnId(insertExecutor.getTxnId()); - 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. 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 d76fe8451d7f7a..3cd62f21edd837 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 @@ -427,9 +427,9 @@ public List getInvolvedBackends() { return Utils.fastToImmutableList(coordinatorContext.backends.get().keySet()); } - public Set getDispatchedBackendIdsForAudit() { + public Set getDispatchedBackendIds() { return executionTask == null - ? ImmutableSet.of() : executionTask.getDispatchedBackendIdsForAudit(); + ? ImmutableSet.of() : executionTask.getDispatchedBackendIds(); } public List getFragmentInstanceInfos() { 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 51eb28265ba055..f4072d611207ed 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 @@ -211,7 +211,6 @@ public class StmtExecutor { // Keep this state scoped to the coordinator publication handoff: other execution targets // retain their existing cancellation contracts. private final AtomicReference pendingCoordinatorCancelReason = new AtomicReference<>(); - private volatile Coordinator externalDmlAuditCoordinator = null; // The coordinator may be cleared after a post-dispatch exception or closed before audit. // Pair the snapshot with its query id atomically so a retried attempt cannot inherit // stale BEs even when audit and query retry run on different threads. @@ -577,35 +576,24 @@ public boolean hasForwardedToMaster() { return masterOpExecutor != null; } - public void setExternalDmlAuditCoordinator(Coordinator coordinator) { - externalDmlAuditCoordinator = coordinator; - } - - public Set getExternalDmlAuditBackendIds() { - if (masterOpExecutor != null) { - return masterOpExecutor.getAuditStatisticsBackendIds(); - } - return externalDmlAuditCoordinator == null - ? Collections.emptySet() : externalDmlAuditCoordinator.getDispatchedBackendIdsForAudit(); - } - public Set getAuditStatisticsBackendIds() { if (masterOpExecutor != null) { return masterOpExecutor.getAuditStatisticsBackendIds(); } - Set backendIds = Sets.newHashSet(getExternalDmlAuditBackendIds()); + Set backendIds = Sets.newHashSet(); AuditBackendSnapshot snapshot = dispatchedAuditBackends; if (snapshot != null && Objects.equals(snapshot.queryId, context.queryId())) { backendIds.addAll(snapshot.backendIds); } - if (coord != null) { - backendIds.addAll(coord.getDispatchedBackendIdsForAudit()); + Coordinator currentCoordinator = coord; + if (currentCoordinator != null) { + backendIds.addAll(currentCoordinator.getDispatchedBackendIds()); } return backendIds; } private void rememberDispatchedAuditBackends(Coordinator executedCoordinator) { - Set backendIds = executedCoordinator.getDispatchedBackendIdsForAudit(); + Set backendIds = executedCoordinator.getDispatchedBackendIds(); if (!backendIds.isEmpty()) { dispatchedAuditBackends = new AuditBackendSnapshot(context.queryId().deepCopy(), Set.copyOf(backendIds)); @@ -829,7 +817,6 @@ public void execute(TUniqueId queryId) throws Exception { profile.getSummaryProfile().startQueryAttempt(queryId, TimeUtils.getStartTimeMs()); SessionVariable sessionVariable = context.getSessionVariable(); context.setEffectiveCloudCluster(null); - externalDmlAuditCoordinator = null; // Every attempt starts as the statement did. queryRetry() runs this again after a replan // error without going back through the processor's beforeStatement, and the failed attempt // may have moved the result to the backends (beforeQuery) and registered where. 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 ef5a2df707954b..6d85bc6087f9eb 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 @@ -67,7 +67,7 @@ public class PipelineExecutionTask extends AbstractRuntimeTask dispatchedBackendIdsForAudit = ConcurrentHashMap.newKeySet(); + private final Set dispatchedBackendIds = ConcurrentHashMap.newKeySet(); // mutable states public PipelineExecutionTask( @@ -101,7 +101,7 @@ public void execute() throws Exception { if (!status.ok()) { throw new UserException(status.getErrorMsg()); } - dispatchedBackendIdsForAudit.clear(); + dispatchedBackendIds.clear(); sendAndWaitPhaseOneRpc(); if (coordinatorContext.twoPhaseExecution()) { sendAndWaitPhaseTwoRpc(); @@ -125,7 +125,7 @@ private void sendAndWaitPhaseOneRpc() throws UserException, RpcException { List 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()); + dispatchedBackendIds.add(fragmentsTask.getBackend().getId()); rpcs.add(new RpcInfo( fragmentsTask, DateTime.now().getMillis(), @@ -140,8 +140,8 @@ private void sendAndWaitPhaseOneRpc() throws UserException, RpcException { coordinatorContext.updateProfileIfPresent(profile -> profile.setRpcPhase1Latency(rpcPhase1Latency)); } - public Set getDispatchedBackendIdsForAudit() { - return ImmutableSet.copyOf(dispatchedBackendIdsForAudit); + public Set getDispatchedBackendIds() { + return ImmutableSet.copyOf(dispatchedBackendIds); } private void sendAndWaitPhaseTwoRpc() throws RpcException, UserException { diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java index 6d6f42de911a50..bc822793d2aa0e 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java @@ -360,7 +360,6 @@ private AuditEvent runTask(RuntimeException commandFailure, String sql, new NereidsParser().parseSingle(sql), new StatementContext()); StmtExecutor executor = Mockito.mock(StmtExecutor.class); Mockito.when(executor.getParsedStmt()).thenReturn(parsedStmt); - Mockito.when(executor.getExternalDmlAuditBackendIds()).thenReturn(Collections.emptySet()); Mockito.when(executor.getAuditStatisticsBackendIds()).thenReturn(Collections.emptySet()); Mockito.when(executor.getSummaryProfile()).thenReturn(Mockito.mock(SummaryProfile.class)); ctx.setExecutor(executor); diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTVFCommandTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTVFCommandTest.java new file mode 100644 index 00000000000000..d3e1567ca30c5b --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTVFCommandTest.java @@ -0,0 +1,98 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.nereids.trees.plans.commands.insert; + +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.EnvFactory; +import org.apache.doris.common.Status; +import org.apache.doris.common.UserException; +import org.apache.doris.mysql.privilege.AccessControllerManager; +import org.apache.doris.mysql.privilege.PrivPredicate; +import org.apache.doris.nereids.NereidsPlanner; +import org.apache.doris.nereids.trees.plans.logical.LogicalPlan; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.Coordinator; +import org.apache.doris.qe.QeProcessorImpl; +import org.apache.doris.qe.QueryState.MysqlStateType; +import org.apache.doris.qe.StmtExecutor; +import org.apache.doris.thrift.TQueryOptions; +import org.apache.doris.thrift.TUniqueId; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.MockedConstruction; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.Optional; +import java.util.Set; + +class InsertIntoTVFCommandTest { + @Test + void testSuccessfulExecutionPublishesCoordinator() throws Exception { + assertExecutionPublishesCoordinator(false); + } + + @Test + void testPartialDispatchFailurePublishesCoordinator() throws Exception { + assertExecutionPublishesCoordinator(true); + } + + private void assertExecutionPublishesCoordinator(boolean dispatchFails) throws Exception { + ConnectContext context = new ConnectContext(); + context.setQueryId(new TUniqueId(1, dispatchFails ? 2 : 1)); + context.getSessionVariable().parallelPipelineTaskNum = 1; + StmtExecutor executor = Mockito.spy(new StmtExecutor(context, "insert into files select 1")); + Mockito.doNothing().when(executor).checkBlockRules(); + Coordinator coordinator = Mockito.mock(Coordinator.class); + Mockito.when(coordinator.getQueryOptions()).thenReturn(new TQueryOptions()); + Mockito.when(coordinator.getDispatchedBackendIds()).thenReturn(Set.of(10001L)); + Mockito.when(coordinator.getExecStatus()).thenReturn(Status.OK); + Mockito.when(coordinator.getLoadCounters()).thenReturn(Collections.emptyMap()); + if (dispatchFails) { + Mockito.doThrow(new UserException("fragment dispatch failed")).when(coordinator).exec(); + } + + Env env = Mockito.mock(Env.class); + AccessControllerManager accessManager = Mockito.mock(AccessControllerManager.class); + Mockito.when(env.getAccessManager()).thenReturn(accessManager); + Mockito.when(accessManager.checkGlobalPriv(context, PrivPredicate.ADMIN)).thenReturn(true); + EnvFactory factory = Mockito.mock(EnvFactory.class); + Mockito.when(factory.createCoordinator(Mockito.eq(context), Mockito.any(NereidsPlanner.class), Mockito.any())) + .thenReturn(coordinator); + InsertIntoTVFCommand command = new InsertIntoTVFCommand( + Mockito.mock(LogicalPlan.class), Optional.empty(), Optional.empty()); + + try (MockedStatic envs = Mockito.mockStatic(Env.class); + MockedStatic factories = Mockito.mockStatic(EnvFactory.class); + MockedConstruction planners = Mockito.mockConstruction(NereidsPlanner.class)) { + envs.when(Env::getCurrentEnv).thenReturn(env); + factories.when(EnvFactory::getInstance).thenReturn(factory); + + command.run(context, executor); + + Assertions.assertSame(coordinator, executor.getCoord()); + Assertions.assertEquals(Set.of(10001L), executor.getAuditStatisticsBackendIds()); + Assertions.assertEquals(dispatchFails ? MysqlStateType.ERR : MysqlStateType.OK, + context.getState().getStateType()); + Assertions.assertNull(QeProcessorImpl.INSTANCE.getCoordinator(context.queryId())); + Mockito.verify(coordinator).close(); + } + } +} 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 a96af534591278..5923db01512cc6 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 @@ -218,18 +218,6 @@ void testConnectorSupportsWriteBranchForNonBranchCapableAndNonPluginTables() { "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))); - } - @Test void testInsertIntoTableCommandIsCancelable() { InsertIntoTableCommand command = new InsertIntoTableCommand( 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 6989c1e851632c..029748667f1c41 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 @@ -78,42 +78,42 @@ public void testCloudAuditLogKeepsCloudComputeGroup() throws Exception { } @Test - public void testSuccessfulExternalInsertWaitsForItsCoordinatorBackends() throws Exception { + public void testSuccessfulInsertWaitsForItsCoordinatorBackends() throws Exception { InsertIntoTableCommand command = Mockito.mock(InsertIntoTableCommand.class); Mockito.when(command.stmtType()).thenReturn(StmtType.INSERT); - assertExternalDmlWaitsForCoordinatorBackends(command, true); + assertInsertWaitsForCoordinatorBackends(command, true); } @Test - public void testFailedExternalInsertWaitsForItsCoordinatorBackends() throws Exception { + public void testFailedInsertWaitsForItsCoordinatorBackends() throws Exception { InsertIntoTableCommand command = Mockito.mock(InsertIntoTableCommand.class); Mockito.when(command.stmtType()).thenReturn(StmtType.INSERT); - assertExternalDmlWaitsForCoordinatorBackends(command, false); + assertInsertWaitsForCoordinatorBackends(command, false); } @Test - public void testExternalInsertOverwriteWaitsForItsCoordinatorBackends() throws Exception { + public void testInsertOverwriteWaitsForItsCoordinatorBackends() throws Exception { InsertOverwriteTableCommand command = Mockito.mock(InsertOverwriteTableCommand.class); Mockito.when(command.stmtType()).thenReturn(StmtType.INSERT); - assertExternalDmlWaitsForCoordinatorBackends(command, true); + assertInsertWaitsForCoordinatorBackends(command, true); } @Test public void testFilesInsertWaitsForItsCoordinatorBackends() throws Exception { InsertIntoTVFCommand command = Mockito.mock(InsertIntoTVFCommand.class); Mockito.when(command.stmtType()).thenReturn(StmtType.INSERT); - assertExternalDmlWaitsForCoordinatorBackends(command, true); + assertInsertWaitsForCoordinatorBackends(command, true); } @Test - public void testInternalInsertDoesNotUseExternalDmlBarrier() { - StmtExecutor executor = Mockito.mock(StmtExecutor.class, Mockito.CALLS_REAL_METHODS); + public void testExecutionWithoutCoordinatorHasNoAuditParticipants() { + StmtExecutor executor = new StmtExecutor(new ConnectContext(), "select 1"); Assertions.assertTrue(AuditLogHelper.getAuditStatisticsBackendIds(executor).isEmpty()); } @Test - public void testForwardedExternalDmlUsesBackendIdsReturnedByMaster() { + public void testForwardedExecutionUsesBackendIdsReturnedByMaster() { Set expectedBackendIds = Set.of(10001L, 10002L); MasterOpExecutor masterExecutor = Mockito.mock(MasterOpExecutor.class); Mockito.when(masterExecutor.getAuditStatisticsBackendIds()).thenReturn(expectedBackendIds); @@ -125,31 +125,25 @@ public void testForwardedExternalDmlUsesBackendIdsReturnedByMaster() { } @Test - public void testResolvedExternalDmlUsesOnlyDispatchedBackends() { + public void testExecutionUsesOnlyDispatchedBackends() { Coordinator coordinator = Mockito.mock(Coordinator.class); - Mockito.when(coordinator.getDispatchedBackendIdsForAudit()).thenReturn(Set.of(10001L)); - StmtExecutor executor = Mockito.mock(StmtExecutor.class, Mockito.CALLS_REAL_METHODS); + Mockito.when(coordinator.getDispatchedBackendIds()).thenReturn(Set.of(10001L)); + StmtExecutor executor = new StmtExecutor(new ConnectContext(), "select 1"); - executor.setExternalDmlAuditCoordinator(coordinator); + executor.setCoord(coordinator); - Assertions.assertEquals(Set.of(10001L), executor.getExternalDmlAuditBackendIds()); + Assertions.assertEquals(Set.of(10001L), executor.getAuditStatisticsBackendIds()); } @Test public void testOrdinaryQueryWaitsForEveryDispatchedBackend() { Coordinator coordinator = Mockito.mock(Coordinator.class); - Mockito.when(coordinator.getDispatchedBackendIdsForAudit()).thenReturn(Set.of(10001L, 10002L)); + Mockito.when(coordinator.getDispatchedBackendIds()).thenReturn(Set.of(10001L, 10002L)); StmtExecutor executor = Mockito.mock(StmtExecutor.class, Mockito.CALLS_REAL_METHODS); Deencapsulation.setField(executor, "coord", coordinator); Assertions.assertEquals(Set.of(10001L, 10002L), AuditLogHelper.getAuditStatisticsBackendIds(executor)); - - Coordinator externalCoordinator = Mockito.mock(Coordinator.class); - Mockito.when(externalCoordinator.getDispatchedBackendIdsForAudit()).thenReturn(Set.of(10003L)); - executor.setExternalDmlAuditCoordinator(externalCoordinator); - Assertions.assertEquals(Set.of(10001L, 10002L, 10003L), - AuditLogHelper.getAuditStatisticsBackendIds(executor)); } @Test @@ -158,7 +152,7 @@ public void testPostDispatchErrorKeepsAuditBackendsAfterCoordinatorIsCleared() { context.setQueryId(new TUniqueId(1L, 2L)); StmtExecutor executor = new StmtExecutor(context, "select 1"); Coordinator coordinator = Mockito.mock(Coordinator.class); - Mockito.when(coordinator.getDispatchedBackendIdsForAudit()).thenReturn(Set.of(10001L, 10002L)); + Mockito.when(coordinator.getDispatchedBackendIds()).thenReturn(Set.of(10001L, 10002L)); executor.setCoord(coordinator); // executeAndSendResult's finally snapshots the dispatched BEs even when its exception @@ -184,7 +178,7 @@ public void testPostDispatchErrorWaitsForDelayedFinalRemoteSpillReport() { context.setQueryId(new TUniqueId(5L, 6L)); StmtExecutor executor = new StmtExecutor(context, "select 1"); Coordinator coordinator = Mockito.mock(Coordinator.class); - Mockito.when(coordinator.getDispatchedBackendIdsForAudit()).thenReturn(Set.of(10001L, 10002L)); + Mockito.when(coordinator.getDispatchedBackendIds()).thenReturn(Set.of(10001L, 10002L)); executor.setCoord(coordinator); executor.setCoord(null); Deencapsulation.invoke(executor, "rememberDispatchedAuditBackends", coordinator); @@ -222,7 +216,7 @@ private static TReportWorkloadRuntimeStatusParams finalReport(long backendId, St return params; } - private void assertExternalDmlWaitsForCoordinatorBackends(LogicalPlan command, boolean success) + private void assertInsertWaitsForCoordinatorBackends(LogicalPlan command, boolean success) throws Exception { ConnectContext context = Mockito.spy(new ConnectContext()); context.setStartTime(); @@ -234,7 +228,6 @@ private void assertExternalDmlWaitsForCoordinatorBackends(LogicalPlan command, b } StmtExecutor executor = Mockito.mock(StmtExecutor.class); - Mockito.when(executor.getExternalDmlAuditBackendIds()).thenReturn(Set.of(10001L, 10002L)); Mockito.when(executor.getAuditStatisticsBackendIds()).thenReturn(Set.of(10001L, 10002L)); Mockito.when(executor.getSummaryProfile()).thenReturn(Mockito.mock(SummaryProfile.class)); context.setExecutor(executor); 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 703938e671b22a..5f5ac69b5f78fb 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 @@ -131,7 +131,7 @@ public void testGroupCommitUpdateLoadDataDoesNotWaitJournalReplay() throws Excep } @Test - public void testForwardResultExposesExternalDmlAuditBackendIds() throws Exception { + public void testForwardResultExposesAuditBackendIds() throws Exception { TMasterOpResult result = new TMasterOpResult(); result.setMaxJournalId(0L); result.setAuditStatisticsBackendIds(List.of(10001L, 10002L)); 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 25bc54ad0d6e66..c8298eab03ea8f 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 @@ -104,7 +104,7 @@ void auditParticipantsIncludeOnlyAttemptedDispatches() throws Exception { Assertions.assertThrows(IllegalStateException.class, executionTask::execute); Assertions.assertEquals(Set.of(10001L, 10002L), - executionTask.getDispatchedBackendIdsForAudit()); + executionTask.getDispatchedBackendIds()); Mockito.verify(neverAttempted, Mockito.never()).sendPhaseOneRpc(false); } 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 97ab626f2cda62..bbc73deb3a12f6 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 @@ -286,7 +286,7 @@ public void testSnapshotReadRequiresRebuild() { } @Test - public void testExternalDmlAuditWaitsForEveryBackendFinalSnapshot() { + public void testAuditWaitsForEveryBackendFinalSnapshot() { int originalAuditTimeout = Config.query_audit_log_timeout_ms; try { Config.query_audit_log_timeout_ms = 10; @@ -299,7 +299,7 @@ public void testExternalDmlAuditWaitsForEveryBackendFinalSnapshot() { mgr.updateBeQueryStats(buildParams(10002L, "q1", buildStats(20, 4), false)); List events = Deencapsulation.invoke(mgr, "getQueryNeedAudit"); - Assertions.assertTrue(events.isEmpty(), "external DML audit must wait for every participating BE"); + Assertions.assertTrue(events.isEmpty(), "audit must wait for every participating BE"); mgr.updateBeQueryStats(buildParams(10002L, "q1", buildStats(20, 4), true)); events = Deencapsulation.invoke(mgr, "getQueryNeedAudit"); @@ -353,7 +353,7 @@ public void testOrdinaryQueryAuditWaitsForFinalRemoteSpillTotals() { } @Test - public void testExternalDmlAuditUsesBoundedFallback() { + public void testAuditUsesBoundedFallback() { int originalAuditTimeout = Config.query_audit_log_timeout_ms; int originalReportTimeout = Config.be_report_query_statistics_timeout_ms; try { @@ -373,7 +373,7 @@ public void testExternalDmlAuditUsesBoundedFallback() { } @Test - public void testExternalDmlAuditUsesFinalCumulativeStatistics() { + public void testAuditUsesFinalCumulativeStatistics() { int originalAuditTimeout = Config.query_audit_log_timeout_ms; try { Config.query_audit_log_timeout_ms = 10; diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift index 63250ae7be9454..fdd97c1a954cb0 100644 --- a/gensrc/thrift/FrontendService.thrift +++ b/gensrc/thrift/FrontendService.thrift @@ -486,7 +486,7 @@ 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. + // Lets the forwarding FE wait for the final statistics of dispatched fragments. 12: optional list auditStatisticsBackendIds; // Confirms that the executing FE serialized raw MySQL packets with CLIENT_DEPRECATE_EOF. 13: optional bool clientDeprecatedEofApplied;