Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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<GroupId> getAllGroupIds() {
readLock();
try {
Expand Down Expand Up @@ -711,7 +724,11 @@ public List<List<String>> 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<String> cols = groupSchema.getDistributionColTypes().stream().map(
e -> e.toSql()).collect(Collectors.toList());
info.add(Joiner.on(", ").join(cols));
Expand Down
21 changes: 21 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/Replica.java
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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<String, Long> getClusterToBackendForProcDisplay(Map<String, List<Backend>> computeGroupBackendCache) {
Map<String, Long> 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");
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -107,8 +108,17 @@ private boolean isColocated() {
}

public long getColocatedBeId(String clusterId) throws ComputeGroupException {
List<Backend> 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<Backend> clusterBackends) throws ComputeGroupException {
CloudSystemInfoService infoService = ((CloudSystemInfoService) Env.getCurrentSystemInfo());
List<Backend> bes = infoService.getBackendsByClusterId(clusterId).stream()
List<Backend> bes = clusterBackends.stream()
.filter(be -> be.isQueryAvailable()).collect(Collectors.toList());
String clusterName = infoService.getClusterNameByClusterId(clusterId);
if (bes.isEmpty()) {
Expand Down Expand Up @@ -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<String, Long> getClusterToBackendForProcDisplay(Map<String, List<Backend>> computeGroupBackendCache) {
if (!isColocated()) {
return new HashMap<>(primaryClusterToBackend);
}
Map<String, Long> result = new HashMap<>();
CloudSystemInfoService infoService = (CloudSystemInfoService) Env.getCurrentSystemInfo();
for (String clusterId : infoService.getCloudClusterIds()) {
try {
List<Backend> 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;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -30,10 +29,13 @@
* show proc "/colocation_group/group_name";
*/
public class ColocationGroupBackendSeqsProcNode implements ProcNodeInterface {
private Map<Tag, List<List<Long>>> 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<String, List<List<Long>>> backendsSeqByColumn;

public ColocationGroupBackendSeqsProcNode(Map<Tag, List<List<Long>>> backendsSeq) {
this.backendsSeq = backendsSeq;
public ColocationGroupBackendSeqsProcNode(Map<String, List<List<Long>>> backendsSeqByColumn) {
this.backendsSeqByColumn = backendsSeqByColumn;
}

@Override
Expand All @@ -42,21 +44,21 @@ public ProcResult fetchResult() throws AnalysisException {
List<String> 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<String> info = Lists.newArrayList();
info.add(String.valueOf(i)); // bucket index
for (Tag tag : backendsSeq.keySet()) {
List<List<Long>> bucketBackends = backendsSeq.get(tag);
for (String column : backendsSeqByColumn.keySet()) {
List<List<Long>> bucketBackends = backendsSeqByColumn.get(column);
info.add(Joiner.on(", ").join(bucketBackends.get(i)));
}
result.addRow(info);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -61,7 +77,22 @@ public ProcNodeInterface lookup(String groupIdStr) throws AnalysisException {
GroupId groupId = new GroupId(dbId, grpId);
ColocateTableIndex index = Env.getCurrentColocateIndex();
Map<Tag, List<List<Long>>> beSeqs = index.getBackendsPerBucketSeq(groupId);
return new ColocationGroupBackendSeqsProcNode(beSeqs);
Map<String, List<List<Long>>> 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<Tag, List<List<Long>>> entry : beSeqs.entrySet()) {
columns.put(entry.getKey().toString(), entry.getValue());
}
}
}
return new ColocationGroupBackendSeqsProcNode(columns);
}

@Override
Expand All @@ -74,4 +105,129 @@ public ProcResult fetchResult() throws AnalysisException {
result.setRows(infos);
return result;
}

private Map<String, List<List<Long>>> getCloudBackendSeqsFromTablets(GroupId groupId, ColocateTableIndex index) {
Map<String, List<List<Long>>> backendsSeq = Maps.newLinkedHashMap();
List<Long> 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<String, List<List<Long>>> 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<List<Replica>> 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<List<Map<String, Long>>> tabletReplicaBackends = Lists.newArrayListWithCapacity(bucketReplicas.size());
Set<String> 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<String, List<Backend>> computeGroupBackendCache = Maps.newHashMap();
for (List<Replica> replicas : bucketReplicas) {
List<Map<String, Long>> replicaBackends = new ArrayList<>();
for (Replica replica : replicas) {
Map<String, Long> clusterToBackend =
replica.getClusterToBackendForProcDisplay(computeGroupBackendCache);
replicaBackends.add(clusterToBackend);
scopeKeys.addAll(clusterToBackend.keySet());
}
tabletReplicaBackends.add(replicaBackends);
}

Map<String, List<List<Long>>> seqByScopeKey = Maps.newLinkedHashMap();
for (String scopeKey : scopeKeys) {
List<List<Long>> bucketSeq = Lists.newArrayListWithCapacity(tabletReplicaBackends.size());
boolean hasBackend = false;
for (List<Map<String, Long>> replicaBackends : tabletReplicaBackends) {
List<Long> bucketBackends = new ArrayList<>();
for (Map<String, Long> 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<String, List<List<Long>>> backendsSeq = Maps.newLinkedHashMap();
for (Map.Entry<String, List<List<Long>>> 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;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
Loading
Loading