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 + """ + } +}