This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new ccbc1e26f56 branch-4.0: [fix](vcg) Fix vcg regression case due to pick
code #56411 (#56503)
ccbc1e26f56 is described below
commit ccbc1e26f56e94a6f855bfce5f051a32a049c932
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Sep 26 17:33:48 2025 +0800
branch-4.0: [fix](vcg) Fix vcg regression case due to pick code #56411
(#56503)
Cherry-picked from #56411
Co-authored-by: deardeng <[email protected]>
---
.../cloud/catalog/CloudInstanceStatusChecker.java | 35 +++++++---------
.../trees/plans/commands/ShowClustersCommand.java | 47 ++++++++++++++++++----
.../trees/plans/commands/WarmUpClusterCommand.java | 34 +++++++++++++++-
.../resource/computegroup/ComputeGroupMgr.java | 10 +++--
.../virtual_compute_group/test_vcg.groovy | 16 ++++----
.../virtual_compute_group/test_vcg_metrics.groovy | 2 +-
6 files changed, 102 insertions(+), 42 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java
index 395fd495bdd..1db4a00379b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java
@@ -17,24 +17,24 @@
package org.apache.doris.cloud.catalog;
-// import org.apache.doris.analysis.WarmUpClusterStmt;
import org.apache.doris.catalog.Env;
-// import org.apache.doris.cloud.CacheHotspotManager;
-// import org.apache.doris.cloud.CloudWarmUpJob;
+import org.apache.doris.cloud.CacheHotspotManager;
+import org.apache.doris.cloud.CloudWarmUpJob;
import org.apache.doris.cloud.proto.Cloud;
import org.apache.doris.cloud.system.CloudSystemInfoService;
-// import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
-// import org.apache.doris.common.DdlException;
+import org.apache.doris.common.DdlException;
import org.apache.doris.common.Pair;
import org.apache.doris.common.util.MasterDaemon;
import org.apache.doris.metric.MetricRepo;
+import org.apache.doris.nereids.trees.plans.commands.WarmUpClusterCommand;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
-// import java.util.Arrays;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -147,14 +147,12 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
}
private void cancelCacheJobs(ComputeGroup vcgInFe, List<String> jobIds) {
- // TODO(dx)
- /*
CacheHotspotManager cacheHotspotManager = ((CloudEnv)
Env.getCurrentEnv()).getCacheHotspotMgr();
for (String jobId : jobIds) {
try {
if (Env.getCurrentEnv().isMaster()) {
// cancel old jobId, will write editlog, so just master
can do
- cacheHotspotManager.cancel(Long.parseLong(jobId));
+ cacheHotspotManager.cancel(Long.parseLong(jobId), "vcg
cancel");
LOG.info("virtual compute group {}, cancel jobId {}",
vcgInFe.getName(), jobId);
}
} catch (DdlException e) {
@@ -162,12 +160,9 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
vcgInFe.getName(), jobId, e);
}
}
- */
}
private void checkNeedRebuildFileCache(ComputeGroup virtualGroupInFe,
List<String> jobIdsInMs) {
- // TODO(dx):
- /*
CacheHotspotManager cacheHotspotManager = ((CloudEnv)
Env.getCurrentEnv()).getCacheHotspotMgr();
// check jobIds in Ms valid, if been cancelled, start new jobs
for (String jobId : jobIdsInMs) {
@@ -210,7 +205,6 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
return;
}
}
- */
}
/**
@@ -222,7 +216,7 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
virtualGroupInFe.getName(), virtualGroupInMs);
return;
}
- // CacheHotspotManager cacheHotspotManager = ((CloudEnv)
Env.getCurrentEnv()).getCacheHotspotMgr();
+ CacheHotspotManager cacheHotspotManager = ((CloudEnv)
Env.getCurrentEnv()).getCacheHotspotMgr();
List<String> jobIdsInMs =
new
ArrayList<>(virtualGroupInMs.getClusterPolicy().getCacheWarmupJobidsList());
@@ -230,8 +224,6 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
LOG.debug("virtual compute group {}, get from ms file cache sync task
jobIds {}",
virtualGroupInFe, jobIdsInMs);
// virtual group has been changed in before step
- // TODO(dx)
- /*
if (virtualGroupInFe.isNeedRebuildFileCache()) {
String srcCg = virtualGroupInFe.getActiveComputeGroup();
String dstCg = virtualGroupInFe.getStandbyComputeGroup();
@@ -247,8 +239,9 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
syncInterValSec = 600;
}
periodicProperties.put("sync_interval_sec",
String.valueOf(syncInterValSec));
- WarmUpClusterStmt periodicStmtPeriodic =
- new WarmUpClusterStmt(dstCg, srcCg, true,
periodicProperties);
+ WarmUpClusterCommand periodicStmtPeriodic =
+ new WarmUpClusterCommand(Collections.emptyList(),
srcCg, dstCg,
+ true, false, periodicProperties);
long jobIdPeriodic =
cacheHotspotManager.createJob(periodicStmtPeriodic);
// load event
@@ -256,8 +249,9 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
// "sync_mode" = "event_driven", "sync_event" = "load"
eventProperties.put("sync_mode", "event_driven");
eventProperties.put("sync_event", "load");
- WarmUpClusterStmt eventStmtPeriodic =
- new WarmUpClusterStmt(dstCg, srcCg, true,
eventProperties);
+ WarmUpClusterCommand eventStmtPeriodic =
+ new WarmUpClusterCommand(Collections.emptyList(),
srcCg, dstCg,
+ true, false, eventProperties);
long jobIdEvent =
cacheHotspotManager.createJob(eventStmtPeriodic);
// send jobIds to ms
List<String> newJobIds =
Arrays.asList(Long.toString(jobIdPeriodic), Long.toString(jobIdEvent));
@@ -270,7 +264,6 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
}
virtualGroupInFe.setNeedRebuildFileCache(false);
}
- */
}
private void handleExistingVirtualComputeGroup(Cloud.ClusterPB
clusterInMs, ComputeGroup virtualGroupInFe) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowClustersCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowClustersCommand.java
index 27b1d20f1d2..da61a861ce6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowClustersCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowClustersCommand.java
@@ -22,6 +22,7 @@ import org.apache.doris.analysis.ResourceTypeEnum;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.cloud.catalog.ComputeGroup;
import org.apache.doris.cloud.qe.ComputeGroupException;
import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.cluster.ClusterNamespace;
@@ -50,16 +51,18 @@ import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
+import java.util.stream.Collectors;
/**
* Represents the command for SHOW CLUSTERS.
*/
public class ShowClustersCommand extends ShowCommand {
+ // sql: show clusters;
public static final ImmutableList<String> CLUSTER_TITLE_NAMES = new
ImmutableList.Builder<String>()
-
.add("cluster").add("is_current").add("users").add("backend_num").build();
-
+
.add("cluster").add("is_current").add("users").add("backend_num").add("sub_clusters").add("policy").build();
+ // sql: show compute groups;
public static final ImmutableList<String> COMPUTE_GROUP_TITLE_NAMES = new
ImmutableList.Builder<String>()
-
.add("Name").add("IsCurrent").add("Users").add("BackendNum").build();
+
.add("Name").add("IsCurrent").add("Users").add("BackendNum").add("SubComputeGroups").add("Policy").build();
private static final Logger LOG =
LogManager.getLogger(ShowClustersCommand.class);
private final boolean isComputeGroup;
@@ -89,7 +92,14 @@ public class ShowClustersCommand extends ShowCommand {
}
List<String> clusterNames = null;
- clusterNames = ((CloudSystemInfoService)
Env.getCurrentSystemInfo()).getCloudClusterNames();
+ CloudSystemInfoService cloudSys = ((CloudSystemInfoService)
Env.getCurrentSystemInfo());
+ clusterNames = cloudSys.getCloudClusterNames();
+ // virtual cluster info
+ List<ComputeGroup> virtualComputeGroup =
cloudSys.getComputeGroups(true);
+ List<String> virtualComputeGroupNames = virtualComputeGroup.stream()
+ .map(ComputeGroup::getName).collect(Collectors.toList());
+
+ clusterNames.addAll(virtualComputeGroupNames);
final Set<String> clusterNameSet = Sets.newTreeSet();
clusterNameSet.addAll(clusterNames);
@@ -122,10 +132,33 @@ public class ShowClustersCommand extends ShowCommand {
String result = Joiner.on(", ").join(users);
row.add(result);
- int backendNum = ((CloudSystemInfoService)
Env.getCurrentEnv().getCurrentSystemInfo())
- .getBackendsByClusterName(clusterName).size();
- row.add(String.valueOf(backendNum));
+ // subClusters
+ String subClusterNames = "";
+ // Policy
+ String policy = "";
+ if (!virtualComputeGroupNames.contains(clusterName)) {
+ int backendNum =
cloudSys.getBackendsByClusterName(clusterName).size();
+ row.add(String.valueOf(backendNum));
+ rows.add(row);
+ row.add(subClusterNames);
+ row.add(policy);
+ continue;
+ }
+ // virtual compute group
+ // virtual cg backends eq 0
+ row.add(String.valueOf(0));
rows.add(row);
+ ComputeGroup cg = cloudSys.getComputeGroupByName(clusterName);
+ if (cg == null) {
+ continue;
+ }
+ String activeCluster = cg.getPolicy().getActiveComputeGroup();
+ String standbyCluster = cg.getPolicy().getStandbyComputeGroup();
+ // first active, second standby
+ subClusterNames = Joiner.on(", ").join(activeCluster,
standbyCluster);
+ row.add(subClusterNames);
+ // Policy
+ row.add(cg.getPolicy().toString());
}
return new ShowResultSet(getMetaData(), rows);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/WarmUpClusterCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/WarmUpClusterCommand.java
index a66c703d4cd..52c3a4af740 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/WarmUpClusterCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/WarmUpClusterCommand.java
@@ -23,6 +23,7 @@ import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.ScalarType;
import org.apache.doris.cloud.catalog.CloudEnv;
+import org.apache.doris.cloud.catalog.ComputeGroup;
import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
@@ -120,17 +121,46 @@ public class WarmUpClusterCommand extends Command
implements ForwardWithSync {
handleWarmUp(ctx, executor);
}
+ private void checkWarmupCgs(CloudSystemInfoService cloudSys) throws
AnalysisException {
+ if (!Strings.isNullOrEmpty(srcCluster)) {
+ ComputeGroup srcCg = cloudSys.getComputeGroupByName(srcCluster);
+ if (srcCg != null && srcCg.isVirtual()) {
+ throw new AnalysisException("The srcClusterName " + srcCluster
+ + " is a virtual compute group, not support");
+ }
+ }
+
+ if (!Strings.isNullOrEmpty(dstCluster)) {
+ ComputeGroup dstCg = cloudSys.getComputeGroupByName(dstCluster);
+ if (dstCg != null && dstCg.isVirtual()) {
+ throw new AnalysisException("The dstClusterName " + dstCluster
+ + " is a virtual compute group, not support");
+ }
+ }
+
+ if (!Strings.isNullOrEmpty(srcCluster) &&
!Strings.isNullOrEmpty(dstCluster)) {
+ String srcMayOwnedVcg =
cloudSys.ownedByVirtualComputeGroup(srcCluster);
+ String dstMayOwnedVcg =
cloudSys.ownedByVirtualComputeGroup(srcCluster);
+ if (srcMayOwnedVcg != null &&
srcMayOwnedVcg.equals(dstMayOwnedVcg)) {
+ throw new AnalysisException("The srcClusterName " + srcCluster
+ " dstClusterName " + dstCluster
+ + " is owned by virtual compute group " + srcMayOwnedVcg +
" not support");
+ }
+ }
+ }
+
/**
* validate
*/
public void validate(ConnectContext connectContext) throws UserException {
if (!Config.isCloudMode()) {
- throw new UserException("The sql is illegal in disk mode ");
+ throw new UserException("The sql is just support in cloud mode");
}
- if (!((CloudSystemInfoService)
Env.getCurrentSystemInfo()).containClusterName(dstCluster)) {
+ CloudSystemInfoService cloudSys = ((CloudSystemInfoService)
Env.getCurrentSystemInfo());
+ if (!cloudSys.containClusterName(dstCluster)) {
throw new AnalysisException("The dstClusterName " + dstCluster + "
doesn't exist");
}
+ checkWarmupCgs(cloudSys);
if (!isWarmUpWithTable
&& !((CloudSystemInfoService)
Env.getCurrentSystemInfo()).containClusterName(srcCluster)) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/ComputeGroupMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/ComputeGroupMgr.java
index 3b31f7e2bd2..e7a58ead33b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/ComputeGroupMgr.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/ComputeGroupMgr.java
@@ -17,6 +17,7 @@
package org.apache.doris.resource.computegroup;
+import org.apache.doris.catalog.Env;
import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.Config;
import org.apache.doris.common.UserException;
@@ -40,11 +41,14 @@ public class ComputeGroupMgr {
public ComputeGroup getComputeGroupByName(String name) throws
UserException {
if (Config.isCloudMode()) {
CloudSystemInfoService cloudSystemInfoService =
(CloudSystemInfoService) systemInfoService;
- String clusterId =
cloudSystemInfoService.getCloudClusterIdByName(name);
+ String physicalClusterName = ((CloudSystemInfoService)
Env.getCurrentSystemInfo())
+ .getPhysicalCluster(name);
+ String clusterId =
cloudSystemInfoService.getCloudClusterIdByName(physicalClusterName);
if (StringUtils.isEmpty(clusterId)) {
- throw new UserException("Can not find compute group:" + name);
+ throw new UserException("Can not find compute group:" + name
+ + " real compute group name:" + physicalClusterName);
}
- return new CloudComputeGroup(clusterId, name,
cloudSystemInfoService);
+ return new CloudComputeGroup(clusterId, physicalClusterName,
cloudSystemInfoService);
} else {
return new ComputeGroup(name, name, systemInfoService);
}
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg.groovy
index 74aef9e628e..88cb8ed4e1d 100644
---
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg.groovy
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg.groovy
@@ -198,7 +198,7 @@ suite('test_vcg', 'multi_cluster,docker') {
// test manual cancel warm up job, generate new jobs
sql """CANCEL WARM UP JOB WHERE ID=${showWarmup[0].JobId}"""
- dockerAwaitUntil(50, 3) {
+ awaitUntil(50, 3) {
showWarmup = sql_return_maparray """SHOW WARM UP JOB"""
// cancel 2, generate 2
showWarmup.size() == 4
@@ -295,7 +295,7 @@ suite('test_vcg', 'multi_cluster,docker') {
assertTrue(json.code.equalsIgnoreCase("OK"))
}
- dockerAwaitUntil(20) {
+ awaitUntil(20) {
showComputeGroup = sql_return_maparray """ SHOW COMPUTE GROUPS
"""
log.info("show compute group after alter {}", showComputeGroup)
vcgInShow = showComputeGroup.find { it.Name ==
normalVclusterName }
@@ -307,7 +307,7 @@ suite('test_vcg', 'multi_cluster,docker') {
assertTrue(vcgInShow.Policy.contains('"activeComputeGroup":"newcluster2","standbyComputeGroup":"newcluster1"'))
// alter active -> sync to fe -> cancel old jobs -> generate new
jobs
// -> sync to ms -> sync to fe -> save new jobs
- dockerAwaitUntil(50, 3) {
+ awaitUntil(50, 3) {
showWarmup = sql_return_maparray """SHOW WARM UP JOB"""
// cancel 2, generate 2
showWarmup.size() == 6
@@ -326,7 +326,7 @@ suite('test_vcg', 'multi_cluster,docker') {
cluster.stopBackends(6)
// test warm up job destory and generate new jobs
- dockerAwaitUntil(50, 3) {
+ awaitUntil(50, 3) {
sql """USE @${normalVclusterName}"""
sql """select count(*) from ${tbl}"""
showComputeGroup = sql_return_maparray """ SHOW COMPUTE GROUPS
"""
@@ -337,7 +337,7 @@ suite('test_vcg', 'multi_cluster,docker') {
assertEquals(showWarmup.size(), 6)
cluster.startBackends(6)
- dockerAwaitUntil(50, 3) {
+ awaitUntil(50, 3) {
showWarmup = sql_return_maparray """SHOW WARM UP JOB"""
showWarmup.size() == 8
}
@@ -350,7 +350,7 @@ suite('test_vcg', 'multi_cluster,docker') {
// test rename vcg
def newNormalVclusterName = "newNormalVirtualClusterName"
rename_cloud_cluster.call(newNormalVclusterName, normalVclusterId,
ms)
- dockerAwaitUntil(20) {
+ awaitUntil(20) {
showComputeGroup = sql_return_maparray """ SHOW COMPUTE GROUPS
"""
log.info("show compute group after rename {}",
showComputeGroup)
vcgInShow = showComputeGroup.find { it.Name ==
normalVclusterName }
@@ -365,7 +365,7 @@ suite('test_vcg', 'multi_cluster,docker') {
// rename back to
rename_cloud_cluster.call(normalVclusterName, normalVclusterId, ms)
- dockerAwaitUntil(20) {
+ awaitUntil(20) {
showComputeGroup = sql_return_maparray """ SHOW COMPUTE GROUPS
"""
log.info("show compute group after rename back {}",
showComputeGroup)
vcgInShow = showComputeGroup.find { it.Name ==
normalVclusterName }
@@ -393,7 +393,7 @@ suite('test_vcg', 'multi_cluster,docker') {
jsonObject = jsonSlurper.parseText(tag)
cloudClusterId = jsonObject.compute_group_id
drop_cluster(clusterName2, cloudClusterId, ms)
- dockerAwaitUntil(20) {
+ awaitUntil(20) {
def showRet = sql """SHOW COMPUTE GROUPS"""
log.info("show cgs: {}", showRet)
showRet.size() == 2
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_metrics.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_metrics.groovy
index 9939539568c..b698799b753 100644
---
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_metrics.groovy
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_metrics.groovy
@@ -139,7 +139,7 @@ suite('test_vcg_metrics', 'multi_cluster,docker') {
cluster.stopBackends(4, 5)
// test warm up job destory and generate new jobs
- dockerAwaitUntil(50, 3) {
+ awaitUntil(50, 3) {
sql """USE @${normalVclusterName}"""
sql """select count(*) from ${tbl}"""
showComputeGroup = sql_return_maparray """ SHOW COMPUTE GROUPS
"""
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
