update

Author: sollhui

fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java

@@ -662,12 +662,15 @@ public static Map<Long, Map<Long, AdaptiveBucketAssignment>> computeAdaptiveRand
                                     + "sinkBackendId={}, selectedBucketSeqs={}, beToBucketSeqs={}",
                             bucketSeq, partition.getId(), sinkBackendId, selectedBucketSeqs, beToBucketSeqs);
                 }
+                List<Integer> localBucketSeqs = rotateBucketSeqsForStartBucket(
+                        beToBucketSeqs.get(bucketBeId), bucketSeq);
                 assignments.get(sinkBackendId).put(partition.getId(),
-                        new AdaptiveBucketAssignment(bucketBeId, bucketSeq, Collections.singletonList(bucketSeq)));
+                        new AdaptiveBucketAssignment(bucketBeId, bucketSeq, localBucketSeqs));
                 bucketUseCounts.merge(bucketSeq, 1, Integer::sum);
                 if (sinkAssignments != null) {
                     sinkAssignments.put(sinkBackendId,
-                            "bucket=" + bucketSeq + ",bucketBeId=" + bucketBeId);
+                            "bucket=" + bucketSeq + ",bucketBeId=" + bucketBeId
+                                    + ",localBucketSeqs=" + localBucketSeqs);
                 }
             }
             if (sinkAssignments != null) {
@@ -835,6 +838,20 @@ private static int selectLeastUsedBucketSeq(List<Integer> bucketSeqs, Map<Intege
         return selectedBucketSeq;
     }
 
+    private static List<Integer> rotateBucketSeqsForStartBucket(List<Integer> bucketSeqs, int startBucketSeq) {
+        if (bucketSeqs == null || bucketSeqs.isEmpty()) {
+            return Collections.singletonList(startBucketSeq);
+        }
+        int startIdx = bucketSeqs.indexOf(startBucketSeq);
+        Preconditions.checkState(startIdx >= 0,
+                "start bucket %s must exist in bucketSeqs %s", startBucketSeq, bucketSeqs);
+        List<Integer> rotatedBucketSeqs = new ArrayList<>(bucketSeqs.size());
+        for (int offset = 0; offset < bucketSeqs.size(); offset++) {
+            rotatedBucketSeqs.add(bucketSeqs.get((startIdx + offset) % bucketSeqs.size()));
+        }
+        return rotatedBucketSeqs;
+    }
+
     private TOlapTablePartitionParam createPartition(long dbId, OlapTable table)
             throws UserException {
         TOlapTablePartitionParam partitionParam = new TOlapTablePartitionParam();