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

The position-fallback tabletId->bucketSeq fill policy was implemented twice:
OlapScanNode.fillTabletId2BucketSeq (early-return form) and inline in
PlanFragmentBuilder.visitPhysicalOlapScan (nested if/else form). Two independent
producers of the same policy can silently drift.

Collapse both into a single static helper
RangeColocateScanDispatch.fillTabletId2BucketSeq that writes into a
caller-supplied target map, and route both call sites through it.
RangeColocateScanDispatch already owns computeBucketSeq and the sibling
range-colocate scan helpers, so it is the natural home.

This is a behavior-preserving extraction: the helper reproduces both sites
exactly for dispatch absent, dispatch present + aligned, and dispatch present +
transiently unaligned. Add focused unit tests for the three cases.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: xiangguangyxg <xiangguangyxg@gmail.com>

Author: xiangguangyxg

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

@@ -885,40 +885,15 @@ private void computeTabletInfo() throws StarRocksException {
                     scanTabletIds.addAll(allTabletIds);
                 }
 
-                fillTabletId2BucketSeq(dispatch, selectedIndex, allTabletIds);
+                RangeColocateScanDispatch.fillTabletId2BucketSeq(
+                        dispatch, selectedIndex, allTabletIds, tabletId2BucketSeq);
                 totalTabletsNum += selectedIndex.getTablets().size();
                 selectedTabletsNum += tablets.size();
                 addScanRangeLocations(partition, physicalPartition, selectedIndex, tablets, List.of(), localBeId);
             }
         }
     }
 
-    /**
-     * 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.
-     */
-    private void fillTabletId2BucketSeq(@Nullable RangeColocateScanDispatch dispatch,
-                                          MaterializedIndex selectedIndex,
-                                          List<Long> allTabletIds) {
-        if (dispatch != null) {
-            Map<Long, Integer> rangeColocateMap = dispatch.computeBucketSeq(selectedIndex);
-            if (rangeColocateMap != null) {
-                tabletId2BucketSeq.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);
-        }
-    }
-
     public void checkIfScanRangeNumSafe(long scanRangeSize) {
         long totalPartitionNum = 0;
         long totalTabletsNum = 0;

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

@@ -334,4 +334,40 @@ public Map<Long, Integer> computeBucketSeq(MaterializedIndex selectedIndex) {
         }
         return output;
     }
+
+    /**
+     * Fills {@code target} with the tablet-id → bucket-sequence assignment for
+     * {@code selectedIndex}, applying the position-fallback policy shared by every
+     * scan-time bucketSeq producer ({@link OlapScanNode} and
+     * {@link com.starrocks.sql.plan.PlanFragmentBuilder}):
+     *
+     * <ul>
+     *   <li>range colocate and aligned — {@code dispatch != null} and
+     *       {@link #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 OlapScanNode#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.
+     */
+    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) {
+                target.putAll(rangeColocateMap);
+                return;
+            }
+        }
+        for (int i = 0; i < allTabletIds.size(); 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);
-                            }
-                        }
+                        RangeColocateScanDispatch.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,78 @@ public void testRequireAlignedOnUnalignedGroupThrows() throws Exception {
         Assertions.assertTrue(exception.getMessage().contains("unaligned state"),
                 "actual: " + exception.getMessage());
     }
+
+    // ---- fillTabletId2BucketSeq facade (shared by OlapScanNode + 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<>();
+
+        RangeColocateScanDispatch.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<>();
+        RangeColocateScanDispatch.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<>();
+        RangeColocateScanDispatch.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);
+    }
 }