From 8d034976c158ae43f515b3975dc82be082ae8630 Mon Sep 17 00:00:00 2001 From: deardeng Date: Fri, 10 Jul 2026 11:48:16 +0800 Subject: [PATCH] [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. (cherry picked from commit 355ede4577fa0cc023536eba87b2e0fe427d28aa) --- .../java/org/apache/doris/common/Config.java | 6 + .../doris/catalog/ColocateTableIndex.java | 3 + .../cloud/catalog/CloudColocatePlacement.java | 64 +++++ .../doris/cloud/catalog/CloudReplica.java | 54 +++- .../cloud/system/CloudSystemInfoService.java | 144 +++++++++++ .../doris/system/SystemInfoService.java | 4 + .../catalog/CloudColocatePlacementTest.java | 232 ++++++++++++++++++ ...st_cloud_colocate_consistent_hash_join.out | 5 + ...cloud_colocate_consistent_hash_join.groovy | 67 +++++ 9 files changed, 570 insertions(+), 9 deletions(-) create mode 100644 fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudColocatePlacement.java create mode 100644 fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudColocatePlacementTest.java create mode 100644 regression-test/data/cloud_p0/multi_cluster/test_cloud_colocate_consistent_hash_join.out create mode 100644 regression-test/suites/cloud_p0/multi_cluster/test_cloud_colocate_consistent_hash_join.groovy 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 04d8dd375b9111..5b0d5245b47d63 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 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; 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 9fa53d062100a3..72a7fc6f09dbad 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 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 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 00000000000000..baec8f4b0e3435 --- /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 1e109c3350e74a..5050f4e6e630c1 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 @@ 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 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 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 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 availableBes) { + if (Config.enable_cloud_colocate_consistent_hash) { + List 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 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 a60d5d46e3fa75..c334db2b824c0a 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 @@ 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 computeGroupIdToComputeGroup = new ConcurrentHashMap<>(); + private final Map colocatePlacementCache = + new ConcurrentHashMap<>(); + private InstanceInfoPB.Status instanceStatus; + public long getCloudColocateHrwBeId(GroupId groupId, String clusterId, List availableBeIds, long idx) { + return getCloudColocateHrwBeIdInternal(groupId, clusterId, availableBeIds, idx, -1); + } + + @VisibleForTesting + public long getCloudColocateHrwBeIdForTest(GroupId groupId, String clusterId, List availableBeIds, + long idx, int bucketNumForTest) { + return getCloudColocateHrwBeIdInternal(groupId, clusterId, availableBeIds, idx, bucketNumForTest); + } + + private long getCloudColocateHrwBeIdInternal(GroupId groupId, String clusterId, List 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(); } 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 4f1ddd23343d00..e36b9687fe9860 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 @@ 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 pathHashToDiskInfoRef = ImmutableMap.of(); + public void invalidateCloudColocatePlacement(GroupId groupId) { + } + public static class HostInfo implements Comparable { 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 00000000000000..c04bd13bff7a20 --- /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 originalBeIds = Arrays.asList(1L, 2L, 3L); + List 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 originalBeIds = Arrays.asList(1L, 2L, 3L); + List 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 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 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 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 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 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 createBackends(long... beIds) { + List 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 00000000000000..869f407b9f37cd --- /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 00000000000000..76070a93017f8d --- /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 + """ + } +}