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 6bf1cf2e19b0c7..4ef335e40e9b53 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 @@ -437,6 +437,19 @@ public GroupId getGroup(long tableId) { } } + public Long getDbIdByTblIdNullable(GroupId groupId, long tableId) { + readLock(); + try { + GroupId tableGroupId = table2Group.get(tableId); + if (tableGroupId == null || !tableGroupId.equals(groupId)) { + return null; + } + return tableGroupId.tblId2DbId.get(tableId); + } finally { + readUnlock(); + } + } + public Set getAllGroupIds() { readLock(); try { @@ -711,7 +724,11 @@ public List> getInfos() { info.add(Joiner.on(", ").join(group2Tables.get(groupId))); ColocateGroupSchema groupSchema = group2Schema.get(groupId); info.add(String.valueOf(groupSchema.getBucketsNum())); - info.add(String.valueOf(groupSchema.getReplicaAlloc().toCreateStmt())); + if (Config.isCloudMode()) { + info.add("null"); + } else { + info.add(String.valueOf(groupSchema.getReplicaAlloc().toCreateStmt())); + } List cols = groupSchema.getDistributionColTypes().stream().map( e -> e.toSql()).collect(Collectors.toList()); info.add(Joiner.on(", ").join(cols)); diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Replica.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Replica.java index ad83287925c0c4..b287c68ff0d50d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Replica.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Replica.java @@ -29,6 +29,9 @@ import org.apache.logging.log4j.Logger; import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; /** * This class represents the olap replica related metadata. @@ -173,6 +176,24 @@ protected long getBackendIdValue() { return -1L; } + public long getBackendIdForProcDisplay() { + return getBackendIdValue(); + } + + // For proc display only. Returns a "scope key -> backendId" mapping used to render the + // replica's placement. In local deployment there is a single scope (empty key) and the + // cache is unused; in cloud deployment each compute group is a separate scope and the + // cache lets a single proc call fetch each compute group's backends once (see + // CloudReplica). The cache is keyed by compute group id. + public Map getClusterToBackendForProcDisplay(Map> computeGroupBackendCache) { + Map result = new HashMap<>(); + long backendId = getBackendIdForProcDisplay(); + if (backendId != -1L) { + result.put("", backendId); + } + return result; + } + public void setBackendId(long backendId) { if (backendId != -1) { throw new UnsupportedOperationException("setBackendId is not supported in Replica"); 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 07a2f4166354d6..0924406c50ba6c 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 @@ -40,6 +40,7 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Random; @@ -107,8 +108,17 @@ private boolean isColocated() { } public long getColocatedBeId(String clusterId) throws ComputeGroupException { + List clusterBackends = ((CloudSystemInfoService) Env.getCurrentSystemInfo()) + .getBackendsByClusterId(clusterId); + return getColocatedBeId(clusterId, clusterBackends); + } + + // Same as getColocatedBeId(clusterId) but reuses an already fetched backend list of + // the compute group. Lets callers that resolve many replicas across the same compute + // groups (e.g. the colocate proc display) fetch each group's backends only once. + public long getColocatedBeId(String clusterId, List clusterBackends) throws ComputeGroupException { CloudSystemInfoService infoService = ((CloudSystemInfoService) Env.getCurrentSystemInfo()); - List bes = infoService.getBackendsByClusterId(clusterId).stream() + List bes = clusterBackends.stream() .filter(be -> be.isQueryAvailable()).collect(Collectors.toList()); String clusterName = infoService.getClusterNameByClusterId(clusterId); if (bes.isEmpty()) { @@ -230,6 +240,47 @@ public long getClusterPrimaryBackendId(String clusterId) { return primaryClusterToBackend.getOrDefault(clusterId, -1L); } + // For proc display only. In cloud mode a replica is hashed to a different BE in each + // compute group, so expose a clusterId -> backendId mapping; the proc display builds + // a separate bucket sequence per compute group from it so each group's sequence is + // self-consistent. Do not collapse this into a single BE (e.g. the first one): + // backends differ across compute groups and would not match. + // + // ATTN: colocated replicas do NOT use primaryClusterToBackend (see getBackendIdImpl / + // getClusterPrimaryBackendId, which short-circuit to getColocatedBeId), so that cache + // is empty for them. Resolve their placement per compute group on the fly instead. + // This reads CloudSystemInfoService / the colocate index, so callers must invoke it + // OUTSIDE any table lock to avoid nested lock acquisition. It does not auto-start any + // compute group: getColocatedBeId only reads the already-known backends of a clusterId. + // computeGroupBackendCache maps compute group id -> getBackendsByClusterId() result and + // is shared across all replicas resolved in a single proc call, so each compute group's + // backend list is fetched only once instead of once per replica. + @Override + public Map getClusterToBackendForProcDisplay(Map> computeGroupBackendCache) { + if (!isColocated()) { + return new HashMap<>(primaryClusterToBackend); + } + Map result = new HashMap<>(); + CloudSystemInfoService infoService = (CloudSystemInfoService) Env.getCurrentSystemInfo(); + for (String clusterId : infoService.getCloudClusterIds()) { + try { + List clusterBackends = + computeGroupBackendCache.computeIfAbsent(clusterId, infoService::getBackendsByClusterId); + long backendId = getColocatedBeId(clusterId, clusterBackends); + if (backendId != -1L) { + result.put(clusterId, backendId); + } + } catch (ComputeGroupException e) { + // Skip compute groups that currently have no available backend. + if (LOG.isDebugEnabled()) { + LOG.debug("skip compute group {} for colocate proc display, replica {}", + clusterId, getId(), e); + } + } + } + return result; + } + private long getBackendIdImpl(String clusterId) throws ComputeGroupException { if (Strings.isNullOrEmpty(clusterId)) { return -1L; diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/proc/ColocationGroupBackendSeqsProcNode.java b/fe/fe-core/src/main/java/org/apache/doris/common/proc/ColocationGroupBackendSeqsProcNode.java index 4ef502b1193c34..a2a7caeaa311d6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/common/proc/ColocationGroupBackendSeqsProcNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/common/proc/ColocationGroupBackendSeqsProcNode.java @@ -18,7 +18,6 @@ package org.apache.doris.common.proc; import org.apache.doris.common.AnalysisException; -import org.apache.doris.resource.Tag; import com.google.common.base.Joiner; import com.google.common.collect.Lists; @@ -30,10 +29,13 @@ * show proc "/colocation_group/group_name"; */ public class ColocationGroupBackendSeqsProcNode implements ProcNodeInterface { - private Map>> backendsSeq; + // Column name -> per-bucket backend id sequence. The column name is a resource Tag in + // local mode, a compute group name in cloud mode, or "BackendIds" when there is no + // per-scope breakdown. + private Map>> backendsSeqByColumn; - public ColocationGroupBackendSeqsProcNode(Map>> backendsSeq) { - this.backendsSeq = backendsSeq; + public ColocationGroupBackendSeqsProcNode(Map>> backendsSeqByColumn) { + this.backendsSeqByColumn = backendsSeqByColumn; } @Override @@ -42,21 +44,21 @@ public ProcResult fetchResult() throws AnalysisException { List titleNames = Lists.newArrayList(); titleNames.add("BucketIndex"); int bucketNum = 0; - for (Tag tag : backendsSeq.keySet()) { - titleNames.add(tag.toString()); + for (String column : backendsSeqByColumn.keySet()) { + titleNames.add(column); if (bucketNum == 0) { - bucketNum = backendsSeq.get(tag).size(); - } else if (bucketNum != backendsSeq.get(tag).size()) { + bucketNum = backendsSeqByColumn.get(column).size(); + } else if (bucketNum != backendsSeqByColumn.get(column).size()) { throw new AnalysisException("Invalid bucket number: " - + bucketNum + " vs. " + backendsSeq.get(tag).size()); + + bucketNum + " vs. " + backendsSeqByColumn.get(column).size()); } } result.setNames(titleNames); for (int i = 0; i < bucketNum; i++) { List info = Lists.newArrayList(); info.add(String.valueOf(i)); // bucket index - for (Tag tag : backendsSeq.keySet()) { - List> bucketBackends = backendsSeq.get(tag); + for (String column : backendsSeqByColumn.keySet()) { + List> bucketBackends = backendsSeqByColumn.get(column); info.add(Joiner.on(", ").join(bucketBackends.get(i))); } result.addRow(info); diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/proc/ColocationGroupProcDir.java b/fe/fe-core/src/main/java/org/apache/doris/common/proc/ColocationGroupProcDir.java index 85dd2c97be6590..b2569c44919336 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/common/proc/ColocationGroupProcDir.java +++ b/fe/fe-core/src/main/java/org/apache/doris/common/proc/ColocationGroupProcDir.java @@ -19,14 +19,30 @@ import org.apache.doris.catalog.ColocateTableIndex; import org.apache.doris.catalog.ColocateTableIndex.GroupId; +import org.apache.doris.catalog.Database; import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.MaterializedIndex; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.Partition; +import org.apache.doris.catalog.Replica; +import org.apache.doris.catalog.Table; +import org.apache.doris.catalog.Tablet; +import org.apache.doris.cloud.system.CloudSystemInfoService; import org.apache.doris.common.AnalysisException; +import org.apache.doris.common.Config; import org.apache.doris.resource.Tag; +import org.apache.doris.system.Backend; +import com.google.common.base.Strings; import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import com.google.common.collect.Sets; +import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.Set; /* * show proc "/colocation_group"; @@ -61,7 +77,22 @@ public ProcNodeInterface lookup(String groupIdStr) throws AnalysisException { GroupId groupId = new GroupId(dbId, grpId); ColocateTableIndex index = Env.getCurrentColocateIndex(); Map>> beSeqs = index.getBackendsPerBucketSeq(groupId); - return new ColocationGroupBackendSeqsProcNode(beSeqs); + Map>> columns; + if ((beSeqs == null || beSeqs.isEmpty()) && Config.isCloudMode()) { + // In cloud mode, legacy backend sequence metadata may be empty. Derive the + // sequence from current tablets, one column per compute group. This path must + // not resolve cloud backends in a way that auto-starts a compute group. + columns = getCloudBackendSeqsFromTablets(groupId, index); + } else { + // Local mode: one column per resource tag. + columns = Maps.newLinkedHashMap(); + if (beSeqs != null) { + for (Map.Entry>> entry : beSeqs.entrySet()) { + columns.put(entry.getKey().toString(), entry.getValue()); + } + } + } + return new ColocationGroupBackendSeqsProcNode(columns); } @Override @@ -74,4 +105,129 @@ public ProcResult fetchResult() throws AnalysisException { result.setRows(infos); return result; } + + private Map>> getCloudBackendSeqsFromTablets(GroupId groupId, ColocateTableIndex index) { + Map>> backendsSeq = Maps.newLinkedHashMap(); + List tableIds = index.getAllTableIds(groupId); + for (Long tableId : tableIds) { + long dbId = groupId.dbId; + if (dbId == 0) { + Long tableDbId = index.getDbIdByTblIdNullable(groupId, tableId); + if (tableDbId == null) { + continue; + } + dbId = tableDbId; + } + Database db = Env.getCurrentInternalCatalog().getDbNullable(dbId); + if (db == null) { + continue; + } + Table table = db.getTableNullable(tableId); + if (!(table instanceof OlapTable)) { + continue; + } + backendsSeq = getCloudBackendSeqsFromTable((OlapTable) table); + if (!backendsSeq.isEmpty()) { + return backendsSeq; + } + } + return backendsSeq; + } + + private Map>> getCloudBackendSeqsFromTable(OlapTable olapTable) { + // Snapshot replicas (ordered by bucket) under the table lock only. Resolving the + // per-compute-group placement of colocate cloud replicas calls into + // CloudSystemInfoService / the colocate index, which must run outside the table + // lock to avoid nested lock acquisition. + List> bucketReplicas = Lists.newArrayList(); + olapTable.readLock(); + try { + Partition firstPartition = null; + for (Partition partition : olapTable.getAllPartitions()) { + firstPartition = partition; + break; + } + if (firstPartition == null) { + return Maps.newLinkedHashMap(); + } + MaterializedIndex baseIndex = firstPartition.getBaseIndex(); + for (Tablet tablet : baseIndex.getTablets()) { + bucketReplicas.add(new ArrayList<>(tablet.getReplicas())); + } + } finally { + olapTable.readUnlock(); + } + + // Resolve each replica's per-compute-group placement outside the table lock. In + // cloud mode a replica is hashed to a different BE in each compute group, so build + // a separate bucket sequence per compute group. Merging across groups (picking an + // arbitrary first BE) would mix BEs from different compute groups into one bucket + // sequence, which is meaningless. For colocate cloud tables placement is computed + // on the fly; otherwise it comes from the cached clusterId -> backendId map (or an + // empty scope key for local-style replicas). + List>> tabletReplicaBackends = Lists.newArrayListWithCapacity(bucketReplicas.size()); + Set scopeKeys = Sets.newLinkedHashSet(); + // Shared across all replicas in this proc call so each compute group's backend + // list is fetched only once (colocate placement is resolved per compute group). + Map> computeGroupBackendCache = Maps.newHashMap(); + for (List replicas : bucketReplicas) { + List> replicaBackends = new ArrayList<>(); + for (Replica replica : replicas) { + Map clusterToBackend = + replica.getClusterToBackendForProcDisplay(computeGroupBackendCache); + replicaBackends.add(clusterToBackend); + scopeKeys.addAll(clusterToBackend.keySet()); + } + tabletReplicaBackends.add(replicaBackends); + } + + Map>> seqByScopeKey = Maps.newLinkedHashMap(); + for (String scopeKey : scopeKeys) { + List> bucketSeq = Lists.newArrayListWithCapacity(tabletReplicaBackends.size()); + boolean hasBackend = false; + for (List> replicaBackends : tabletReplicaBackends) { + List bucketBackends = new ArrayList<>(); + for (Map clusterToBackend : replicaBackends) { + Long backendId = clusterToBackend.get(scopeKey); + if (backendId == null || backendId < 0) { + continue; + } + bucketBackends.add(backendId); + hasBackend = true; + } + bucketSeq.add(bucketBackends); + } + if (hasBackend) { + seqByScopeKey.put(scopeKey, bucketSeq); + } + } + + // Resolve scope keys to display column names (also outside the table lock): name + // resolution acquires CloudSystemInfoService's lock. + Map>> backendsSeq = Maps.newLinkedHashMap(); + for (Map.Entry>> entry : seqByScopeKey.entrySet()) { + backendsSeq.put(scopeKeyToColumnName(entry.getKey()), entry.getValue()); + } + return backendsSeq; + } + + // Map a proc-display scope key to its column name. An empty key means there is no + // per-compute-group breakdown (local-style replicas), shown as a single "BackendIds" + // column. Otherwise the key is a cloud compute group id, shown by its compute group + // name (falling back to the raw id when the name cannot be resolved). + private String scopeKeyToColumnName(String scopeKey) { + if (Strings.isNullOrEmpty(scopeKey)) { + return "BackendIds"; + } + try { + String name = ((CloudSystemInfoService) Env.getCurrentSystemInfo()) + .getClusterNameByClusterId(scopeKey); + if (!Strings.isNullOrEmpty(name)) { + return name; + } + } catch (Exception e) { + // Fall back to the raw compute group id if name resolution is unavailable. + } + return scopeKey; + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/proc/TabletHealthProcDir.java b/fe/fe-core/src/main/java/org/apache/doris/common/proc/TabletHealthProcDir.java index b1be9c79ab6252..46c41dabebca7b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/common/proc/TabletHealthProcDir.java +++ b/fe/fe-core/src/main/java/org/apache/doris/common/proc/TabletHealthProcDir.java @@ -199,7 +199,12 @@ static class DBTabletStatistic { Tablet tablet = tablets.get(i); ++tabletNum; Tablet.TabletStatus res = null; - if (groupId != null) { + if (Config.isCloudMode()) { + // In cloud mode, tablet replica health is managed by cloud components. + // getHealth/getColocateHealth follows local deployment logic and may + // misclassify tablets as UNRECOVERABLE. + res = Tablet.TabletStatus.HEALTHY; + } else if (groupId != null) { ColocateGroupSchema groupSchema = colocateTableIndex.getGroupSchema(groupId); if (groupSchema != null) { replicaAlloc = groupSchema.getReplicaAlloc(); diff --git a/fe/fe-core/src/test/java/org/apache/doris/common/proc/CloudProcVersionDisplayTest.java b/fe/fe-core/src/test/java/org/apache/doris/common/proc/CloudProcVersionDisplayTest.java index 57c7df79b183a4..1957e565787468 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/common/proc/CloudProcVersionDisplayTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/common/proc/CloudProcVersionDisplayTest.java @@ -69,7 +69,7 @@ public class CloudProcVersionDisplayTest { private SystemInfoService systemInfoService; private TabletInvertedIndex invertedIndex; private InternalCatalog internalCatalog; - private MockedStatic envStatic; + private MockedStatic mockedEnv; private String originDeployMode; private String originCloudUniqueId; @@ -92,15 +92,15 @@ public void setUp() throws AnalysisException { Mockito.when(env.isReady()).thenReturn(true); Mockito.when(systemInfoService.getAllBackendsByAllCluster()).thenReturn(ImmutableMap.of()); - envStatic = Mockito.mockStatic(Env.class, Mockito.CALLS_REAL_METHODS); - envStatic.when(Env::getServingEnv).thenReturn(env); - envStatic.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService); + mockedEnv = Mockito.mockStatic(Env.class, Mockito.CALLS_REAL_METHODS); + mockedEnv.when(Env::getServingEnv).thenReturn(env); + mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService); } @After public void tearDown() { - if (envStatic != null) { - envStatic.close(); + if (mockedEnv != null) { + mockedEnv.close(); } Config.deploy_mode = originDeployMode; Config.cloud_unique_id = originCloudUniqueId; @@ -124,9 +124,9 @@ public void testIndicesLookupPropagatesPartitionCachedVersionToTabletsProc() thr public void testReplicasProcNodeShowsPartitionCachedVersionInCloudMode() throws AnalysisException { ProcTestContext context = createProcTestContext(); - envStatic.when(Env::getCurrentInvertedIndex).thenReturn(invertedIndex); - envStatic.when(Env::getCurrentInternalCatalog).thenReturn(internalCatalog); + mockedEnv.when(Env::getCurrentInvertedIndex).thenReturn(invertedIndex); Mockito.when(invertedIndex.getTabletMeta(TABLET_ID)).thenReturn(context.tabletMeta); + mockedEnv.when(Env::getCurrentInternalCatalog).thenReturn(internalCatalog); Mockito.when(internalCatalog.getDbNullable(DB_ID)).thenReturn(context.db); ReplicasProcNode procNode = new ReplicasProcNode(TABLET_ID, context.tablet.getReplicas()); diff --git a/fe/fe-core/src/test/java/org/apache/doris/common/proc/ColocationGroupProcDirTest.java b/fe/fe-core/src/test/java/org/apache/doris/common/proc/ColocationGroupProcDirTest.java new file mode 100644 index 00000000000000..bfb53825947cb8 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/common/proc/ColocationGroupProcDirTest.java @@ -0,0 +1,294 @@ +// 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.common.proc; + +import org.apache.doris.catalog.ColocateTableIndex; +import org.apache.doris.catalog.ColocateTableIndex.GroupId; +import org.apache.doris.catalog.Database; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.Replica.ReplicaState; +import org.apache.doris.catalog.Table; +import org.apache.doris.cloud.catalog.CloudReplica; +import org.apache.doris.cloud.qe.ComputeGroupException; +import org.apache.doris.cloud.system.CloudSystemInfoService; +import org.apache.doris.common.Config; +import org.apache.doris.resource.Tag; +import org.apache.doris.system.Backend; +import org.apache.doris.utframe.TestWithFeService; + +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.List; +import java.util.Map; + +public class ColocationGroupProcDirTest extends TestWithFeService { + private Database db; + + @Override + protected int backendNum() { + return 1; + } + + @Override + protected void runBeforeAll() throws Exception { + createDatabase("test"); + useDatabase("test"); + db = Env.getCurrentInternalCatalog().getDbOrMetaException("test"); + } + + @Override + protected void runBeforeEach() throws Exception { + for (Table table : db.getTables()) { + dropTable(table.getName(), true); + } + } + + @Test + public void testLocalColocationGroupDetailKeepsTagColumns() throws Exception { + Tag tag1 = Tag.create(Tag.TYPE_LOCATION, "tag1"); + Tag tag2 = Tag.create(Tag.TYPE_LOCATION, "tag2"); + Map>> backendsSeq = Maps.newLinkedHashMap(); + backendsSeq.put(tag1.toString(), Lists.newArrayList( + Lists.newArrayList(10001L, 10002L), + Lists.newArrayList(10003L))); + backendsSeq.put(tag2.toString(), Lists.newArrayList( + Lists.newArrayList(20001L), + Lists.newArrayList(20002L, 20003L))); + + ProcResult result = new ColocationGroupBackendSeqsProcNode(backendsSeq).fetchResult(); + + Assertions.assertEquals(Lists.newArrayList("BucketIndex", tag1.toString(), tag2.toString()), + result.getColumnNames()); + Assertions.assertEquals(Lists.newArrayList("0", "10001, 10002", "20001"), result.getRows().get(0)); + Assertions.assertEquals(Lists.newArrayList("1", "10003", "20002, 20003"), result.getRows().get(1)); + } + + @Test + public void testCloudColocationGroupDetailWithoutTag() throws Exception { + String originDeployMode = Config.deploy_mode; + createTable("CREATE TABLE colocate_t1 (k INT) DISTRIBUTED BY HASH(k) BUCKETS 2 " + + "PROPERTIES ('replication_num' = '1', 'colocate_with' = 'g1')"); + createTable("CREATE TABLE colocate_t2 (k INT) DISTRIBUTED BY HASH(k) BUCKETS 2 " + + "PROPERTIES ('replication_num' = '1', 'colocate_with' = 'g1')"); + + OlapTable table1 = (OlapTable) db.getTableOrMetaException("colocate_t1"); + GroupId groupId = Env.getCurrentColocateIndex().getGroup(table1.getId()); + Assertions.assertNotNull(groupId); + + ColocateTableIndex colocateTableIndex = Mockito.spy(Env.getCurrentColocateIndex()); + Mockito.doReturn(Maps.>>newHashMap()).when(colocateTableIndex) + .getBackendsPerBucketSeq(groupId); + Config.deploy_mode = "cloud"; + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class, Mockito.CALLS_REAL_METHODS)) { + mockedEnv.when(Env::getCurrentColocateIndex).thenReturn(colocateTableIndex); + ProcNodeInterface node = new ColocationGroupProcDir().lookup(groupId.toString()); + ProcResult result = node.fetchResult(); + Assertions.assertEquals(Lists.newArrayList("BucketIndex", "BackendIds"), result.getColumnNames()); + Assertions.assertFalse(result.getRows().isEmpty()); + Assertions.assertTrue(result.getRows().stream().anyMatch(row -> row.size() == 2 && !row.get(1).isEmpty())); + } finally { + Config.deploy_mode = originDeployMode; + } + } + + @Test + public void testCloudGlobalColocationGroupDetailFallback() throws Exception { + String originDeployMode = Config.deploy_mode; + createTable("CREATE TABLE global_colocate_t1 (k INT) DISTRIBUTED BY HASH(k) BUCKETS 2 " + + "PROPERTIES ('replication_num' = '1', 'colocate_with' = '__global__g1')"); + + OlapTable table1 = (OlapTable) db.getTableOrMetaException("global_colocate_t1"); + GroupId groupId = Env.getCurrentColocateIndex().getGroup(table1.getId()); + Assertions.assertEquals(0L, groupId.dbId.longValue()); + + ColocateTableIndex colocateTableIndex = Mockito.spy(Env.getCurrentColocateIndex()); + Mockito.doReturn(Maps.>>newHashMap()).when(colocateTableIndex) + .getBackendsPerBucketSeq(groupId); + Config.deploy_mode = "cloud"; + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class, Mockito.CALLS_REAL_METHODS)) { + mockedEnv.when(Env::getCurrentColocateIndex).thenReturn(colocateTableIndex); + ProcNodeInterface node = new ColocationGroupProcDir().lookup(groupId.toString()); + ProcResult result = node.fetchResult(); + Assertions.assertEquals(Lists.newArrayList("BucketIndex", "BackendIds"), result.getColumnNames()); + Assertions.assertFalse(result.getRows().isEmpty()); + Assertions.assertTrue(result.getRows().stream().anyMatch(row -> row.size() == 2 && !row.get(1).isEmpty())); + } finally { + Config.deploy_mode = originDeployMode; + } + } + + @Test + public void testCloudColocationGroupDetailFallbackSkipsUnusableFirstTable() throws Exception { + String originDeployMode = Config.deploy_mode; + createTable("CREATE TABLE colocate_t5 (k INT) DISTRIBUTED BY HASH(k) BUCKETS 2 " + + "PROPERTIES ('replication_num' = '1', 'colocate_with' = 'g3')"); + createTable("CREATE TABLE colocate_t6 (k INT) DISTRIBUTED BY HASH(k) BUCKETS 2 " + + "PROPERTIES ('replication_num' = '1', 'colocate_with' = 'g3')"); + + OlapTable table1 = (OlapTable) db.getTableOrMetaException("colocate_t5"); + GroupId groupId = Env.getCurrentColocateIndex().getGroup(table1.getId()); + db.unregisterTable(table1.getId()); + + ColocateTableIndex colocateTableIndex = Mockito.spy(Env.getCurrentColocateIndex()); + Mockito.doReturn(Maps.>>newHashMap()).when(colocateTableIndex) + .getBackendsPerBucketSeq(groupId); + Config.deploy_mode = "cloud"; + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class, Mockito.CALLS_REAL_METHODS)) { + mockedEnv.when(Env::getCurrentColocateIndex).thenReturn(colocateTableIndex); + ProcNodeInterface node = new ColocationGroupProcDir().lookup(groupId.toString()); + ProcResult result = node.fetchResult(); + Assertions.assertEquals(Lists.newArrayList("BucketIndex", "BackendIds"), result.getColumnNames()); + Assertions.assertFalse(result.getRows().isEmpty()); + Assertions.assertTrue(result.getRows().stream().anyMatch(row -> row.size() == 2 && !row.get(1).isEmpty())); + } finally { + db.registerTable(table1); + Config.deploy_mode = originDeployMode; + } + } + + @Test + public void testCloudColocationGroupReplicaAllocationIsNull() throws Exception { + String originDeployMode = Config.deploy_mode; + createTable("CREATE TABLE colocate_t3 (k INT) DISTRIBUTED BY HASH(k) BUCKETS 2 " + + "PROPERTIES ('replication_num' = '1', 'colocate_with' = 'g2')"); + createTable("CREATE TABLE colocate_t4 (k INT) DISTRIBUTED BY HASH(k) BUCKETS 2 " + + "PROPERTIES ('replication_num' = '1', 'colocate_with' = 'g2')"); + + Config.deploy_mode = "cloud"; + try { + ProcResult result = new ColocationGroupProcDir().fetchResult(); + int groupNameIdx = ColocationGroupProcDir.TITLE_NAMES.indexOf("GroupName"); + int replicaAllocIdx = ColocationGroupProcDir.TITLE_NAMES.indexOf("ReplicaAllocation"); + List groupRow = result.getRows().stream() + .filter(row -> "test.g2".equals(row.get(groupNameIdx))) + .findFirst() + .orElseThrow(() -> new AssertionError("can not find colocate group test.g2")); + Assertions.assertEquals("null", groupRow.get(replicaAllocIdx)); + } finally { + Config.deploy_mode = originDeployMode; + } + } + + @Test + public void testCloudReplicaProcDisplayExposesPerComputeGroup() { + CloudReplica replica = new CloudReplica(1L, null, ReplicaState.NORMAL, 1L, 1, + db.getId(), 2L, 3L, 4L, 0L); + + Assertions.assertTrue(replica.getClusterToBackendForProcDisplay(Maps.newHashMap()).isEmpty()); + + replica.updateClusterToPrimaryBe("cluster_a", 10001L); + replica.updateClusterToPrimaryBe("cluster_b", 20001L); + + Map clusterToBackend = replica.getClusterToBackendForProcDisplay(Maps.newHashMap()); + Assertions.assertEquals(2, clusterToBackend.size()); + Assertions.assertEquals(Long.valueOf(10001L), clusterToBackend.get("cluster_a")); + Assertions.assertEquals(Long.valueOf(20001L), clusterToBackend.get("cluster_b")); + } + + @Test + public void testCloudColocatedReplicaProcDisplayResolvesPerComputeGroup() throws Exception { + // A colocate cloud replica does not use primaryClusterToBackend (placement is + // resolved on the fly via getColocatedBeId), so that cache stays empty. The proc + // display must still produce one backend per compute group instead of nothing. + CloudReplica replica = Mockito.spy(new CloudReplica(1L, null, ReplicaState.NORMAL, 1L, 1, + db.getId(), 2L, 3L, 4L, 0L)); + + ColocateTableIndex colocateTableIndex = Mockito.mock(ColocateTableIndex.class); + Mockito.when(colocateTableIndex.isColocateTableNoLock(2L)).thenReturn(true); + CloudSystemInfoService infoService = Mockito.mock(CloudSystemInfoService.class); + Mockito.when(infoService.getCloudClusterIds()).thenReturn(Lists.newArrayList("cg_a", "cg_b")); + // Backend list is resolved through the per-clusterId cache; return a non-null list + // so the cache stores it (placement itself is stubbed on getColocatedBeId below). + Mockito.when(infoService.getBackendsByClusterId(Mockito.anyString())).thenReturn(Lists.newArrayList()); + + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class, Mockito.CALLS_REAL_METHODS)) { + mockedEnv.when(Env::getCurrentColocateIndex).thenReturn(colocateTableIndex); + mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(infoService); + Mockito.doReturn(10001L).when(replica).getColocatedBeId(Mockito.eq("cg_a"), Mockito.any()); + // cg_b currently has no available backend; it must be skipped, not fail. + Mockito.doThrow(new ComputeGroupException("no alive be", + ComputeGroupException.FailedTypeEnum.COMPUTE_GROUPS_NO_ALIVE_BE)) + .when(replica).getColocatedBeId(Mockito.eq("cg_b"), Mockito.any()); + + Map clusterToBackend = + replica.getClusterToBackendForProcDisplay(Maps.newHashMap()); + Assertions.assertEquals(1, clusterToBackend.size()); + Assertions.assertEquals(Long.valueOf(10001L), clusterToBackend.get("cg_a")); + Assertions.assertFalse(clusterToBackend.containsKey("cg_b")); + } + } + + @Test + public void testColocateProcDisplayCachesBackendsPerComputeGroup() throws Exception { + // A shared cache makes a single proc call fetch each compute group's backend list + // only once, even when many replicas resolve placement across the same groups. + CloudReplica replica1 = Mockito.spy(new CloudReplica(1L, null, ReplicaState.NORMAL, 1L, 1, + db.getId(), 2L, 3L, 4L, 0L)); + CloudReplica replica2 = Mockito.spy(new CloudReplica(2L, null, ReplicaState.NORMAL, 1L, 1, + db.getId(), 2L, 3L, 4L, 1L)); + + ColocateTableIndex colocateTableIndex = Mockito.mock(ColocateTableIndex.class); + Mockito.when(colocateTableIndex.isColocateTableNoLock(2L)).thenReturn(true); + CloudSystemInfoService infoService = Mockito.mock(CloudSystemInfoService.class); + Mockito.when(infoService.getCloudClusterIds()).thenReturn(Lists.newArrayList("cg_a")); + Mockito.when(infoService.getBackendsByClusterId(Mockito.anyString())).thenReturn(Lists.newArrayList()); + + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class, Mockito.CALLS_REAL_METHODS)) { + mockedEnv.when(Env::getCurrentColocateIndex).thenReturn(colocateTableIndex); + mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(infoService); + Mockito.doReturn(10001L).when(replica1).getColocatedBeId(Mockito.eq("cg_a"), Mockito.any()); + Mockito.doReturn(10002L).when(replica2).getColocatedBeId(Mockito.eq("cg_a"), Mockito.any()); + + Map> computeGroupBackendCache = Maps.newHashMap(); + Assertions.assertEquals(Long.valueOf(10001L), + replica1.getClusterToBackendForProcDisplay(computeGroupBackendCache).get("cg_a")); + Assertions.assertEquals(Long.valueOf(10002L), + replica2.getClusterToBackendForProcDisplay(computeGroupBackendCache).get("cg_a")); + + // Fetched once for cg_a despite two replicas resolving against it. + Mockito.verify(infoService, Mockito.times(1)).getBackendsByClusterId("cg_a"); + } + } + + @Test + public void testColocationGroupDetailPerComputeGroupColumns() throws Exception { + // Two compute groups each get their own column (named by the compute group), and + // within a compute group the per-bucket backend sequence is self-consistent (not + // mixed across groups). + Map>> backendsSeq = Maps.newLinkedHashMap(); + backendsSeq.put("cg_a", Lists.newArrayList( + Lists.newArrayList(10001L), + Lists.newArrayList(10002L))); + backendsSeq.put("cg_b", Lists.newArrayList( + Lists.newArrayList(20001L), + Lists.newArrayList(20002L))); + + ProcResult result = new ColocationGroupBackendSeqsProcNode(backendsSeq).fetchResult(); + + Assertions.assertEquals(Lists.newArrayList("BucketIndex", "cg_a", "cg_b"), + result.getColumnNames()); + Assertions.assertEquals(Lists.newArrayList("0", "10001", "20001"), result.getRows().get(0)); + Assertions.assertEquals(Lists.newArrayList("1", "10002", "20002"), result.getRows().get(1)); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/common/proc/TabletHealthProcDirTest.java b/fe/fe-core/src/test/java/org/apache/doris/common/proc/TabletHealthProcDirTest.java new file mode 100644 index 00000000000000..bd15b33930e6eb --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/common/proc/TabletHealthProcDirTest.java @@ -0,0 +1,110 @@ +// 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.common.proc; + +import org.apache.doris.catalog.Database; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.MaterializedIndex.IndexExtState; +import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.Partition; +import org.apache.doris.catalog.Replica; +import org.apache.doris.catalog.Table; +import org.apache.doris.catalog.Tablet; +import org.apache.doris.catalog.Tablet.TabletStatus; +import org.apache.doris.common.Config; +import org.apache.doris.utframe.TestWithFeService; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.List; +import java.util.stream.Collectors; + +public class TabletHealthProcDirTest extends TestWithFeService { + private Database db; + + @Override + protected int backendNum() { + return 1; + } + + @Override + protected void runBeforeAll() throws Exception { + createDatabase("test"); + useDatabase("test"); + db = Env.getCurrentInternalCatalog().getDbOrMetaException("test"); + } + + @Override + protected void runBeforeEach() throws Exception { + for (Table table : db.getTables()) { + dropTable(table.getName(), true); + } + } + + @Test + public void testCloudModeCountsTabletAsHealthy() throws Exception { + String originDeployMode = Config.deploy_mode; + createTable("CREATE TABLE tbl_proc_health (k INT) DISTRIBUTED BY HASH(k) BUCKETS 1" + + " PROPERTIES ('replication_num' = '1')"); + + OlapTable table = (OlapTable) db.getTableOrMetaException("tbl_proc_health"); + Partition partition = table.getPartitions().iterator().next(); + Tablet tablet = partition.getMaterializedIndices(IndexExtState.ALL).iterator().next() + .getTablets().iterator().next(); + + partition.updateVisibleVersion(10L); + Replica replica = tablet.getReplicas().get(0); + replica.updateVersion(10L); + replica.setBad(true); + + TabletStatus localStatus = tablet.getHealth(Env.getCurrentSystemInfo(), partition.getVisibleVersion(), + table.getPartitionInfo().getReplicaAllocation(partition.getId()), + Env.getCurrentSystemInfo().getAllBackendIds(true)).status; + Assertions.assertEquals(TabletStatus.UNRECOVERABLE, localStatus); + + Config.deploy_mode = "cloud"; + try (MockedStatic mockedPartition = Mockito.mockStatic(Partition.class, Mockito.CALLS_REAL_METHODS)) { + mockedPartition.when(() -> Partition.getVisibleVersions(Mockito.anyList())).thenAnswer(invocation -> { + List partitions = invocation.getArgument(0); + return partitions.stream().map(Partition::getVisibleVersion).collect(Collectors.toList()); + }); + TabletHealthProcDir.DBTabletStatistic statistic = new TabletHealthProcDir.DBTabletStatistic(db); + List row = statistic.toRow(); + int tabletNumIdx = TabletHealthProcDir.TITLE_NAMES.indexOf("TabletNum"); + int healthyNumIdx = TabletHealthProcDir.TITLE_NAMES.indexOf("HealthyNum"); + int unrecoverableNumIdx = TabletHealthProcDir.TITLE_NAMES.indexOf("UnrecoverableNum"); + + Assertions.assertEquals("1", row.get(tabletNumIdx)); + Assertions.assertEquals("1", row.get(healthyNumIdx)); + Assertions.assertEquals("0", row.get(unrecoverableNumIdx)); + } finally { + Config.deploy_mode = originDeployMode; + } + } + + private List findDbRow(ProcResult result, long dbId) { + int dbIdIdx = TabletHealthProcDir.TITLE_NAMES.indexOf("DbId"); + return result.getRows().stream() + .filter(row -> String.valueOf(dbId).equals(row.get(dbIdIdx))) + .findFirst() + .orElseThrow(() -> new AssertionError("can not find db row in tablet health proc result")); + } +}