[Refactor] Unify position-fallback tabletId-to-bucketSeq fill into one facade helper

fe/fe-core/src/main/java/com/starrocks/planner/OlapScanNode.java

@@ -885,7 +885,7 @@ private void computeTabletInfo() throws StarRocksException {
                     scanTabletIds.addAll(allTabletIds);
                 }
 
-                fillTabletId2BucketSeq(dispatch, selectedIndex, allTabletIds);
+                fillTabletId2BucketSeq(dispatch, selectedIndex, allTabletIds, tabletId2BucketSeq);
                 totalTabletsNum += selectedIndex.getTablets().size();
                 selectedTabletsNum += tablets.size();
                 addScanRangeLocations(partition, physicalPartition, selectedIndex, tablets, List.of(), localBeId);
@@ -894,28 +894,36 @@ private void computeTabletInfo() throws StarRocksException {
     }
 
     /**
-     * Populates {@link #tabletId2BucketSeq} for one {@link MaterializedIndex}.
-     * Range-colocate scans use the bucket sequence supplied by the dispatch
-     * facade when alignment holds; everything else (HASH, range non-colocate,
-     * or transiently unaligned range colocate) falls back to position-based
-     * bucketSeq. The dispatch-time alignment guard fires later in
-     * {@link #getBucketNums()} for the colocate-dispatch path.
+     * Fills {@code target} with the tablet-id → bucket-sequence assignment for {@code selectedIndex},
+     * applying the position-fallback policy shared by both scan-time producers of
+     * {@link #tabletId2BucketSeq} — this class's {@link #computeTabletInfo} and
+     * {@link com.starrocks.sql.plan.PlanFragmentBuilder}:
+     *
+     * <ul>
+     *   <li>range colocate and aligned — {@code dispatch != null} and
+     *       {@link RangeColocateScanDispatch#computeBucketSeq} returns a mapping — uses that mapping;</li>
+     *   <li>everything else (HASH, range non-colocate, or a transiently unaligned range colocate group
+     *       where {@code computeBucketSeq} returns {@code null}) falls back to position-based bucketSeq.</li>
+     * </ul>
+     *
+     * <p>Entries are added to {@code target} without clearing it, so a caller can accumulate the
+     * whole-scan assignment across sub-partitions. The dispatch-time alignment guard in
+     * {@link #getBucketNums()} fails closed on the colocate-dispatch path when the built assignment is a
+     * stale position fallback, so no unaligned observation needs to be recorded here.
      */
-    private void fillTabletId2BucketSeq(@Nullable RangeColocateScanDispatch dispatch,
-                                          MaterializedIndex selectedIndex,
-                                          List<Long> allTabletIds) {
+    public static void fillTabletId2BucketSeq(@Nullable RangeColocateScanDispatch dispatch,
+                                              MaterializedIndex selectedIndex,
+                                              List<Long> allTabletIds,
+                                              Map<Long, Integer> target) {
         if (dispatch != null) {
             Map<Long, Integer> rangeColocateMap = dispatch.computeBucketSeq(selectedIndex);
             if (rangeColocateMap != null) {
-                tabletId2BucketSeq.putAll(rangeColocateMap);
+                target.putAll(rangeColocateMap);
                 return;
             }
-            // Range-colocate but transiently unaligned: fall back to position-based bucketSeq so a
-            // non-colocate scan of this table still works. getBucketNums() fails closed on the
-            // colocate-dispatch path by validating this built assignment against the aligned mapping.
         }
         for (int i = 0; i < allTabletIds.size(); i++) {
-            tabletId2BucketSeq.put(allTabletIds.get(i), i);
+            target.put(allTabletIds.get(i), i);
         }
     }
 

fe/fe-core/src/main/java/com/starrocks/sql/plan/PlanFragmentBuilder.java

@@ -1016,30 +1016,8 @@ public PlanFragment visitPhysicalOlapScan(OptExpression optExpr, ExecPlan contex
                         final MaterializedIndex selectedIndex = physicalPartition.getLatestIndex(selectedIndexMetaId);
                         totalTabletsNum += selectedIndex.getTablets().size();
                         List<Long> allTabletIds = selectedIndex.getTabletIdsInOrder();
-                        // Range-colocate scans use the dispatch facade when alignment
-                        // holds; if alignment is transiently broken the same scan still
-                        // needs a bucketSeq for non-colocate consumers, so fall back to
-                        // position-based and let the dispatch-time alignment guard in
-                        // OlapScanNode.getBucketNums() handle the colocate-dispatch path.
-                        if (dispatch != null) {
-                            Map<Long, Integer> rangeColocateMap = dispatch.computeBucketSeq(selectedIndex);
-                            if (rangeColocateMap != null) {
-                                tabletId2BucketSeq.putAll(rangeColocateMap);
-                            } else {
-                                // Range-colocate but transiently unaligned: fall back to position-based
-                                // bucketSeq so a non-colocate scan still works. getBucketNums() fails closed
-                                // on the colocate-dispatch path by validating this built assignment against
-                                // the aligned mapping, so no unaligned observation needs to be recorded here.
-                                for (int i = 0; i < allTabletIds.size(); i++) {
-                                    tabletId2BucketSeq.put(allTabletIds.get(i), i);
-                                }
-                            }
-                        } else {
-                            // HASH or range non-colocate: position-based bucketSeq.
-                            for (int i = 0; i < allTabletIds.size(); i++) {
-                                tabletId2BucketSeq.put(allTabletIds.get(i), i);
-                            }
-                        }
+                        OlapScanNode.fillTabletId2BucketSeq(
+                                dispatch, selectedIndex, allTabletIds, tabletId2BucketSeq);
                         List<Tablet> tablets =
                                 selectTabletIds.stream().map(selectedIndex::getTablet).collect(Collectors.toList());
                         scanNode.addScanRangeLocations(partition, physicalPartition, selectedIndex, tablets, partitionRange,

fe/fe-core/src/test/java/com/starrocks/planner/RangeColocateScanDispatchTest.java

@@ -40,6 +40,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -328,4 +329,79 @@ public void testRequireAlignedOnUnalignedGroupThrows() throws Exception {
         Assertions.assertTrue(exception.getMessage().contains("unaligned state"),
                 "actual: " + exception.getMessage());
     }
+
+    // ---- OlapScanNode.fillTabletId2BucketSeq: the position-fallback fill policy over this dispatch,
+    //      shared by OlapScanNode.computeTabletInfo + PlanFragmentBuilder ----
+
+    @Test
+    public void testFillTabletId2BucketSeqNullDispatchUsesPositionBased() {
+        // dispatch == null (HASH or range non-colocate): position-based fill, keyed by scan order.
+        MaterializedIndex index = new MaterializedIndex(50L);
+        List<Long> allTabletIds = Arrays.asList(101L, 202L, 303L);
+        Map<Long, Integer> target = new HashMap<>();
+
+        OlapScanNode.fillTabletId2BucketSeq(null, index, allTabletIds, target);
+
+        Assertions.assertEquals(Map.of(101L, 0, 202L, 1, 303L, 2), target);
+    }
+
+    @Test
+    public void testFillTabletId2BucketSeqAlignedUsesDispatchMap() throws Exception {
+        starRocksAssert.withTable(
+                "create table t_facade_fill_aligned (k1 int, k2 int)\n"
+                        + "order by(k1, k2)\n"
+                        + "properties('replication_num' = '1', 'colocate_with' = 'rg_facade_fill_aligned:k1');");
+
+        OlapTable table = (OlapTable) GlobalStateMgr.getCurrentState().getLocalMetastore()
+                .getTable(db.getFullName(), "t_facade_fill_aligned");
+        MaterializedIndex index = baseIndexOf(table);
+
+        RangeColocateScanDispatch dispatch =
+                Objects.requireNonNull(RangeColocateScanDispatch.forTable(table));
+        Map<Long, Integer> aligned = dispatch.computeBucketSeq(index);
+        Assertions.assertNotNull(aligned, "precondition: group must be aligned");
+
+        // Aligned range colocate: the facade uses the dispatch mapping verbatim.
+        Map<Long, Integer> target = new HashMap<>();
+        OlapScanNode.fillTabletId2BucketSeq(
+                dispatch, index, index.getTabletIdsInOrder(), target);
+
+        Assertions.assertEquals(aligned, target);
+    }
+
+    @Test
+    public void testFillTabletId2BucketSeqUnalignedFallsBackToPositionBased() throws Exception {
+        starRocksAssert.withTable(
+                "create table t_facade_fill_unaligned (k1 int, k2 int)\n"
+                        + "order by(k1, k2)\n"
+                        + "properties('replication_num' = '1', 'colocate_with' = 'rg_facade_fill_unaligned:k1');");
+
+        OlapTable table = (OlapTable) GlobalStateMgr.getCurrentState().getLocalMetastore()
+                .getTable(db.getFullName(), "t_facade_fill_unaligned");
+        ColocateTableIndex colocateTableIndex = GlobalStateMgr.getCurrentState().getColocateTableIndex();
+        long groupId = colocateTableIndex.getGroup(table.getId()).grpId;
+
+        // Inject 3 ranges without re-aligning the single base tablet that still covers
+        // Range.all() -> computeBucketSeq returns null (transiently unaligned, spanning-tablet state).
+        colocateTableIndex.getColocateRangeMgr().setColocateRanges(groupId, Arrays.asList(
+                new ColocateRange(Range.lt(makeTuple(100)), 9001L),
+                new ColocateRange(Range.gelt(makeTuple(100), makeTuple(200)), 9002L),
+                new ColocateRange(Range.ge(makeTuple(200)), 9003L)));
+
+        RangeColocateScanDispatch dispatch =
+                Objects.requireNonNull(RangeColocateScanDispatch.forTable(table));
+        MaterializedIndex index = baseIndexOf(table);
+        Assertions.assertNull(dispatch.computeBucketSeq(index), "precondition: group must be unaligned");
+
+        // Unaligned range colocate: the facade falls back to position-based bucketSeq.
+        List<Long> allTabletIds = index.getTabletIdsInOrder();
+        Map<Long, Integer> target = new HashMap<>();
+        OlapScanNode.fillTabletId2BucketSeq(dispatch, index, allTabletIds, target);
+
+        Map<Long, Integer> expected = new HashMap<>();
+        for (int i = 0; i < allTabletIds.size(); i++) {
+            expected.put(allTabletIds.get(i), i);
+        }
+        Assertions.assertEquals(expected, target);
+    }
 }