fix bucket_seq_to_instance_index

924060929 · Oct 17, 2024 · 304a8fc · 304a8fc
1 parent 775c14c
commit 304a8fc
Showing 1 changed file with 9 additions and 5 deletions.
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
@@ -472,12 +472,16 @@ private static Map<Integer, Integer> computeBucketIdToInstanceId(PipelineDistrib
         }
 
         Map<Integer, Integer> bucketIdToInstanceId = Maps.newLinkedHashMap();
-        filterInstancesWhichReceiveDataFromRemote(receivePlan, (instanceJob, instanceIdInThisBackend) -> {
-            BucketScanSource scanSource = (BucketScanSource) instanceJob.getScanSource();
-            for (Integer bucketIndex : scanSource.bucketIndexToScanNodeToTablets.keySet()) {
-                bucketIdToInstanceId.put(bucketIndex, instanceIdInThisBackend);
+        for (AssignedJob instanceJob : instanceJobs) {
+            if (instanceJob instanceof LocalShuffleAssignedJob
+                    && ((LocalShuffleAssignedJob) instanceJob).receiveDataFromLocal) {
+                continue;
             }
-        });
+            BucketScanSource bucketScanSource = (BucketScanSource) instanceJob.getScanSource();
+            for (Integer bucketIndex : bucketScanSource.bucketIndexToScanNodeToTablets.keySet()) {
+                bucketIdToInstanceId.put(bucketIndex, instanceJob.indexInUnassignedJob());
+            }
+        }
         return bucketIdToInstanceId;
     }