This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 355ede4577f [improvement](cloud) Use rendezvous hashing for colocate
tablet placement (#64638)
355ede4577f is described below
commit 355ede4577fa0cc023536eba87b2e0fe427d28aa
Author: deardeng <[email protected]>
AuthorDate: Fri Jul 10 11:48:16 2026 +0800
[improvement](cloud) Use rendezvous hashing for colocate tablet placement
(#64638)
In cloud mode, colocate tablet-to-BE placement used modulo hashing
((murmur3(grpId) + idx) % availableBeNum), so scaling a compute group up
or down remapped almost all buckets and caused large cold-cache query
jitter for colocate tables.
Replace it with rendezvous (HRW) hashing behind a new
enable_cloud_colocate_consistent_hash (default true): each bucket picks
the BE with the highest score(grpId, idx, beId). This keeps the colocate
invariant (the same bucket index of all member tables lands on the same
BE) while moving only ~1/N buckets when a single BE is added or removed.
The legacy modulo path is kept as a rollback branch.
Placement is memoized per (group, compute group) in
CloudSystemInfoService with a lock-free cache hit and an
order-independent fingerprint of the available BE set for invalidation.
The cache is local-only (no editlog, no image), and the dead-BE grace
window keeps its existing semantics.
Add unit tests (movement ratio vs modulo, tie-break, cache invalidation,
order independence, modulo rollback) and a cloud docker regression case
for colocate join correctness.
---
.../main/java/org/apache/doris/common/Config.java | 6 +
.../apache/doris/catalog/ColocateTableIndex.java | 3 +
.../cloud/catalog/CloudColocatePlacement.java | 64 ++++++
.../apache/doris/cloud/catalog/CloudReplica.java | 42 +++-
.../doris/cloud/system/CloudSystemInfoService.java | 144 +++++++++++++
.../org/apache/doris/system/SystemInfoService.java | 4 +
.../cloud/catalog/CloudColocatePlacementTest.java | 232 +++++++++++++++++++++
.../doris/cloud/catalog/CloudReplicaTest.java | 15 ++
.../test_cloud_colocate_consistent_hash_join.out | 5 +
...test_cloud_colocate_consistent_hash_join.groovy | 67 ++++++
10 files changed, 574 insertions(+), 8 deletions(-)
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index c6b600104d2..00526509177 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -3405,6 +3405,12 @@ public class Config extends ConfigBase {
+ "to other BEs in cloud mode."})
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 = {
"Whether to enable the automatic start-stop feature in cloud
model, default is true."})
public static boolean enable_auto_start_for_cloud_cluster = true;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/ColocateTableIndex.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/ColocateTableIndex.java
index 06d45ba763a..381fa9dcbf5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/ColocateTableIndex.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/ColocateTableIndex.java
@@ -346,6 +346,9 @@ public class ColocateTableIndex implements Writable {
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)) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudColocatePlacement.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudColocatePlacement.java
new file mode 100644
index 00000000000..baec8f4b0e3
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudColocatePlacement.java
@@ -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;
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
index c663efe1f8c..e507eb49e8d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
@@ -167,28 +167,54 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
}
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);
+ 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
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
index 9d960df6507..13b347694c9 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java
@@ -18,8 +18,11 @@
package org.apache.doris.cloud.system;
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;
@@ -51,6 +54,7 @@ import org.apache.doris.system.Frontend;
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;
@@ -67,6 +71,7 @@ import java.util.Comparator;
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;
@@ -97,8 +102,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);
@@ -284,6 +426,7 @@ public class CloudSystemInfoService extends
SystemInfoService {
wlock.lock();
computeGroupIdToComputeGroup.remove(computeGroupId);
removeVirtualClusterInfoFromMapsNoLock(computeGroupId,
computeGroupName);
+ invalidateCloudColocatePlacement(computeGroupId);
} finally {
wlock.unlock();
}
@@ -1110,6 +1253,7 @@ public class CloudSystemInfoService extends
SystemInfoService {
try {
clusterNameToId.remove(clusterName, clusterId);
clusterIdToBackend.remove(clusterId);
+ invalidateCloudColocatePlacement(clusterId);
} finally {
wlock.unlock();
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
index 5b728267e1b..8b5a80a978e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
@@ -17,6 +17,7 @@
package org.apache.doris.system;
+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;
@@ -96,6 +97,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;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudColocatePlacementTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudColocatePlacementTest.java
new file mode 100644
index 00000000000..c04bd13bff7
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudColocatePlacementTest.java
@@ -0,0 +1,232 @@
+// 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 org.apache.doris.catalog.ColocateTableIndex.GroupId;
+import org.apache.doris.catalog.Replica.ReplicaState;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
+import org.apache.doris.common.Config;
+import org.apache.doris.system.Backend;
+
+import com.google.common.hash.HashCode;
+import com.google.common.hash.Hashing;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public class CloudColocatePlacementTest {
+ private final boolean oldEnableCloudColocateConsistentHash =
Config.enable_cloud_colocate_consistent_hash;
+
+ @AfterEach
+ public void tearDown() {
+ Config.enable_cloud_colocate_consistent_hash =
oldEnableCloudColocateConsistentHash;
+ }
+
+ @Test
+ public void testAddAndRemoveOneBackendMovesFarFewerBucketsThanModulo() {
+ long groupId = 100L;
+ int bucketNum = 4096;
+ long[] originalBeIds = range(1, 16);
+ long[] addedBeIds = range(1, 17);
+ long[] removedBeIds = range(1, 15);
+
+ int hrwAddMoved = changedBucketsByHrw(groupId, bucketNum,
originalBeIds, addedBeIds);
+ int hrwRemoveMoved = changedBucketsByHrw(groupId, bucketNum,
originalBeIds, removedBeIds);
+ int moduloAddMoved = changedBucketsByModulo(groupId, bucketNum,
originalBeIds, addedBeIds);
+ int moduloRemoveMoved = changedBucketsByModulo(groupId, bucketNum,
originalBeIds, removedBeIds);
+
+ Assertions.assertTrue(hrwAddMoved < bucketNum / 8,
+ "HRW should move about 1/N buckets after adding one BE, but
moved " + hrwAddMoved);
+ Assertions.assertTrue(hrwRemoveMoved < bucketNum / 8,
+ "HRW should move about 1/N buckets after removing one BE, but
moved " + hrwRemoveMoved);
+ Assertions.assertTrue(moduloAddMoved > bucketNum * 9 / 10,
+ "Modulo should move almost all buckets after adding one BE,
but moved " + moduloAddMoved);
+ Assertions.assertTrue(moduloRemoveMoved > bucketNum * 9 / 10,
+ "Modulo should move almost all buckets after removing one BE,
but moved " + moduloRemoveMoved);
+ }
+
+ @Test
+ public void testTieBreakPicksSmallerBackendId() {
+ long[] beIds = new long[] {30L, 20L, 10L};
+
+ long pickedBeId = CloudColocatePlacement.pickBackendId(100L, 1L,
beIds, (groupId, idx, beId) -> 1L);
+
+ Assertions.assertEquals(10L, pickedBeId);
+ }
+
+ @Test
+ public void testCacheInvalidatesWhenBackendSetChanges() {
+ CloudSystemInfoService infoService = new CloudSystemInfoService();
+ GroupId groupId = new GroupId(1L, 100L);
+ List<Long> originalBeIds = Arrays.asList(1L, 2L, 3L);
+ List<Long> addedBeIds = Arrays.asList(1L, 2L, 3L, 4L);
+ int bucketNum = 128;
+
+ int changed = 0;
+ for (int idx = 0; idx < bucketNum; idx++) {
+ long original =
infoService.getCloudColocateHrwBeIdForTest(groupId, "cluster0",
+ originalBeIds, idx, bucketNum);
+ long cached = infoService.getCloudColocateHrwBeIdForTest(groupId,
"cluster0",
+ originalBeIds, idx, bucketNum);
+ long afterAdd =
infoService.getCloudColocateHrwBeIdForTest(groupId, "cluster0",
+ addedBeIds, idx, bucketNum);
+ Assertions.assertEquals(original, cached);
+ if (original != afterAdd) {
+ changed++;
+ }
+ }
+
+ Assertions.assertTrue(changed > 0, "Changing BE set should invalidate
cached placement");
+ Assertions.assertTrue(changed < bucketNum / 2, "HRW should not rebuild
as modulo-style reshuffle");
+ }
+
+ @Test
+ public void testCachePlacementSameForSameBackendSetInDifferentOrder() {
+ CloudSystemInfoService infoService = new CloudSystemInfoService();
+ GroupId groupId = new GroupId(1L, 100L);
+ List<Long> originalBeIds = Arrays.asList(1L, 2L, 3L);
+ List<Long> reorderedBeIds = Arrays.asList(3L, 1L, 2L);
+ int bucketNum = 128;
+
+ for (int idx = 0; idx < bucketNum; idx++) {
+ long original =
infoService.getCloudColocateHrwBeIdForTest(groupId, "cluster0",
+ originalBeIds, idx, bucketNum);
+ long reordered =
infoService.getCloudColocateHrwBeIdForTest(groupId, "cluster0",
+ reorderedBeIds, idx, bucketNum);
+ Assertions.assertEquals(original, reordered);
+ }
+ }
+
+ @Test
+ public void testCacheEvictedByClusterId() {
+ CloudSystemInfoService infoService = new CloudSystemInfoService();
+ GroupId groupId = new GroupId(1L, 100L);
+ List<Long> beIds = Arrays.asList(1L, 2L, 3L);
+
+ long original = infoService.getCloudColocateHrwBeIdForTest(groupId,
"cluster0", beIds, 0L, 16);
+ infoService.invalidateCloudColocatePlacement("cluster0");
+ long rebuilt = infoService.getCloudColocateHrwBeIdForTest(groupId,
"cluster0", beIds, 0L, 16);
+
+ Assertions.assertEquals(original, rebuilt);
+ }
+
+ @Test
+ public void testHrwPlacementRejectsInvalidBucketIdxForTest() {
+ CloudSystemInfoService infoService = new CloudSystemInfoService();
+ GroupId groupId = new GroupId(1L, 100L);
+ List<Long> beIds = Arrays.asList(1L, 2L, 3L);
+
+ IllegalStateException exception =
Assertions.assertThrows(IllegalStateException.class,
+ () -> infoService.getCloudColocateHrwBeIdForTest(groupId,
"cluster0", beIds, 8L, 8));
+ Assertions.assertTrue(exception.getMessage().contains("outside bucket
num 8"));
+ }
+
+ @Test
+ public void testDeadGraceHrwPlacementRejectsInvalidBucketIdx() throws
Exception {
+ CloudSystemInfoService infoService = Mockito.spy(new
CloudSystemInfoService());
+ GroupId groupId = new GroupId(1L, 100L);
+ List<Backend> backends = createBackends(1L, 2L, 3L);
+ CloudReplica replica = new CloudReplica(1L, -1L, ReplicaState.NORMAL,
0L, 0, 1L, 2L, 3L, 4L, 8L);
+
Mockito.doReturn(8).when(infoService).getCloudColocateBucketsNum(groupId);
+
+ IllegalStateException exception =
Assertions.assertThrows(IllegalStateException.class,
+ () -> invokePickColocatedBackendForDeadGrace(replica,
infoService, groupId, "cluster0", backends));
+ Assertions.assertTrue(exception.getMessage().contains("outside bucket
num 8"));
+ }
+
+ @Test
+ public void testConfigOffUsesLegacyModuloResult() {
+ Config.enable_cloud_colocate_consistent_hash = false;
+ GroupId groupId = new GroupId(1L, 100L);
+ List<Backend> backends = createBackends(10L, 20L, 30L);
+ CloudReplica replica = new CloudReplica(1L, -1L, ReplicaState.NORMAL,
0L, 0, 1L, 2L, 3L, 4L, 5L);
+
+ long pickedBeId = replica.pickColocatedBackend(new
CloudSystemInfoService(), groupId, "cluster0", backends)
+ .getId();
+ long expectedBeId = pickModulo(groupId.grpId, 5L,
backends.stream().mapToLong(Backend::getId).toArray());
+
+ Assertions.assertEquals(expectedBeId, pickedBeId);
+ }
+
+ private static long[] range(int fromInclusive, int toExclusive) {
+ long[] values = new long[toExclusive - fromInclusive];
+ for (int i = 0; i < values.length; i++) {
+ values[i] = fromInclusive + i;
+ }
+ return values;
+ }
+
+ private static int changedBucketsByHrw(long groupId, int bucketNum, long[]
originalBeIds, long[] changedBeIds) {
+ int changed = 0;
+ for (int idx = 0; idx < bucketNum; idx++) {
+ long before = CloudColocatePlacement.pickBackendId(groupId, idx,
originalBeIds);
+ long after = CloudColocatePlacement.pickBackendId(groupId, idx,
changedBeIds);
+ if (before != after) {
+ changed++;
+ }
+ }
+ return changed;
+ }
+
+ private static int changedBucketsByModulo(long groupId, int bucketNum,
long[] originalBeIds, long[] changedBeIds) {
+ int changed = 0;
+ for (int idx = 0; idx < bucketNum; idx++) {
+ long before = pickModulo(groupId, idx, originalBeIds);
+ long after = pickModulo(groupId, idx, changedBeIds);
+ if (before != after) {
+ changed++;
+ }
+ }
+ return changed;
+ }
+
+ private static long pickModulo(long groupId, long idx, long[] beIds) {
+ HashCode hashCode = Hashing.murmur3_128().hashLong(groupId);
+ long index = (hashCode.asLong() + idx) % beIds.length;
+ index = (index + beIds.length) % beIds.length;
+ return beIds[(int) index];
+ }
+
+ private static Backend invokePickColocatedBackendForDeadGrace(CloudReplica
replica,
+ CloudSystemInfoService infoService, GroupId groupId, String
clusterId, List<Backend> backends)
+ throws Throwable {
+ Method method =
CloudReplica.class.getDeclaredMethod("pickColocatedBackendForDeadGrace",
+ CloudSystemInfoService.class, GroupId.class, String.class,
List.class);
+ method.setAccessible(true);
+ try {
+ return (Backend) method.invoke(replica, infoService, groupId,
clusterId, backends);
+ } catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ }
+
+ private static List<Backend> createBackends(long... beIds) {
+ List<Backend> backends = new ArrayList<>();
+ for (long beId : beIds) {
+ backends.add(new Backend(beId, "127.0.0." + beId, 9050));
+ }
+ return backends;
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudReplicaTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudReplicaTest.java
index 6edff8211c2..ca80fde8e0e 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudReplicaTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudReplicaTest.java
@@ -45,6 +45,7 @@ public class CloudReplicaTest {
private static final long INDEX_ID = 40001L;
private static final long REPLICA_ID = 50001L;
private static final long IDX = 0;
+ private static final int BUCKET_NUM = 128;
private static final String CLUSTER_ID_1 = "cluster_id_1";
private static final String CLUSTER_ID_2 = "cluster_id_2";
@@ -59,6 +60,7 @@ public class CloudReplicaTest {
private boolean savedEnableCloudMultiReplica;
private int savedCloudReplicaNum;
private String savedCloudUniqueId;
+ private boolean savedEnableCloudColocateConsistentHash;
@BeforeEach
public void setUp() {
@@ -66,6 +68,8 @@ public class CloudReplicaTest {
savedEnableCloudMultiReplica = Config.enable_cloud_multi_replica;
savedCloudReplicaNum = Config.cloud_replica_num;
savedCloudUniqueId = Config.cloud_unique_id;
+ savedEnableCloudColocateConsistentHash =
Config.enable_cloud_colocate_consistent_hash;
+ Config.enable_cloud_colocate_consistent_hash = true;
Config.cloud_unique_id = "test_cloud_unique_id";
mockEnv = Mockito.mock(Env.class);
@@ -76,6 +80,16 @@ public class CloudReplicaTest {
envMockedStatic.when(Env::getCurrentEnv).thenReturn(mockEnv);
envMockedStatic.when(Env::getCurrentColocateIndex).thenReturn(mockColocateIndex);
envMockedStatic.when(Env::getCurrentSystemInfo).thenReturn(mockInfoService);
+
+
Mockito.when(mockInfoService.getCloudColocateBucketsNum(Mockito.any(GroupId.class))).thenReturn(BUCKET_NUM);
+
Mockito.when(mockInfoService.getCloudColocateHrwBeId(Mockito.any(GroupId.class),
Mockito.anyString(),
+ Mockito.anyList(), Mockito.anyLong())).thenAnswer(invocation
-> {
+ GroupId groupId = invocation.getArgument(0);
+ List<Long> candidateBeIds = invocation.getArgument(2);
+ long bucketIdx = invocation.getArgument(3);
+ return CloudColocatePlacement.pickBackendId(groupId.grpId,
bucketIdx,
+
candidateBeIds.stream().mapToLong(Long::longValue).toArray());
+ });
}
@AfterEach
@@ -85,6 +99,7 @@ public class CloudReplicaTest {
Config.enable_cloud_multi_replica = savedEnableCloudMultiReplica;
Config.cloud_replica_num = savedCloudReplicaNum;
Config.cloud_unique_id = savedCloudUniqueId;
+ Config.enable_cloud_colocate_consistent_hash =
savedEnableCloudColocateConsistentHash;
}
private CloudReplica createReplica() {
diff --git
a/regression-test/data/cloud_p0/multi_cluster/test_cloud_colocate_consistent_hash_join.out
b/regression-test/data/cloud_p0/multi_cluster/test_cloud_colocate_consistent_hash_join.out
new file mode 100644
index 00000000000..869f407b9f3
--- /dev/null
+++
b/regression-test/data/cloud_p0/multi_cluster/test_cloud_colocate_consistent_hash_join.out
@@ -0,0 +1,5 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !join --
+1 10 100
+2 20 200
+4 40 400
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/test_cloud_colocate_consistent_hash_join.groovy
b/regression-test/suites/cloud_p0/multi_cluster/test_cloud_colocate_consistent_hash_join.groovy
new file mode 100644
index 00000000000..76070a93017
--- /dev/null
+++
b/regression-test/suites/cloud_p0/multi_cluster/test_cloud_colocate_consistent_hash_join.groovy
@@ -0,0 +1,67 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+
+suite("test_cloud_colocate_consistent_hash_join", "multi_cluster,docker") {
+ def options = new ClusterOptions()
+ options.setFeNum(1)
+ options.setBeNum(3)
+ options.cloudMode = true
+ options.feConfigs += [
+ "enable_cloud_colocate_consistent_hash=true"
+ ]
+
+ docker(options) {
+ sql "DROP TABLE IF EXISTS test_cloud_colocate_consistent_hash_join_t1"
+ sql "DROP TABLE IF EXISTS test_cloud_colocate_consistent_hash_join_t2"
+
+ sql """
+ CREATE TABLE test_cloud_colocate_consistent_hash_join_t1 (
+ k INT,
+ v1 INT
+ )
+ DISTRIBUTED BY HASH(k) BUCKETS 8
+ PROPERTIES (
+ "replication_num" = "1",
+ "colocate_with" =
"test_cloud_colocate_consistent_hash_join_group"
+ )
+ """
+
+ sql """
+ CREATE TABLE test_cloud_colocate_consistent_hash_join_t2 (
+ k INT,
+ v2 INT
+ )
+ DISTRIBUTED BY HASH(k) BUCKETS 8
+ PROPERTIES (
+ "replication_num" = "1",
+ "colocate_with" =
"test_cloud_colocate_consistent_hash_join_group"
+ )
+ """
+
+ sql "INSERT INTO test_cloud_colocate_consistent_hash_join_t1 VALUES
(1, 10), (2, 20), (3, 30), (4, 40)"
+ sql "INSERT INTO test_cloud_colocate_consistent_hash_join_t2 VALUES
(1, 100), (2, 200), (4, 400), (5, 500)"
+
+ order_qt_join """
+ SELECT t1.k, t1.v1, t2.v2
+ FROM test_cloud_colocate_consistent_hash_join_t1 t1
+ JOIN test_cloud_colocate_consistent_hash_join_t2 t2 ON t1.k = t2.k
+ ORDER BY t1.k
+ """
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
