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;