This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 2bc3248a335 [improvement](cloud) Use rendezvous hashing for colocate
tablet placement (#64638) (#66067)
2bc3248a335 is described below
commit 2bc3248a335d3e086ffb6c1b30422adc12abca82
Author: deardeng <[email protected]>
AuthorDate: Mon Jul 27 22:27:47 2026 +0800
[improvement](cloud) Use rendezvous hashing for colocate tablet placement
(#64638) (#66067)
pick from https://github.com/apache/doris/pull/64638
---
.../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 | 54 ++++-
.../doris/cloud/system/CloudSystemInfoService.java | 144 +++++++++++++
.../org/apache/doris/system/SystemInfoService.java | 4 +
.../cloud/catalog/CloudColocatePlacementTest.java | 232 +++++++++++++++++++++
.../test_cloud_colocate_consistent_hash_join.out | 5 +
...test_cloud_colocate_consistent_hash_join.groovy | 67 ++++++
9 files changed, 570 insertions(+), 9 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 04d8dd375b9..5b0d5245b47 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
@@ -3649,6 +3649,12 @@ public class Config extends ConfigBase {
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;
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 6bf1cf2e19b..7095ffa4969 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 1e109c3350e..5050f4e6e63 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
@@ -114,6 +114,16 @@ public class CloudReplica extends Replica implements
GsonPostProcessable {
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 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);
- 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
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 a60d5d46e3f..c334db2b824 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
@@ -20,8 +20,11 @@ package org.apache.doris.cloud.system;
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.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;
@@ -69,6 +73,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;
@@ -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 class CloudSystemInfoService extends
SystemInfoService {
wlock.lock();
computeGroupIdToComputeGroup.remove(computeGroupId);
removeVirtualClusterInfoFromMapsNoLock(computeGroupId,
computeGroupName);
+ invalidateCloudColocatePlacement(computeGroupId);
} finally {
wlock.unlock();
}
@@ -1119,6 +1262,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 4f1ddd23343..e36b9687fe9 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
@@ -19,6 +19,7 @@ package org.apache.doris.system;
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;
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/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]
