Gabriel39 · Nov 13, 2026
@@ -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

@@ -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.

@@ -427,9 +427,9 @@ public List<TNetworkAddress> getInvolvedBackends() {
         return Utils.fastToImmutableList(coordinatorContext.backends.get().keySet());
     }
 
-    public Set<Long> getDispatchedBackendIdsForAudit() {
+    public Set<Long> getDispatchedBackendIds() {
         return executionTask == null
-                ? ImmutableSet.of() : executionTask.getDispatchedBackendIdsForAudit();
+                ? ImmutableSet.of() : executionTask.getDispatchedBackendIds();
     }
 
     public List<FragmentInstanceInfo> getFragmentInstanceInfos() {

@@ -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<Status> 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<Long> getExternalDmlAuditBackendIds() {
-        if (masterOpExecutor != null) {
-            return masterOpExecutor.getAuditStatisticsBackendIds();
-        }
-        return externalDmlAuditCoordinator == null
-                ? Collections.emptySet() : externalDmlAuditCoordinator.getDispatchedBackendIdsForAudit();
-    }
-
     public Set<Long> getAuditStatisticsBackendIds() {
         if (masterOpExecutor != null) {
             return masterOpExecutor.getAuditStatisticsBackendIds();
         }
-        Set<Long> backendIds = Sets.newHashSet(getExternalDmlAuditBackendIds());
+        Set<Long> 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<Long> backendIds = executedCoordinator.getDispatchedBackendIdsForAudit();
+        Set<Long> 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.

@@ -67,7 +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();
+    private final Set<Long> 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<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());
+            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<Long> getDispatchedBackendIdsForAudit() {
-        return ImmutableSet.copyOf(dispatchedBackendIdsForAudit);
+    public Set<Long> getDispatchedBackendIds() {
+        return ImmutableSet.copyOf(dispatchedBackendIds);
     }
 
     private void sendAndWaitPhaseTwoRpc() throws RpcException, UserException {

diff --git a/...t/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java b/...t/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);

@@ -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<Env> envs = Mockito.mockStatic(Env.class);
+                MockedStatic<EnvFactory> factories = Mockito.mockStatic(EnvFactory.class);
+                MockedConstruction<NereidsPlanner> 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();
+        }
+    }
+}
@@ -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(