fix

924060929 · Oct 15, 2024 · 542c3a6 · 542c3a6
1 parent 1b8dd5e
commit 542c3a6
Show file tree

Hide file tree

Showing 4 changed files with 11 additions and 3 deletions.
diff --git a/...core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java b/...core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/DistributePlanner.java
@@ -144,6 +144,7 @@ private void linkPipelinePlan(
         DataSink sink = senderPlan.getFragmentJob().getFragment().getSink();
         if (sink instanceof MultiCastDataSink) {
             MultiCastDataSink multiCastDataSink = (MultiCastDataSink) sink;
+            receiverPlan.getFragmentJob().getFragment().setOutputPartition(multiCastDataSink.getOutputPartition());
             for (DataStreamSink realSink : multiCastDataSink.getDataStreamSinks()) {
                 if (realSink.getExchNodeId() == linkNode.getId()) {
                     senderPlan.addDestinations(realSink, receiverInstances);

diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsSqlCoordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsSqlCoordinator.java
@@ -145,6 +145,10 @@ public QueryProcessor asQueryProcessor() {
         return coordinatorContext.asQueryProcessor();
     }
 
+    public JobProcessor getJobProcessor() {
+        return coordinatorContext.getJobProcessor();
+    }
+
     public LoadProcessor asLoadProcessor() {
         return coordinatorContext.asLoadProcessor();
     }

diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SqlCoordinatorContext.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SqlCoordinatorContext.java
@@ -193,7 +193,8 @@ public synchronized Status readCloneStatus() {
     public synchronized Status updateStatusIfOk(Status newStatus) {
         // If query is done, we will ignore their cancelled updates, and let the remote fragments to clean up async.
         Status originStatus = readCloneStatus();
-        if (coordinator.isEof() && newStatus.isCancelled()) {
+        if (coordinator.getJobProcessor() instanceof QueryProcessor && coordinator.isEof()
+                && newStatus.isCancelled()) {
             return originStatus;
         }
         // nothing to update

diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/ThriftPlansBuilder.java b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/ThriftPlansBuilder.java
@@ -99,8 +99,10 @@ private static Map<DistributedPlanWorker, TPipelineFragmentParamsList> plansToTh
         Multiset<DistributedPlanWorker> workerProcessInstanceNum = computeInstanceNumPerWorker(distributedPlans);
         Map<DistributedPlanWorker, TPipelineFragmentParamsList> fragmentsGroupByWorker = Maps.newLinkedHashMap();
         int currentInstanceIndex = 0;
+        Map<Integer, TFileScanRangeParams> sharedFileScanRangeParams = Maps.newLinkedHashMap();
         for (PipelineDistributedPlan currentFragmentPlan : distributedPlans) {
-            Map<Integer, TFileScanRangeParams> fileScanRangeParams = computeFileScanRangeParams(currentFragmentPlan);
+            sharedFileScanRangeParams.putAll(computeFileScanRangeParams(currentFragmentPlan));
+
             Map<Integer, Integer> exchangeSenderNum = computeExchangeSenderNum(currentFragmentPlan);
             Map<DistributedPlanWorker, TPipelineFragmentParams> workerToCurrentFragment = Maps.newLinkedHashMap();
 
@@ -110,7 +112,7 @@ private static Map<DistributedPlanWorker, TPipelineFragmentParamsList> plansToTh
                 AssignedJob instanceJob = currentFragmentPlan.getInstanceJobs().get(instanceNumInCurrentFragment);
                 TPipelineFragmentParams currentFragmentParam = fragmentToThriftIfAbsent(
                         currentFragmentPlan, instanceJob, workerToCurrentFragment,
-                        exchangeSenderNum, fileScanRangeParams,
+                        exchangeSenderNum, sharedFileScanRangeParams,
                         workerProcessInstanceNum, coordinatorContext);
 
                 TPipelineInstanceParams instanceParam = instanceToThrift(