[fix](cloud) Align colocate proc output and tablet health in cloud mode (#60944)

- Fix incorrect tablet health statistics in cloud mode for SHOW PROC '/cluster_health/tablet_health': avoid reporting UNRECOVERABLE due to local-mode health checks.
- Add cloud fallback for colocation group detail: when backend sequence metadata is empty, derive backend ids from current tablets.
- In cloud mode, show ReplicaAllocation as null in SHOW PROC '/colocation_group/{GroupId}' for consistent output semantics.
- Keep local mode behavior unchanged.

Author: deardeng

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<GroupId> getAllGroupIds() {
         readLock();
         try {
@@ -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));

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.
@@ -179,6 +182,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");

fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java

@@ -44,6 +44,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;
@@ -111,8 +112,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()) {
@@ -216,6 +226,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 String getCurrentClusterId() throws ComputeGroupException {
         // Not in a connect session
         String cluster = null;

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<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
@@ -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);

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<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
@@ -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;
+    }
 }

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();