apache · morningman · Jul 27, 2026 · Jul 10, 2026
@@ -3649,6 +3649,12 @@ public static int metaServiceRpcRetryTimes() {
             description = { "存算分离模式下，一个 BE 挂掉多长时间后，它的 tablet 彻底转移到其他 BE 上" })
     public static int rehash_tablet_after_be_dead_seconds = 3600;
 
+    @ConfField(mutable = false, masterOnly = true,
+            description = {
+                    "Whether to use rendezvous hashing for colocate bucket placement in cloud mode. "
+                            + "If false, use the legacy modulo placement. Restart-only."})
+    public static boolean enable_cloud_colocate_consistent_hash = true;
+
     @ConfField(mutable = true, description = {"存算分离模式下是否启用自动启停功能，默认 true",
         "Whether to enable the automatic start-stop feature in cloud model, default is true."})
     public static boolean enable_auto_start_for_cloud_cluster = true;

@@ -346,6 +346,9 @@ public boolean removeTable(long tableId) {
                 group2Schema.remove(groupId);
                 group2ErrMsgs.remove(groupId);
                 unstableGroups.remove(groupId);
+                if (Config.isCloudMode()) {
+                    Env.getCurrentSystemInfo().invalidateCloudColocatePlacement(groupId);
+                }
                 String fullGroupName = null;
                 for (Map.Entry<String, GroupId> entry : groupName2Id.entrySet()) {
                     if (entry.getValue().equals(groupId)) {

@@ -0,0 +1,64 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.cloud.catalog;
+
+import com.google.common.base.Preconditions;
+import com.google.common.hash.Hashing;
+
+import java.util.Arrays;
+
+public class CloudColocatePlacement {
+    @FunctionalInterface
+    interface ScoreFunction {
+        long score(long grpId, long idx, long beId);
+    }
+
+    private CloudColocatePlacement() {
+    }
+
+    public static long score(long grpId, long idx, long beId) {
+        return Hashing.murmur3_128().newHasher()
+                .putLong(grpId)
+                .putLong(idx)
+                .putLong(beId)
+                .hash()
+                .asLong();
+    }
+
+    public static long pickBackendId(long grpId, long idx, long[] candidateBeIds) {
+        return pickBackendId(grpId, idx, candidateBeIds, CloudColocatePlacement::score);
+    }
+
+    static long pickBackendId(long grpId, long idx, long[] candidateBeIds, ScoreFunction scoreFunction) {
+        Preconditions.checkArgument(candidateBeIds.length > 0);
+        long[] sortedBeIds = Arrays.copyOf(candidateBeIds, candidateBeIds.length);
+        Arrays.sort(sortedBeIds);
+
+        long pickedBeId = sortedBeIds[0];
+        long maxScore = scoreFunction.score(grpId, idx, pickedBeId);
+        for (int i = 1; i < sortedBeIds.length; i++) {
+            long beId = sortedBeIds[i];
+            long score = scoreFunction.score(grpId, idx, beId);
+            if (score > maxScore || (score == maxScore && beId < pickedBeId)) {
+                maxScore = score;
+                pickedBeId = beId;
+            }
+        }
+        return pickedBeId;
+    }
+}
@@ -114,6 +114,16 @@ private boolean isColocated() {
         return Env.getCurrentColocateIndex().isColocateTableNoLock(tableId);
     }
 
+    private boolean isDecommissioningOrDecommissioned(Backend be) {
+        boolean decommissioning = be.isDecommissioning();
+        boolean decommissioned = be.isDecommissioned();
+        if ((decommissioning || decommissioned) && LOG.isDebugEnabled()) {
+            LOG.debug("backend {} is filtered by decommission state, decommissioning={}, decommissioned={}, "
+                    + "backend={}", be.getId(), decommissioning, decommissioned, be);
+        }
+        return decommissioning || decommissioned;
+    }
+
     public long getColocatedBeId(String clusterId) throws ComputeGroupException {
         CloudSystemInfoService infoService = ((CloudSystemInfoService) Env.getCurrentSystemInfo());
         List<Backend> bes = infoService.getBackendsByClusterId(clusterId).stream()
@@ -148,28 +158,54 @@ public long getColocatedBeId(String clusterId) throws ComputeGroupException {
         }
 
         GroupId groupId = Env.getCurrentColocateIndex().getGroupNoLock(tableId);
-        HashCode hashCode = Hashing.murmur3_128().hashLong(groupId.grpId);
         if (availableBes.size() != bes.size()) {
-            // some be is dead recently, still hash tablets on all backends.
             long needRehashDeadTime = System.currentTimeMillis() - Config.rehash_tablet_after_be_dead_seconds * 1000L;
             if (bes.stream().anyMatch(be -> !be.isAlive() && be.getLastUpdateMs() > needRehashDeadTime)) {
                 List<Backend> beAliveOrDeadShort = bes.stream()
                         .filter(be -> be.isAlive() || be.getLastUpdateMs() > needRehashDeadTime)
                         .collect(Collectors.toList());
-                long index = getIndexByBeNum(hashCode.asLong() + idx, beAliveOrDeadShort.size());
-                Backend be = beAliveOrDeadShort.get((int) index);
-                if (be.isAlive() && !be.isDecommissioned()) {
+                Backend be = pickColocatedBackendForDeadGrace(infoService, groupId, clusterId, beAliveOrDeadShort);
+                if (be.isAlive() && !isDecommissioningOrDecommissioned(be)) {
                     return be.getId();
                 }
             }
         }
 
-        // Tablets with the same idx will be hashed to the same BE, which
-        // meets the requirements of colocated table.
+        return pickColocatedBackend(infoService, groupId, clusterId, availableBes).getId();
+    }
+
+    private Backend pickColocatedBackendForDeadGrace(CloudSystemInfoService infoService, GroupId groupId,
+            String clusterId, List<Backend> availableBes) {
+        if (!Config.enable_cloud_colocate_consistent_hash) {
+            return pickColocatedBackend(infoService, groupId, clusterId, availableBes);
+        }
+        int bucketNum = infoService.getCloudColocateBucketsNum(groupId);
+        CloudSystemInfoService.checkCloudColocateBucketIdx(groupId, clusterId, idx, bucketNum);
+        long[] availableBeIds = availableBes.stream().mapToLong(Backend::getId).toArray();
+        long pickedBeId = CloudColocatePlacement.pickBackendId(groupId.grpId, idx, availableBeIds);
+        return findPickedBackend(pickedBeId, groupId, clusterId, availableBes);
+    }
+
+    Backend pickColocatedBackend(CloudSystemInfoService infoService, GroupId groupId, String clusterId,
+            List<Backend> availableBes) {
+        if (Config.enable_cloud_colocate_consistent_hash) {
+            List<Long> availableBeIds = availableBes.stream().map(Backend::getId).collect(Collectors.toList());
+            long pickedBeId = infoService.getCloudColocateHrwBeId(groupId, clusterId, availableBeIds, idx);
+            return findPickedBackend(pickedBeId, groupId, clusterId, availableBes);
+        }
+
+        HashCode hashCode = Hashing.murmur3_128().hashLong(groupId.grpId);
         long index = getIndexByBeNum(hashCode.asLong() + idx, availableBes.size());
-        long pickedBeId = availableBes.get((int) index).getId();
+        return availableBes.get((int) index);
+    }
 
-        return pickedBeId;
+    private Backend findPickedBackend(long pickedBeId, GroupId groupId, String clusterId, List<Backend> availableBes) {
+        return availableBes.stream().filter(be -> be.getId() == pickedBeId).findFirst()
+                .orElseThrow(() -> new IllegalStateException(String.format(
+                        "picked colocate backend %s is not in candidate set, group %s, cluster %s, bucket idx %s, "
+                                + "candidate backend ids %s",
+                        pickedBeId, groupId, clusterId, idx,
+                        availableBes.stream().map(Backend::getId).collect(Collectors.toList()))));
     }
 
     @Override

@@ -20,8 +20,11 @@
 import org.apache.doris.analysis.ModifyBackendClause;
 import org.apache.doris.analysis.ModifyBackendHostNameClause;
 import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.ColocateGroupSchema;
+import org.apache.doris.catalog.ColocateTableIndex.GroupId;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.ReplicaAllocation;
+import org.apache.doris.cloud.catalog.CloudColocatePlacement;
 import org.apache.doris.cloud.catalog.CloudEnv;
 import org.apache.doris.cloud.catalog.ComputeGroup;
 import org.apache.doris.cloud.proto.Cloud;
@@ -53,6 +56,7 @@
 import org.apache.doris.system.SystemInfoService;
 import org.apache.doris.thrift.TStorageMedium;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableMap;
@@ -69,6 +73,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -99,8 +104,145 @@ public class CloudSystemInfoService extends SystemInfoService {
     // clusterId -> ComputeGroup
     protected Map<String, ComputeGroup> computeGroupIdToComputeGroup = new ConcurrentHashMap<>();
 
+    private final Map<ColocatePlacementKey, ColocatePlacementCache> colocatePlacementCache =
+            new ConcurrentHashMap<>();
+
     private InstanceInfoPB.Status instanceStatus;
 
+    public long getCloudColocateHrwBeId(GroupId groupId, String clusterId, List<Long> availableBeIds, long idx) {
+        return getCloudColocateHrwBeIdInternal(groupId, clusterId, availableBeIds, idx, -1);
+    }
+
+    @VisibleForTesting
+    public long getCloudColocateHrwBeIdForTest(GroupId groupId, String clusterId, List<Long> availableBeIds,
+            long idx, int bucketNumForTest) {
+        return getCloudColocateHrwBeIdInternal(groupId, clusterId, availableBeIds, idx, bucketNumForTest);
+    }
+
+    private long getCloudColocateHrwBeIdInternal(GroupId groupId, String clusterId, List<Long> availableBeIds,
+            long idx, int bucketNumForTest) {
+        long[] candidateBeIds = availableBeIds.stream().mapToLong(Long::longValue).toArray();
+        ColocatePlacementKey key = new ColocatePlacementKey(groupId, clusterId);
+        long fingerprint = fingerprintBackendIds(candidateBeIds);
+        ColocatePlacementCache cache = colocatePlacementCache.get(key);
+        if (cache != null && cache.same(fingerprint)) {
+            checkCloudColocateBucketIdx(groupId, clusterId, idx, cache.bucketNum);
+            return cache.beIdByBucket[(int) idx];
+        }
+
+        // Resolve bucketNum BEFORE compute(): getColocateBucketsNum acquires the colocate-index
+        // read lock, while removeTable() evicts this cache holding the colocate-index write lock.
+        // Acquiring the colocate lock inside the ConcurrentHashMap compute() bin lock would invert
+        // that order and risk an ABBA deadlock, so the locked fetch must stay outside compute().
+        int bucketNum = bucketNumForTest > 0 ? bucketNumForTest : getColocateBucketsNum(groupId);
+        cache = colocatePlacementCache.compute(key, (ignored, oldCache) -> {
+            if (oldCache != null && oldCache.same(fingerprint, bucketNum)) {
+                return oldCache;
+            }
+            return ColocatePlacementCache.build(fingerprint, candidateBeIds, groupId.grpId, bucketNum);
+        });
+        checkCloudColocateBucketIdx(groupId, clusterId, idx, cache.bucketNum);
+        return cache.beIdByBucket[(int) idx];
+    }
+
+    private static long fingerprintBackendIds(long[] beIds) {
+        long sum = 0;
+        for (long beId : beIds) {
+            sum += mix64(beId);
+        }
+        return mix64(sum) ^ mix64(beIds.length);
+    }
+
+    private static long mix64(long value) {
+        value ^= value >>> 33;
+        value *= 0xff51afd7ed558ccdL;
+        value ^= value >>> 33;
+        value *= 0xc4ceb9fe1a85ec53L;
+        value ^= value >>> 33;
+        return value;
+    }
+
+    private static int getColocateBucketsNum(GroupId groupId) {
+        ColocateGroupSchema groupSchema = Env.getCurrentColocateIndex().getGroupSchema(groupId);
+        Preconditions.checkState(groupSchema != null, "missing colocate group schema for group %s", groupId);
+        return groupSchema.getBucketsNum();
+    }
+
+    public int getCloudColocateBucketsNum(GroupId groupId) {
+        return getColocateBucketsNum(groupId);
+    }
+
+    public static void checkCloudColocateBucketIdx(GroupId groupId, String clusterId, long idx, int bucketNum) {
+        if (idx < 0 || idx >= bucketNum) {
+            throw new IllegalStateException(String.format(
+                    "colocate bucket idx %s is outside bucket num %s for group %s, cluster %s",
+                    idx, bucketNum, groupId, clusterId));
+        }
+    }
+
+    @Override
+    public void invalidateCloudColocatePlacement(GroupId groupId) {
+        colocatePlacementCache.keySet().removeIf(key -> key.groupId.equals(groupId));
+    }
+
+    public void invalidateCloudColocatePlacement(String clusterId) {
+        colocatePlacementCache.keySet().removeIf(key -> key.clusterId.equals(clusterId));
+    }
+
+    private static class ColocatePlacementKey {
+        private final GroupId groupId;
+        private final String clusterId;
+
+        private ColocatePlacementKey(GroupId groupId, String clusterId) {
+            this.groupId = groupId;
+            this.clusterId = clusterId;
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (!(obj instanceof ColocatePlacementKey)) {
+                return false;
+            }
+            ColocatePlacementKey other = (ColocatePlacementKey) obj;
+            return groupId.equals(other.groupId) && clusterId.equals(other.clusterId);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(groupId, clusterId);
+        }
+    }
+
+    private static class ColocatePlacementCache {
+        private final long fingerprint;
+        private final int bucketNum;
+        private final long[] beIdByBucket;
+
+        private ColocatePlacementCache(long fingerprint, int bucketNum, long[] beIdByBucket) {
+            this.fingerprint = fingerprint;
+            this.bucketNum = bucketNum;
+            this.beIdByBucket = beIdByBucket;
+        }
+
+        private static ColocatePlacementCache build(long fingerprint, long[] candidateBeIds, long grpId,
+                int bucketNum) {
+            long[] beIdByBucket = new long[bucketNum];
+            for (int i = 0; i < bucketNum; i++) {
+                beIdByBucket[i] = CloudColocatePlacement.pickBackendId(grpId, i, candidateBeIds);
+            }
+            return new ColocatePlacementCache(fingerprint, bucketNum, beIdByBucket);
+        }
+
+        private boolean same(long otherFingerprint) {
+            return fingerprint == otherFingerprint;
+        }
+
+        private boolean same(long otherFingerprint, int otherBucketNum) {
+            return fingerprint == otherFingerprint && bucketNum == otherBucketNum;
+        }
+
+    }
+
     public void addVirtualClusterInfoToMapsNoLock(String clusterId, String clusterName) {
         LOG.info("add virtual cluster info to maps, clusterId={}, clusterName={}", clusterId, clusterName);
         clusterNameToId.put(clusterName, clusterId);
@@ -286,6 +428,7 @@ public void removeComputeGroup(String computeGroupId, String computeGroupName) {
             wlock.lock();
             computeGroupIdToComputeGroup.remove(computeGroupId);
             removeVirtualClusterInfoFromMapsNoLock(computeGroupId, computeGroupName);
+            invalidateCloudColocatePlacement(computeGroupId);
         } finally {
             wlock.unlock();
         }
@@ -1119,6 +1262,7 @@ public void dropCluster(final String clusterId, final String clusterName) {
         try {
             clusterNameToId.remove(clusterName, clusterId);
             clusterIdToBackend.remove(clusterId);
+            invalidateCloudColocatePlacement(clusterId);
         } finally {
             wlock.unlock();
         }

@@ -19,6 +19,7 @@
 
 import org.apache.doris.analysis.ModifyBackendClause;
 import org.apache.doris.analysis.ModifyBackendHostNameClause;
+import org.apache.doris.catalog.ColocateTableIndex.GroupId;
 import org.apache.doris.catalog.DiskInfo;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.ReplicaAllocation;
@@ -94,6 +95,9 @@ public class SystemInfoService {
 
     private volatile ImmutableMap<Long, DiskInfo> pathHashToDiskInfoRef = ImmutableMap.of();
 
+    public void invalidateCloudColocatePlacement(GroupId groupId) {
+    }
+
     public static class HostInfo implements Comparable<HostInfo> {
         public String host;
         public int port;