fix

Author: 924060929

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/AbstractUnassignedScanJob.java

@@ -87,11 +87,12 @@ protected List<AssignedJob> insideMachineParallelization(
             ScanSource scanSource = entry.getValue().scanSource;
 
             // usually, its tablets num, or buckets num
-            int scanSourceMaxParallel = Math.max(scanSource.maxParallel(scanNodes), 1);
+            int scanSourceMaxParallel = scanSource.maxParallel(scanNodes);
 
             // now we should compute how many instances to process the data,
             // for example: two instances
             int instanceNum = degreeOfParallelism(scanSourceMaxParallel, useLocalShuffleToAddParallel);
+
             if (useLocalShuffleToAddParallel) {
                 assignLocalShuffleJobs(scanSource, instanceNum, instances, context, worker);
             } else {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/LocalShuffleAssignedJob.java

@@ -50,7 +50,7 @@ protected Map<String, String> extraInfo() {
     @Override
     protected String formatScanSourceString() {
         if (receiveDataFromLocal) {
-            return "read data from first instance of " + getAssignedWorker();
+            return "read data from other instances";
         } else {
             return super.formatScanSourceString();
         }

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanBucketOlapTableJob.java

@@ -170,26 +170,28 @@ protected List<AssignedJob> insideMachineParallelization(
     protected void assignLocalShuffleJobs(ScanSource scanSource, int instanceNum, List<AssignedJob> instances,
             ConnectContext context, DistributedPlanWorker worker) {
         // only generate one instance to scan all data, in this step
-        List<ScanSource> assignJoinBuckets = scanSource.parallelize(
-                scanNodes, 1
-        );
+        List<ScanSource> assignJoinBuckets = scanSource.parallelize(scanNodes, instanceNum);
 
-        // one scan range generate multiple instances,
-        // different instances reference the same scan source
         int shareScanId = shareScanIdGenerator.getAndIncrement();
-
-        Set<Integer> firstInstanceAssignedJoinBuckets
-                = ((BucketScanSource) assignJoinBuckets.get(0)).bucketIndexToScanNodeToTablets.keySet();
-
         BucketScanSource shareScanSource = (BucketScanSource) scanSource;
-        ScanSource emptyShareScanSource = shareScanSource.newEmpty();
+        for (int i = 0; i < assignJoinBuckets.size(); i++) {
+            BucketScanSource assignedJoinBucket = (BucketScanSource) assignJoinBuckets.get(i);
+            LocalShuffleBucketJoinAssignedJob instance = new LocalShuffleBucketJoinAssignedJob(
+                    instances.size(), shareScanId, false,
+                    context.nextInstanceId(), this, worker,
+                    assignedJoinBucket,
+                    Utils.fastToImmutableSet(assignedJoinBucket.bucketIndexToScanNodeToTablets.keySet())
+            );
+            instances.add(instance);
+        }
 
-        for (int i = 0; i < instanceNum; i++) {
+        ScanSource emptyShareScanSource = shareScanSource.newEmpty();
+        for (int i = assignJoinBuckets.size(); i < instanceNum; i++) {
             LocalShuffleBucketJoinAssignedJob instance = new LocalShuffleBucketJoinAssignedJob(
-                    instances.size(), shareScanId, i > 0,
+                    instances.size(), shareScanId, true,
                     context.nextInstanceId(), this, worker,
-                    i == 0 ? shareScanSource : emptyShareScanSource,
-                    i == 0 ? Utils.fastToImmutableSet(firstInstanceAssignedJoinBuckets) : ImmutableSet.of()
+                    emptyShareScanSource,
+                    ImmutableSet.of()
             );
             instances.add(instance);
         }