You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by wa...@apache.org on 2025/04/28 12:55:54 UTC
(doris) branch master updated: [Feature]bind workload group to compute group (#49884)
This is an automated email from the ASF dual-hosted git repository.
wangbo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 3c44f18ddd5 [Feature]bind workload group to compute group (#49884)
3c44f18ddd5 is described below
commit 3c44f18ddd5f44f1d61eb3a7b40dce93b30a2dc0
Author: wangbo <wa...@selectdb.com>
AuthorDate: Mon Apr 28 20:55:40 2025 +0800
[Feature]bind workload group to compute group (#49884)
---
.../schema_workload_groups_scanner.cpp | 2 +-
.../main/java/org/apache/doris/common/Config.java | 4 -
.../antlr4/org/apache/doris/nereids/DorisParser.g4 | 6 +-
.../doris/analysis/AlterRoutineLoadStmt.java | 26 +-
.../doris/analysis/CreateRoutineLoadStmt.java | 16 +-
.../main/java/org/apache/doris/catalog/Env.java | 3 +
.../java/org/apache/doris/catalog/SchemaTable.java | 2 +-
.../java/org/apache/doris/common/FeConstants.java | 3 +-
.../doris/common/publish/TopicPublisherThread.java | 6 +-
.../org/apache/doris/load/loadv2/LoadManager.java | 9 +-
.../doris/load/routineload/KafkaTaskInfo.java | 49 +-
.../doris/nereids/parser/LogicalPlanBuilder.java | 9 +-
.../plans/commands/AlterWorkloadGroupCommand.java | 36 +-
.../plans/commands/CreateWorkloadGroupCommand.java | 46 +-
.../plans/commands/DropWorkloadGroupCommand.java | 27 +-
.../plans/commands/info/CreateRoutineLoadInfo.java | 18 +-
.../java/org/apache/doris/qe/ConnectContext.java | 6 +-
.../main/java/org/apache/doris/qe/Coordinator.java | 9 +-
.../org/apache/doris/qe/NereidsCoordinator.java | 11 +-
.../computegroup/AllBackendComputeGroup.java | 14 +
.../resource/computegroup/CloudComputeGroup.java | 8 +
.../doris/resource/computegroup/ComputeGroup.java | 6 +
.../resource/computegroup/ComputeGroupMgr.java | 20 +-
.../resource/computegroup/MergedComputeGroup.java | 10 +-
.../workloadgroup/BindWgToComputeGroupThread.java | 98 ++++
.../resource/workloadgroup/WorkloadGroup.java | 47 +-
.../resource/workloadgroup/WorkloadGroupMgr.java | 439 +++++++-------
.../WorkloadSchedPolicyMgr.java | 21 +-
.../apache/doris/service/FrontendServiceImpl.java | 3 +-
.../main/java/org/apache/doris/system/Backend.java | 39 +-
.../doris/tablefunction/MetadataGenerator.java | 2 +-
.../apache/doris/catalog/CreateEncryptKeyTest.java | 1 +
.../apache/doris/planner/ResourceTagQueryTest.java | 6 +-
.../apache/doris/resource/ComputeGroupTest.java | 14 +-
.../workloadgroup/WorkloadGroupMgrTest.java | 645 ++++++++++++---------
.../resource/workloadgroup/WorkloadGroupTest.java | 4 +-
.../apache/doris/utframe/TestWithFeService.java | 3 +-
.../data/workload_manager_p0/test_curd_wlg.out | Bin 3661 -> 3559 bytes
regression-test/pipeline/cloud_p0/deploy.sh | 2 +
regression-test/pipeline/common/doris-utils.sh | 10 +
regression-test/pipeline/vault_p0/deploy.sh | 2 +
.../test_master_slave_consistency_auth.groovy | 16 +-
regression-test/suites/auth_up_down_p0/load.groovy | 18 +-
.../suites/load_p0/broker_load/test_s3_load.groovy | 13 +-
.../load_p0/routine_load/test_routine_load.groovy | 14 +-
.../broker_load/test_s3_load_properties.groovy | 13 +-
.../load_p2/routine_load/test_routine_load.groovy | 14 +-
.../workload_manager_p0/test_curd_wlg.groovy | 420 ++++++--------
.../test_workload_sched_policy.groovy | 44 +-
49 files changed, 1318 insertions(+), 916 deletions(-)
diff --git a/be/src/exec/schema_scanner/schema_workload_groups_scanner.cpp b/be/src/exec/schema_scanner/schema_workload_groups_scanner.cpp
index bc5fb61669c..b9b63a3bc6f 100644
--- a/be/src/exec/schema_scanner/schema_workload_groups_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_workload_groups_scanner.cpp
@@ -43,7 +43,7 @@ std::vector<SchemaScanner::ColumnDesc> SchemaWorkloadGroupsScanner::_s_tbls_colu
{"MIN_REMOTE_SCAN_THREAD_NUM", TYPE_BIGINT, sizeof(int64_t), true},
{"MEMORY_LOW_WATERMARK", TYPE_VARCHAR, sizeof(StringRef), true},
{"MEMORY_HIGH_WATERMARK", TYPE_VARCHAR, sizeof(StringRef), true},
- {"TAG", TYPE_VARCHAR, sizeof(StringRef), true},
+ {"COMPUTE_GROUP", TYPE_VARCHAR, sizeof(StringRef), true},
{"READ_BYTES_PER_SECOND", TYPE_BIGINT, sizeof(int64_t), true},
{"REMOTE_READ_BYTES_PER_SECOND", TYPE_BIGINT, sizeof(int64_t), true},
};
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 15d31a50935..90ee92c0c87 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -2029,10 +2029,6 @@ public class Config extends ConfigBase {
@ConfField(mutable = true)
public static boolean enable_query_queue = true;
- // used for regression test
- @ConfField(mutable = true)
- public static boolean enable_alter_queue_prop_sync = false;
-
@ConfField(mutable = true)
public static long query_queue_update_interval_ms = 5000;
diff --git a/fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4 b/fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4
index 4a82cc9e28e..7679e7354d9 100644
--- a/fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4
+++ b/fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4
@@ -192,7 +192,7 @@ supportedCreateStatement
(WITH ROLLUP (rollupNames=identifierList)?)? #createTableLike
| CREATE ROLE (IF NOT EXISTS)? name=identifierOrText (COMMENT STRING_LITERAL)? #createRole
| CREATE WORKLOAD GROUP (IF NOT EXISTS)?
- name=identifierOrText properties=propertyClause? #createWorkloadGroup
+ name=identifierOrText (FOR computeGroup=identifierOrText)? properties=propertyClause? #createWorkloadGroup
| CREATE CATALOG (IF NOT EXISTS)? catalogName=identifier
(WITH RESOURCE resourceName=identifier)?
(COMMENT STRING_LITERAL)? properties=propertyClause? #createCatalog
@@ -235,7 +235,7 @@ supportedAlterStatement
| ALTER CATALOG name=identifier RENAME newName=identifier #alterCatalogRename
| ALTER ROLE role=identifierOrText commentSpec #alterRole
| ALTER STORAGE VAULT name=multipartIdentifier properties=propertyClause #alterStorageVault
- | ALTER WORKLOAD GROUP name=identifierOrText
+ | ALTER WORKLOAD GROUP name=identifierOrText (FOR computeGroup=identifierOrText)?
properties=propertyClause? #alterWorkloadGroup
| ALTER CATALOG name=identifier SET PROPERTIES
LEFT_PAREN propertyItemList RIGHT_PAREN #alterCatalogProperties
@@ -269,7 +269,7 @@ supportedDropStatement
| DROP SQL_BLOCK_RULE (IF EXISTS)? identifierSeq #dropSqlBlockRule
| DROP USER (IF EXISTS)? userIdentify #dropUser
| DROP STORAGE POLICY (IF EXISTS)? name=identifier #dropStoragePolicy
- | DROP WORKLOAD GROUP (IF EXISTS)? name=identifierOrText #dropWorkloadGroup
+ | DROP WORKLOAD GROUP (IF EXISTS)? name=identifierOrText (FOR computeGroup=identifierOrText)? #dropWorkloadGroup
| DROP CATALOG (IF EXISTS)? name=identifier #dropCatalog
| DROP FILE name=STRING_LITERAL
((FROM | IN) database=identifier)? properties=propertyClause #dropFile
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/AlterRoutineLoadStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/AlterRoutineLoadStmt.java
index f383e02aee1..a45717a485d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/AlterRoutineLoadStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/AlterRoutineLoadStmt.java
@@ -35,6 +35,9 @@ import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
import lombok.Getter;
import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
import java.util.Map;
import java.util.Optional;
@@ -50,6 +53,8 @@ import java.util.Optional;
*/
public class AlterRoutineLoadStmt extends DdlStmt implements NotFallbackInParser {
+ private static final Logger LOG = LogManager.getLogger(AlterRoutineLoadStmt.class);
+
private static final String NAME_TYPE = "ROUTINE LOAD NAME";
private static final ImmutableSet<String> CONFIGURABLE_JOB_PROPERTIES_SET = new ImmutableSet.Builder<String>()
@@ -248,9 +253,24 @@ public class AlterRoutineLoadStmt extends DdlStmt implements NotFallbackInParser
}
if (jobProperties.containsKey(CreateRoutineLoadStmt.WORKLOAD_GROUP)) {
String workloadGroup = jobProperties.get(CreateRoutineLoadStmt.WORKLOAD_GROUP);
- long wgId = Env.getCurrentEnv().getWorkloadGroupMgr()
- .getWorkloadGroup(ConnectContext.get().getCurrentUserIdentity(), workloadGroup);
- analyzedJobProperties.put(CreateRoutineLoadStmt.WORKLOAD_GROUP, String.valueOf(wgId));
+ if (!StringUtils.isEmpty(workloadGroup)) {
+ ConnectContext tmpCtx = new ConnectContext();
+ tmpCtx.setCurrentUserIdentity(ConnectContext.get().getCurrentUserIdentity());
+ tmpCtx.getSessionVariable().setWorkloadGroup(workloadGroup);
+ long wgId;
+ try {
+ // NOTE(wb): why get 0th wg here;
+ // currently a routineload can only be executed in one workload group;
+ // but the workload group thrift sent from FE to BE is a list, this is for scalability.
+ wgId = Env.getCurrentEnv().getWorkloadGroupMgr()
+ .getWorkloadGroup(tmpCtx).get(0)
+ .getId();
+ } catch (Throwable t) {
+ LOG.info("Get workload group failed when alter routine load,", t);
+ throw t;
+ }
+ analyzedJobProperties.put(CreateRoutineLoadStmt.WORKLOAD_GROUP, String.valueOf(wgId));
+ }
}
if (jobProperties.containsKey(LoadStmt.KEY_ENCLOSE)) {
analyzedJobProperties.put(LoadStmt.KEY_ENCLOSE, jobProperties.get(LoadStmt.KEY_ENCLOSE));
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java
index 27379ccd28d..9654a28b758 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java
@@ -582,8 +582,20 @@ public class CreateRoutineLoadStmt extends DdlStmt implements NotFallbackInParse
String inputWorkloadGroupStr = jobProperties.get(WORKLOAD_GROUP);
if (!StringUtils.isEmpty(inputWorkloadGroupStr)) {
- this.workloadGroupId = Env.getCurrentEnv().getWorkloadGroupMgr()
- .getWorkloadGroup(ConnectContext.get().getCurrentUserIdentity(), inputWorkloadGroupStr);
+ ConnectContext tmpCtx = new ConnectContext();
+ tmpCtx.setCurrentUserIdentity(ConnectContext.get().getCurrentUserIdentity());
+ tmpCtx.getSessionVariable().setWorkloadGroup(inputWorkloadGroupStr);
+ try {
+ // NOTE(wb): why get 0th wg here;
+ // currently a routineload can only be executed in one workload group;
+ // but the workload group thrift sent from FE to BE is a list, this is for scalability.
+ this.workloadGroupId = Env.getCurrentEnv().getWorkloadGroupMgr()
+ .getWorkloadGroup(tmpCtx).get(0)
+ .getId();
+ } catch (Throwable t) {
+ LOG.info("Get workload group failed when create routine load,", t);
+ throw t;
+ }
}
if (ConnectContext.get() != null) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
index 6a0dd429feb..e9e2b268871 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
@@ -261,6 +261,7 @@ import org.apache.doris.qe.VariableMgr;
import org.apache.doris.resource.AdmissionControl;
import org.apache.doris.resource.Tag;
import org.apache.doris.resource.computegroup.ComputeGroupMgr;
+import org.apache.doris.resource.workloadgroup.BindWgToComputeGroupThread;
import org.apache.doris.resource.workloadgroup.WorkloadGroupMgr;
import org.apache.doris.resource.workloadschedpolicy.WorkloadRuntimeStatusMgr;
import org.apache.doris.resource.workloadschedpolicy.WorkloadSchedPolicyMgr;
@@ -1956,6 +1957,8 @@ public class Env {
topicPublisherThread.addToTopicPublisherList(wpPublisher);
topicPublisherThread.start();
+ new BindWgToComputeGroupThread().start();
+
// auto analyze related threads.
statisticsCleaner.start();
statisticsAutoCollector.start();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java
index 13c44f2fed8..820c04bbdc6 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java
@@ -494,7 +494,7 @@ public class SchemaTable extends Table {
.column("MIN_REMOTE_SCAN_THREAD_NUM", ScalarType.createType(PrimitiveType.BIGINT))
.column("MEMORY_LOW_WATERMARK", ScalarType.createVarchar(256))
.column("MEMORY_HIGH_WATERMARK", ScalarType.createVarchar(256))
- .column("TAG", ScalarType.createVarchar(256))
+ .column("COMPUTE_GROUP", ScalarType.createVarchar(256))
.column("READ_BYTES_PER_SECOND", ScalarType.createType(PrimitiveType.BIGINT))
.column("REMOTE_READ_BYTES_PER_SECOND", ScalarType.createType(PrimitiveType.BIGINT))
.build()))
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/FeConstants.java b/fe/fe-core/src/main/java/org/apache/doris/common/FeConstants.java
index 214fbe0e410..c4d095ed727 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/FeConstants.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/FeConstants.java
@@ -50,8 +50,7 @@ public class FeConstants {
// set to false to disable internal schema db
public static boolean enableInternalSchemaDb = true;
- // for UT, create internal workload group thread can not start
- public static boolean shouldCreateInternalWorkloadGroup = true;
+ public static boolean bindWgToComputeGroup = true;
// default scheduler interval is 10 seconds
public static int default_scheduler_interval_millisecond = 10000;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/publish/TopicPublisherThread.java b/fe/fe-core/src/main/java/org/apache/doris/common/publish/TopicPublisherThread.java
index df3b06e8271..5eccf24fa90 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/publish/TopicPublisherThread.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/publish/TopicPublisherThread.java
@@ -38,7 +38,6 @@ import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.concurrent.ExecutorService;
public class TopicPublisherThread extends MasterDaemon {
@@ -143,14 +142,13 @@ public class TopicPublisherThread extends MasterDaemon {
Map<TTopicInfoType, List<TopicInfo>> topicMap = copiedRequest.getTopicMap();
List<TopicInfo> topicInfoList = topicMap.get(TTopicInfoType.WORKLOAD_GROUP);
if (topicInfoList != null) {
- Set<String> beTagSet = be.getBeWorkloadGroupTagSet();
+ String beComputeGroup = be.getComputeGroup();
Iterator<TopicInfo> topicIter = topicInfoList.iterator();
while (topicIter.hasNext()) {
TopicInfo topicInfo = topicIter.next();
if (topicInfo.isSetWorkloadGroupInfo()) {
TWorkloadGroupInfo tWgInfo = topicInfo.getWorkloadGroupInfo();
- if (tWgInfo.isSetTag() && !Backend.isMatchWorkloadGroupTag(
- tWgInfo.getTag(), beTagSet)) {
+ if (tWgInfo.isSetTag() && !tWgInfo.getTag().equals(beComputeGroup)) {
// currently TopicInfo could not contain both policy and workload group,
// so we can remove TopicInfo directly.
topicIter.remove();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java
index 5a3f5ffef50..ed56bdabcd7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java
@@ -147,8 +147,13 @@ public class LoadManager implements Writable {
}
if (Config.enable_workload_group) {
- loadJob.settWorkloadGroups(
- Env.getCurrentEnv().getWorkloadGroupMgr().getWorkloadGroup(ConnectContext.get()));
+ try {
+ loadJob.settWorkloadGroups(
+ Env.getCurrentEnv().getWorkloadGroupMgr().getWorkloadGroup(ConnectContext.get()));
+ } catch (Throwable t) {
+ LOG.info("Get workload group failed when create load job,", t);
+ throw t;
+ }
}
Env.getCurrentEnv().getEditLog().logCreateLoadJob(loadJob);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
index e3292dc671f..79f8101422c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
@@ -21,10 +21,12 @@ import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Table;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.Config;
import org.apache.doris.common.UserException;
import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.planner.StreamLoadPlanner;
+import org.apache.doris.qe.ConnectContext;
import org.apache.doris.thrift.TFileFormatType;
import org.apache.doris.thrift.TKafkaLoadInfo;
import org.apache.doris.thrift.TLoadSourceType;
@@ -35,7 +37,10 @@ import org.apache.doris.thrift.TRoutineLoadTask;
import org.apache.doris.thrift.TUniqueId;
import com.google.common.base.Joiner;
+import com.google.common.base.Strings;
import com.google.gson.Gson;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
import java.util.List;
@@ -45,6 +50,8 @@ import java.util.UUID;
public class KafkaTaskInfo extends RoutineLoadTaskInfo {
private RoutineLoadManager routineLoadManager = Env.getCurrentEnv().getRoutineLoadManager();
+ private static final Logger LOG = LogManager.getLogger(KafkaTaskInfo.class);
+
// <partitionId, offset to be consumed>
private Map<Integer, Long> partitionIdToOffset;
@@ -138,21 +145,39 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
tPlanFragment.getOutputSink().getOlapTableSink().setTxnId(txnId);
if (Config.enable_workload_group) {
- long wgId = routineLoadJob.getWorkloadId();
- List<TPipelineWorkloadGroup> tWgList = new ArrayList<>();
- if (wgId > 0) {
- tWgList = Env.getCurrentEnv().getWorkloadGroupMgr()
- .getTWorkloadGroupById(wgId);
+ try {
+ long wgId = routineLoadJob.getWorkloadId();
+ List<TPipelineWorkloadGroup> tWgList = new ArrayList<>();
+ if (wgId > 0) {
+ tWgList = Env.getCurrentEnv().getWorkloadGroupMgr()
+ .getTWorkloadGroupById(wgId);
+ }
+
if (tWgList.size() == 0) {
- throw new UserException("can not find workload group, id=" + wgId);
+ ConnectContext tmpContext = new ConnectContext();
+ if (Config.isCloudMode()) {
+ String clusterName = ((CloudSystemInfoService) Env.getCurrentSystemInfo())
+ .getClusterNameByClusterId(routineLoadJob.getCloudClusterId());
+ if (Strings.isNullOrEmpty(clusterName)) {
+ String err = String.format("cluster name is empty, cluster id is %s",
+ routineLoadJob.getCloudClusterId());
+ LOG.warn(err);
+ throw new UserException(err);
+ }
+ tmpContext.setCloudCluster(clusterName);
+ }
+ tmpContext.setCurrentUserIdentity(routineLoadJob.getUserIdentity());
+ tmpContext.setQualifiedUser(routineLoadJob.getUserIdentity().getQualifiedUser());
+ tWgList = Env.getCurrentEnv().getWorkloadGroupMgr().getWorkloadGroup(tmpContext);
}
- } else {
- tWgList = Env.getCurrentEnv().getWorkloadGroupMgr()
- .getWorkloadGroupByUser(routineLoadJob.getUserIdentity(), false);
- }
- if (tWgList.size() != 0) {
- tExecPlanFragmentParams.setWorkloadGroups(tWgList);
+ if (tWgList.size() != 0) {
+ tExecPlanFragmentParams.setWorkloadGroups(tWgList);
+ }
+ } catch (Throwable t) {
+ LOG.info("Get workload group failed when replan kafka, job id:{} , ", routineLoadJob.getTxnId(), t);
+ throw t;
}
+
}
return tExecPlanFragmentParams;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java
index 474595caaad..ab97bcdc989 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java
@@ -5469,9 +5469,10 @@ public class LogicalPlanBuilder extends DorisParserBaseVisitor<Object> {
@Override
public LogicalPlan visitAlterWorkloadGroup(AlterWorkloadGroupContext ctx) {
+ String cgName = ctx.computeGroup == null ? "" : stripQuotes(ctx.computeGroup.getText());
Map<String, String> properties = ctx.propertyClause() != null
? Maps.newHashMap(visitPropertyClause(ctx.propertyClause())) : Maps.newHashMap();
- return new AlterWorkloadGroupCommand(ctx.name.getText(), properties);
+ return new AlterWorkloadGroupCommand(cgName, ctx.name.getText(), properties);
}
@Override
@@ -5788,10 +5789,11 @@ public class LogicalPlanBuilder extends DorisParserBaseVisitor<Object> {
@Override
public LogicalPlan visitCreateWorkloadGroup(CreateWorkloadGroupContext ctx) {
String workloadGroupName = stripQuotes(ctx.name.getText());
+ String cgName = ctx.computeGroup == null ? "" : stripQuotes(ctx.computeGroup.getText());
boolean ifNotExists = ctx.EXISTS() != null;
Map<String, String> properties = ctx.propertyClause() != null
? Maps.newHashMap(visitPropertyClause(ctx.propertyClause())) : Maps.newHashMap();
- return new CreateWorkloadGroupCommand(workloadGroupName, ifNotExists, properties);
+ return new CreateWorkloadGroupCommand(cgName, workloadGroupName, ifNotExists, properties);
}
@Override
@@ -5833,7 +5835,8 @@ public class LogicalPlanBuilder extends DorisParserBaseVisitor<Object> {
@Override
public LogicalPlan visitDropWorkloadGroup(DropWorkloadGroupContext ctx) {
- return new DropWorkloadGroupCommand(ctx.name.getText(), ctx.EXISTS() != null);
+ String cgName = ctx.computeGroup == null ? "" : stripQuotes(ctx.computeGroup.getText());
+ return new DropWorkloadGroupCommand(cgName, ctx.name.getText(), ctx.EXISTS() != null);
}
@Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterWorkloadGroupCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterWorkloadGroupCommand.java
index 6fc1c0a74d3..67ca191dd99 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterWorkloadGroupCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterWorkloadGroupCommand.java
@@ -18,16 +18,19 @@
package org.apache.doris.nereids.trees.plans.commands;
import org.apache.doris.catalog.Env;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.resource.Tag;
import org.apache.doris.resource.workloadgroup.WorkloadGroup;
-import org.apache.doris.resource.workloadgroup.WorkloadGroupMgr;
import org.apache.commons.lang3.StringUtils;
@@ -39,14 +42,18 @@ import java.util.Map;
public class AlterWorkloadGroupCommand extends AlterCommand {
private final String workloadGroupName;
private final Map<String, String> properties;
+ private String computeGroup;
/**
* constructor
*/
- public AlterWorkloadGroupCommand(String workloadGroupName, Map<String, String> properties) {
+ public AlterWorkloadGroupCommand(String computeGroupName, String workloadGroupName,
+ Map<String, String> properties) {
super(PlanType.ALTER_WORKLOAD_GROUP_COMMAND);
+
this.workloadGroupName = workloadGroupName;
this.properties = properties;
+ this.computeGroup = computeGroupName;
}
@Override
@@ -61,13 +68,30 @@ public class AlterWorkloadGroupCommand extends AlterCommand {
}
String tagStr = properties.get(WorkloadGroup.TAG);
- if (!StringUtils.isEmpty(tagStr) && WorkloadGroupMgr.DEFAULT_GROUP_NAME.equals(workloadGroupName)) {
+ if (!StringUtils.isEmpty(tagStr)) {
throw new AnalysisException(
- WorkloadGroupMgr.DEFAULT_GROUP_NAME
- + " group can not set tag");
+ "tag is deprecated, you can use create workload group [for compute group] as a replacement.");
+ }
+
+ String cgStr = properties.get(WorkloadGroup.COMPUTE_GROUP);
+ if (!StringUtils.isEmpty(cgStr)) {
+ throw new AnalysisException(WorkloadGroup.COMPUTE_GROUP + " can not be set in property.");
+ }
+
+ if (StringUtils.isEmpty(computeGroup)) {
+ computeGroup = Config.isCloudMode() ? Tag.VALUE_DEFAULT_COMPUTE_GROUP_NAME : Tag.DEFAULT_BACKEND_TAG.value;
+ }
+
+ if (Config.isCloudMode()) {
+ String originStr = computeGroup;
+ computeGroup = ((CloudSystemInfoService) Env.getCurrentEnv().getClusterInfo()).getCloudClusterIdByName(
+ computeGroup);
+ if (StringUtils.isEmpty(computeGroup)) {
+ throw new UserException("Can not find compute group " + originStr + ".");
+ }
}
- Env.getCurrentEnv().getWorkloadGroupMgr().alterWorkloadGroup(workloadGroupName, properties);
+ Env.getCurrentEnv().getWorkloadGroupMgr().alterWorkloadGroup(computeGroup, workloadGroupName, properties);
}
@Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateWorkloadGroupCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateWorkloadGroupCommand.java
index 6ef277ee5e4..a0da4e4ce06 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateWorkloadGroupCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateWorkloadGroupCommand.java
@@ -19,15 +19,20 @@ package org.apache.doris.nereids.trees.plans.commands;
import org.apache.doris.analysis.StmtType;
import org.apache.doris.catalog.Env;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.DdlException;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
import org.apache.doris.common.FeNameFormat;
+import org.apache.doris.common.UserException;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.resource.Tag;
import org.apache.doris.resource.workloadgroup.WorkloadGroup;
import org.apache.doris.resource.workloadgroup.WorkloadGroupMgr;
@@ -42,15 +47,19 @@ public class CreateWorkloadGroupCommand extends Command implements ForwardWithSy
private final boolean ifNotExists;
private final String workloadGroupName;
private final Map<String, String> properties;
+ private String computeGroup;
/**
* Constructor for CreateWorkloadGroupCommand
*/
- public CreateWorkloadGroupCommand(String workloadGroupName, boolean ifNotExists, Map<String, String> properties) {
+ public CreateWorkloadGroupCommand(String computeGroup, String workloadGroupName, boolean ifNotExists,
+ Map<String, String> properties) {
super(PlanType.CREATE_WORKLOAD_GROUP_COMMAND);
+
this.workloadGroupName = workloadGroupName;
this.ifNotExists = ifNotExists;
this.properties = properties;
+ this.computeGroup = computeGroup;
}
private void validate(ConnectContext ctx) throws AnalysisException {
@@ -67,21 +76,46 @@ public class CreateWorkloadGroupCommand extends Command implements ForwardWithSy
}
String tagStr = properties.get(WorkloadGroup.TAG);
- if (!StringUtils.isEmpty(tagStr)
- && WorkloadGroupMgr.DEFAULT_GROUP_NAME.equals(workloadGroupName)) {
+ if (!StringUtils.isEmpty(tagStr)) {
throw new AnalysisException(
- WorkloadGroupMgr.DEFAULT_GROUP_NAME
- + " group can not set tag");
+ "tag is deprecated, you can use create workload group [for compute group] as a replacement.");
+ }
+
+ String cgStr = properties.get(WorkloadGroup.COMPUTE_GROUP);
+ if (!StringUtils.isEmpty(cgStr)) {
+ throw new AnalysisException(WorkloadGroup.COMPUTE_GROUP + " can not be set in property.");
}
}
@Override
public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
validate(ctx);
+
+ if (StringUtils.isEmpty(computeGroup)) {
+ computeGroup = Config.isCloudMode() ? Tag.VALUE_DEFAULT_COMPUTE_GROUP_NAME
+ : Tag.DEFAULT_BACKEND_TAG.value;
+ } else if (Config.isNotCloudMode()) {
+ try {
+ FeNameFormat.checkCommonName("Workload group's compute group", computeGroup);
+ } catch (AnalysisException e) {
+ throw new DdlException("Compute group's format is illegal: " + computeGroup);
+ }
+ }
+
+ if (Config.isCloudMode()) {
+ String originStr = computeGroup;
+ computeGroup = ((CloudSystemInfoService) Env.getCurrentEnv().getClusterInfo()).getCloudClusterIdByName(
+ computeGroup);
+ if (StringUtils.isEmpty(computeGroup)) {
+ throw new UserException("Can not find compute group " + originStr + ".");
+ }
+ }
+
// Create workload group
+ properties.put(WorkloadGroup.COMPUTE_GROUP, computeGroup);
WorkloadGroup workloadGroup = WorkloadGroup.create(workloadGroupName, properties);
WorkloadGroupMgr workloadGroupMgr = Env.getCurrentEnv().getWorkloadGroupMgr();
- workloadGroupMgr.createWorkloadGroup(workloadGroup, ifNotExists);
+ workloadGroupMgr.createWorkloadGroup(computeGroup, workloadGroup, ifNotExists);
}
@Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropWorkloadGroupCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropWorkloadGroupCommand.java
index 694fc1411db..c6d47c34348 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropWorkloadGroupCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DropWorkloadGroupCommand.java
@@ -18,13 +18,19 @@
package org.apache.doris.nereids.trees.plans.commands;
import org.apache.doris.catalog.Env;
+import org.apache.doris.cloud.system.CloudSystemInfoService;
+import org.apache.doris.common.Config;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.resource.Tag;
+
+import org.apache.commons.lang3.StringUtils;
/**
* drop workload group command
@@ -32,13 +38,16 @@ import org.apache.doris.qe.StmtExecutor;
public class DropWorkloadGroupCommand extends DropCommand {
private final boolean ifExists;
private final String workloadGroupName;
+ private String computeGroup;
/**
* constructor
*/
- public DropWorkloadGroupCommand(String workloadGroupName, boolean ifExists) {
+ public DropWorkloadGroupCommand(String computeGroupName, String workloadGroupName, boolean ifExists) {
super(PlanType.DROP_WORKLOAD_GROUP_NAME);
+
this.workloadGroupName = workloadGroupName;
+ this.computeGroup = computeGroupName;
this.ifExists = ifExists;
}
@@ -48,7 +57,21 @@ public class DropWorkloadGroupCommand extends DropCommand {
if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
}
- Env.getCurrentEnv().getWorkloadGroupMgr().dropWorkloadGroup(workloadGroupName, ifExists);
+
+ if (StringUtils.isEmpty(computeGroup)) {
+ computeGroup = Config.isCloudMode() ? Tag.VALUE_DEFAULT_COMPUTE_GROUP_NAME : Tag.DEFAULT_BACKEND_TAG.value;
+ }
+
+ if (Config.isCloudMode()) {
+ String originStr = computeGroup;
+ computeGroup = ((CloudSystemInfoService) Env.getCurrentEnv().getClusterInfo()).getCloudClusterIdByName(
+ computeGroup);
+ if (StringUtils.isEmpty(computeGroup)) {
+ throw new UserException("Can not find compute group " + originStr + ".");
+ }
+ }
+
+ Env.getCurrentEnv().getWorkloadGroupMgr().dropWorkloadGroup(computeGroup, workloadGroupName, ifExists);
}
@Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java
index 51fd65acb2e..960fdf68e95 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java
@@ -59,6 +59,8 @@ import com.google.common.base.Strings;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
import org.apache.commons.lang3.StringUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
import java.util.List;
@@ -70,6 +72,7 @@ import java.util.function.Predicate;
* info in creating routine load.
*/
public class CreateRoutineLoadInfo {
+
// routine load properties
public static final String DESIRED_CONCURRENT_NUMBER_PROPERTY = "desired_concurrent_number";
public static final String CURRENT_CONCURRENT_NUMBER_PROPERTY = "current_concurrent_number";
@@ -129,6 +132,8 @@ public class CreateRoutineLoadInfo {
.add(LoadStmt.KEY_ESCAPE)
.build();
+ private static final Logger LOG = LogManager.getLogger(CreateRoutineLoadInfo.class);
+
private final LabelNameInfo labelNameInfo;
private String tableName;
private final Map<String, LoadProperty> loadPropertyMap;
@@ -405,8 +410,17 @@ public class CreateRoutineLoadInfo {
String inputWorkloadGroupStr = jobProperties.get(WORKLOAD_GROUP);
if (!StringUtils.isEmpty(inputWorkloadGroupStr)) {
- this.workloadGroupId = Env.getCurrentEnv().getWorkloadGroupMgr()
- .getWorkloadGroup(ConnectContext.get().getCurrentUserIdentity(), inputWorkloadGroupStr);
+ try {
+ ConnectContext tmpCtx = new ConnectContext();
+ tmpCtx.setCurrentUserIdentity(ConnectContext.get().getCurrentUserIdentity());
+ tmpCtx.getSessionVariable().setWorkloadGroup(inputWorkloadGroupStr);
+ this.workloadGroupId = Env.getCurrentEnv().getWorkloadGroupMgr()
+ .getWorkloadGroup(tmpCtx).get(0)
+ .getId();
+ } catch (Throwable t) {
+ LOG.info("Get workload group failed when create routine load info, ", t);
+ throw t;
+ }
}
if (ConnectContext.get() != null) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
index dc3e5841ef2..bfd87306abe 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
@@ -1421,7 +1421,11 @@ public class ConnectContext {
}
if (Strings.isNullOrEmpty(cluster)) {
- LOG.warn("cant get a valid compute group for user {} to use", getCurrentUserIdentity());
+ List<String> cloudClusterNames
+ = ((CloudSystemInfoService) Env.getCurrentSystemInfo()).getCloudClusterNames();
+ LOG.warn("Can not get a valid compute group for user {} {} to use, all cluster: {}",
+ getCurrentUserIdentity(),
+ getQualifiedUser(), cloudClusterNames);
ComputeGroupException exception = new ComputeGroupException(
"the user is not granted permission to the compute group",
ComputeGroupException.FailedTypeEnum.CURRENT_USER_NO_AUTH_TO_USE_ANY_COMPUTE_GROUP);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
index 8d34183f3cf..bc5c0c2ee32 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
@@ -653,10 +653,15 @@ public class Coordinator implements CoordInterface {
// LoadTask does not have context, not controlled by queue now
if (context != null) {
if (Config.enable_workload_group) {
- this.setTWorkloadGroups(context.getEnv().getWorkloadGroupMgr().getWorkloadGroup(context));
+ List<TPipelineWorkloadGroup> wgList = context.getEnv().getWorkloadGroupMgr().getWorkloadGroup(context);
+ this.setTWorkloadGroups(wgList);
boolean shouldQueue = this.shouldQueue();
if (shouldQueue) {
- queryQueue = context.getEnv().getWorkloadGroupMgr().getWorkloadGroupQueryQueue(context);
+ Set<Long> wgIdSet = Sets.newHashSet();
+ for (TPipelineWorkloadGroup twg : wgList) {
+ wgIdSet.add(twg.getId());
+ }
+ queryQueue = context.getEnv().getWorkloadGroupMgr().getWorkloadGroupQueryQueue(wgIdSet);
if (queryQueue == null) {
// This logic is actually useless, because when could not find query queue, it will
// throw exception during workload group manager.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
index 224e60cf6de..c7714c0f11a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
@@ -67,12 +67,14 @@ import org.apache.doris.thrift.TUniqueId;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/** NereidsCoordinator */
public class NereidsCoordinator extends Coordinator {
@@ -471,9 +473,14 @@ public class NereidsCoordinator extends Coordinator {
// LoadTask does not have context, not controlled by queue now
if (context != null && needEnqueue) {
if (Config.enable_workload_group) {
- coordinatorContext.setWorkloadGroups(context.getEnv().getWorkloadGroupMgr().getWorkloadGroup(context));
+ List<TPipelineWorkloadGroup> wgList = context.getEnv().getWorkloadGroupMgr().getWorkloadGroup(context);
+ this.setTWorkloadGroups(wgList);
if (shouldQueue(context)) {
- QueryQueue queryQueue = context.getEnv().getWorkloadGroupMgr().getWorkloadGroupQueryQueue(context);
+ Set<Long> wgIdSet = Sets.newHashSet();
+ for (TPipelineWorkloadGroup twg : wgList) {
+ wgIdSet.add(twg.getId());
+ }
+ QueryQueue queryQueue = context.getEnv().getWorkloadGroupMgr().getWorkloadGroupQueryQueue(wgIdSet);
if (queryQueue == null) {
// This logic is actually useless, because when could not find query queue, it will
// throw exception during workload group manager.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/AllBackendComputeGroup.java b/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/AllBackendComputeGroup.java
index 665fd9601b0..e1adff87555 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/AllBackendComputeGroup.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/AllBackendComputeGroup.java
@@ -22,6 +22,9 @@ import org.apache.doris.system.Backend;
import org.apache.doris.system.SystemInfoService;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Sets;
+
+import java.util.Set;
public class AllBackendComputeGroup extends ComputeGroup {
@@ -55,6 +58,17 @@ public class AllBackendComputeGroup extends ComputeGroup {
return systemInfoService.getAllClusterBackendsNoException().values().asList();
}
+ @Override
+ public Set<String> getNames() {
+ // in cloud mode, name is cluster id.
+ // in no-cloud mode, name is resource tag's name
+ Set<String> ret = Sets.newHashSet();
+ for (Backend backend : systemInfoService.getAllClusterBackendsNoException().values()) {
+ ret.add(backend.getComputeGroup());
+ }
+ return ret;
+ }
+
@Override
public String toString() {
return AllBackendComputeGroup.class.getSimpleName();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/CloudComputeGroup.java b/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/CloudComputeGroup.java
index 8d2a670d170..2abee7716f2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/CloudComputeGroup.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/CloudComputeGroup.java
@@ -20,7 +20,10 @@ package org.apache.doris.resource.computegroup;
import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.system.Backend;
+import com.google.common.collect.Sets;
+
import java.util.List;
+import java.util.Set;
public class CloudComputeGroup extends ComputeGroup {
@@ -33,4 +36,9 @@ public class CloudComputeGroup extends ComputeGroup {
return ((CloudSystemInfoService) (super.systemInfoService)).getBackendsByClusterName(name);
}
+ @Override
+ public Set<String> getNames() {
+ return Sets.newHashSet(id);
+ }
+
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/ComputeGroup.java b/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/ComputeGroup.java
index 8a9fd7665a8..970c7a2a23c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/ComputeGroup.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/ComputeGroup.java
@@ -20,6 +20,7 @@ package org.apache.doris.resource.computegroup;
import org.apache.doris.system.Backend;
import org.apache.doris.system.SystemInfoService;
+import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -91,6 +92,11 @@ public class ComputeGroup {
return (this == obj);
}
+ // todo(wb) remove getNames, and get workload group from ComputeGroup
+ public Set<String> getNames() {
+ return Sets.newHashSet(name);
+ }
+
private void checkInvalidComputeGroup() {
if (this.getName() == INVALID_COMPUTE_GROUP_NAME) {
throw new RuntimeException(
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 167b6cc5040..ba2613778a0 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
@@ -19,9 +19,12 @@ package org.apache.doris.resource.computegroup;
import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
import org.apache.doris.resource.Tag;
+import org.apache.doris.system.Backend;
import org.apache.doris.system.SystemInfoService;
+import com.aliyuncs.utils.StringUtils;
import com.google.common.collect.Sets;
import java.util.Set;
@@ -34,9 +37,14 @@ public class ComputeGroupMgr {
this.systemInfoService = systemInfoService;
}
- public ComputeGroup getComputeGroupByName(String name) {
+ public ComputeGroup getComputeGroupByName(String name) throws UserException {
if (Config.isCloudMode()) {
- return new CloudComputeGroup("", name, (CloudSystemInfoService) systemInfoService);
+ CloudSystemInfoService cloudSystemInfoService = (CloudSystemInfoService) systemInfoService;
+ String clusterId = cloudSystemInfoService.getCloudClusterIdByName(name);
+ if (StringUtils.isEmpty(clusterId)) {
+ throw new UserException("Can not find compute group:" + name);
+ }
+ return new CloudComputeGroup(clusterId, name, cloudSystemInfoService);
} else {
return new ComputeGroup("", name, systemInfoService);
}
@@ -56,4 +64,12 @@ public class ComputeGroupMgr {
return new AllBackendComputeGroup(systemInfoService);
}
+ public Set<String> getAllComputeGroupIds() {
+ Set<String> ret = Sets.newHashSet();
+ for (Backend backend : systemInfoService.getAllClusterBackendsNoException().values()) {
+ ret.add(backend.getComputeGroup());
+ }
+ return ret;
+ }
+
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/MergedComputeGroup.java b/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/MergedComputeGroup.java
index bebc8a37bdd..6c028c9a2bb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/MergedComputeGroup.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/computegroup/MergedComputeGroup.java
@@ -43,6 +43,11 @@ public class MergedComputeGroup extends ComputeGroup {
return systemInfoService.getBackendListByComputeGroup(computeGroupSet);
}
+ @Override
+ public Set<String> getNames() {
+ return computeGroupSet;
+ }
+
@Override
public String getId() {
throw new RuntimeException("MergedComputeGroup not implements getId.");
@@ -53,11 +58,6 @@ public class MergedComputeGroup extends ComputeGroup {
throw new RuntimeException("MergedComputeGroup not implements getName.");
}
- // current main for UT
- public Set<String> getComputeGroupNameSet() {
- return computeGroupSet;
- }
-
@Override
public String toString() {
return String.format("%s %s", MergedComputeGroup.class.getSimpleName(), String.join(",", computeGroupSet));
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/BindWgToComputeGroupThread.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/BindWgToComputeGroupThread.java
new file mode 100644
index 00000000000..342eb858fec
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/BindWgToComputeGroupThread.java
@@ -0,0 +1,98 @@
+// 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.resource.workloadgroup;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.ha.FrontendNodeType;
+import org.apache.doris.resource.computegroup.ComputeGroupMgr;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.List;
+import java.util.Set;
+
+public class BindWgToComputeGroupThread extends Thread {
+
+ private static final Logger LOG = LogManager.getLogger(BindWgToComputeGroupThread.class);
+
+ public BindWgToComputeGroupThread() {
+ super("BindWgToComputeGroupThread");
+ }
+
+ private void waitCatalogReady() throws InterruptedException {
+ boolean isReady = false;
+ while (!isReady) {
+ FrontendNodeType feType = Env.getCurrentEnv().getFeType();
+ isReady = feType.equals(FrontendNodeType.INIT) || feType.equals(FrontendNodeType.UNKNOWN);
+ if (isReady) {
+ LOG.info("[init_wg]FE is ready");
+ break;
+ } else {
+ LOG.info("[init_wg]FE is not ready, just wait.");
+ Thread.sleep(Config.resource_not_ready_sleep_seconds * 1000);
+ }
+ }
+ Thread.currentThread().join(Config.resource_not_ready_sleep_seconds * 1000L);
+ }
+
+ public void run() {
+ if (!FeConstants.bindWgToComputeGroup) {
+ return;
+ }
+ try {
+ waitCatalogReady();
+ Env.getCurrentEnv().getWorkloadGroupMgr().tryCreateNormalWorkloadGroup();
+ createNewComputeGroup();
+ } catch (Throwable t) {
+ LOG.info("[init_wg]Error happens when drop old workload group, ", t);
+ }
+ }
+
+ private void createNewComputeGroup() throws InterruptedException {
+ WorkloadGroupMgr wgMgr = Env.getCurrentEnv().getWorkloadGroupMgr();
+ LOG.info("[init_wg] print current cg before, id map:{}, name map : {}",
+ wgMgr.getIdToWorkloadGroup(),
+ wgMgr.getNameToWorkloadGroup());
+ List<WorkloadGroup> oldWgList = wgMgr.getOldWorkloadGroup();
+ if (oldWgList.isEmpty()) {
+ LOG.info("[init_wg]There is no old workload group, just return.");
+ return;
+ }
+
+ ComputeGroupMgr cgMgr = Env.getCurrentEnv().getComputeGroupMgr();
+ Set<String> idSet = cgMgr.getAllComputeGroupIds();
+ while (idSet.size() == 0) {
+ LOG.info("[init_wg]Not get any backends, sleep");
+ Thread.sleep(Config.resource_not_ready_sleep_seconds * 1000);
+ idSet = cgMgr.getAllComputeGroupIds();
+ }
+ LOG.info("[init_wg]Get cgs from backend, {}", String.join(",", idSet));
+ for (WorkloadGroup wg : oldWgList) {
+ wgMgr.bindWorkloadGroupToComputeGroup(idSet, wg);
+ }
+ LOG.info(
+ "[init_wg]Finish bing workload group to compute group, wg size: {}, cg size: {}, "
+ + "id map:{}, name map :{}",
+ oldWgList.size(), idSet,
+ wgMgr.getIdToWorkloadGroup(),
+ wgMgr.getNameToWorkloadGroup());
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroup.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroup.java
index 6ca210c86ff..f7a0fdce49c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroup.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroup.java
@@ -18,10 +18,8 @@
package org.apache.doris.resource.workloadgroup;
import org.apache.doris.catalog.Env;
-import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
-import org.apache.doris.common.FeNameFormat;
import org.apache.doris.common.Pair;
import org.apache.doris.common.io.Text;
import org.apache.doris.common.io.Writable;
@@ -49,8 +47,6 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
public class WorkloadGroup implements Writable, GsonPostProcessable {
private static final Logger LOG = LogManager.getLogger(WorkloadGroup.class);
@@ -93,6 +89,8 @@ public class WorkloadGroup implements Writable, GsonPostProcessable {
public static final String SPILL_THRESHOLD_LOW_WATERMARK = "spill_threshold_low_watermark";
public static final String SPILL_THRESHOLD_HIGH_WATERMARK = "spill_threshold_high_watermark";
+ public static final String COMPUTE_GROUP = "compute_group";
+
// NOTE(wb): all property is not required, some properties default value is set in be
// default value is as followed
// cpu_share=1024, memory_limit=0%(0 means not limit), enable_memory_overcommit=true
@@ -101,7 +99,7 @@ public class WorkloadGroup implements Writable, GsonPostProcessable {
.add(MAX_QUEUE_SIZE).add(QUEUE_TIMEOUT).add(CPU_HARD_LIMIT).add(SCAN_THREAD_NUM)
.add(MAX_REMOTE_SCAN_THREAD_NUM).add(MIN_REMOTE_SCAN_THREAD_NUM)
.add(MEMORY_LOW_WATERMARK).add(MEMORY_HIGH_WATERMARK)
- .add(TAG).add(READ_BYTES_PER_SECOND).add(REMOTE_READ_BYTES_PER_SECOND)
+ .add(COMPUTE_GROUP).add(READ_BYTES_PER_SECOND).add(REMOTE_READ_BYTES_PER_SECOND)
.add(WRITE_BUFFER_RATIO).add(SLOT_MEMORY_POLICY).build();
@@ -128,7 +126,7 @@ public class WorkloadGroup implements Writable, GsonPostProcessable {
ALL_PROPERTIES_DEFAULT_VALUE_MAP.put(MIN_REMOTE_SCAN_THREAD_NUM, "-1");
ALL_PROPERTIES_DEFAULT_VALUE_MAP.put(MEMORY_LOW_WATERMARK, "75%");
ALL_PROPERTIES_DEFAULT_VALUE_MAP.put(MEMORY_HIGH_WATERMARK, "85%");
- ALL_PROPERTIES_DEFAULT_VALUE_MAP.put(TAG, "");
+ ALL_PROPERTIES_DEFAULT_VALUE_MAP.put(COMPUTE_GROUP, "");
ALL_PROPERTIES_DEFAULT_VALUE_MAP.put(READ_BYTES_PER_SECOND, "-1");
ALL_PROPERTIES_DEFAULT_VALUE_MAP.put(REMOTE_READ_BYTES_PER_SECOND, "-1");
}
@@ -216,9 +214,6 @@ public class WorkloadGroup implements Writable, GsonPostProcessable {
}
this.properties.put(MEMORY_HIGH_WATERMARK, highWatermarkStr);
}
- if (properties.containsKey(TAG)) {
- this.properties.put(TAG, properties.get(TAG).toLowerCase());
- }
resetQueueProperty(properties);
}
@@ -534,19 +529,6 @@ public class WorkloadGroup implements Writable, GsonPostProcessable {
+ " value should be -1 or an positive integer, but input value is " + readBytesVal);
}
}
-
- String tagStr = properties.get(TAG);
- if (!StringUtils.isEmpty(tagStr)) {
- String[] tagArr = tagStr.split(",");
- for (String tag : tagArr) {
- try {
- FeNameFormat.checkCommonName("workload group tag", tag);
- } catch (AnalysisException e) {
- throw new DdlException("tag format is illegal, " + tagStr);
- }
- }
- }
-
}
public long getId() {
@@ -617,18 +599,9 @@ public class WorkloadGroup implements Writable, GsonPostProcessable {
return memoryLimitPercent;
}
- public Optional<Set<String>> getTag() {
- String tagStr = properties.get(TAG);
- if (StringUtils.isEmpty(tagStr)) {
- return Optional.empty();
- }
-
- Set<String> tagSet = new HashSet<>();
- String[] ss = tagStr.split(",");
- for (String str : ss) {
- tagSet.add(str);
- }
- return Optional.of(tagSet);
+ public String getComputeGroup() {
+ String ret = properties.get(COMPUTE_GROUP);
+ return StringUtils.isEmpty(ret) ? WorkloadGroupMgr.EMPTY_COMPUTE_GROUP : ret;
}
@Override
@@ -729,9 +702,9 @@ public class WorkloadGroup implements Writable, GsonPostProcessable {
tWorkloadGroupInfo.setRemoteReadBytesPerSecond(Long.valueOf(remoteReadBytesPerSecStr));
}
- String tagStr = properties.get(TAG);
- if (!StringUtils.isEmpty(tagStr)) {
- tWorkloadGroupInfo.setTag(tagStr);
+ String cgStr = properties.get(COMPUTE_GROUP);
+ if (!StringUtils.isEmpty(cgStr)) {
+ tWorkloadGroupInfo.setTag(cgStr);
}
String totalQuerySlotCountStr = properties.get(MAX_CONCURRENCY);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java
index 0920abae293..dbe91a2898a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java
@@ -40,6 +40,7 @@ import org.apache.doris.persist.DropWorkloadGroupOperationLog;
import org.apache.doris.persist.gson.GsonPostProcessable;
import org.apache.doris.persist.gson.GsonUtils;
import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.resource.Tag;
import org.apache.doris.thrift.TPipelineWorkloadGroup;
import org.apache.doris.thrift.TUserIdentity;
import org.apache.doris.thrift.TopicInfo;
@@ -57,11 +58,9 @@ import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.Set;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -81,7 +80,7 @@ public class WorkloadGroupMgr extends MasterDaemon implements Writable, GsonPost
.add(WorkloadGroup.SCAN_THREAD_NUM).add(WorkloadGroup.MAX_REMOTE_SCAN_THREAD_NUM)
.add(WorkloadGroup.MIN_REMOTE_SCAN_THREAD_NUM)
.add(WorkloadGroup.MEMORY_LOW_WATERMARK).add(WorkloadGroup.MEMORY_HIGH_WATERMARK)
- .add(WorkloadGroup.TAG)
+ .add(WorkloadGroup.COMPUTE_GROUP)
.add(WorkloadGroup.READ_BYTES_PER_SECOND).add(WorkloadGroup.REMOTE_READ_BYTES_PER_SECOND)
.add(QueryQueue.RUNNING_QUERY_NUM).add(QueryQueue.WAITING_QUERY_NUM)
.build();
@@ -89,11 +88,13 @@ public class WorkloadGroupMgr extends MasterDaemon implements Writable, GsonPost
private static final Logger LOG = LogManager.getLogger(WorkloadGroupMgr.class);
@SerializedName(value = "idToWorkloadGroup")
private final Map<Long, WorkloadGroup> idToWorkloadGroup = Maps.newHashMap();
- private final Map<String, WorkloadGroup> nameToWorkloadGroup = Maps.newHashMap();
+ private final Map<Pair<String, String>, WorkloadGroup> nameToWorkloadGroup = Maps.newHashMap();
private final Map<Long, QueryQueue> idToQueryQueue = Maps.newHashMap();
private final ResourceProcNode procNode = new ResourceProcNode();
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+ public static final String EMPTY_COMPUTE_GROUP = "";
+
@Override
protected void runAfterCatalogReady() {
try {
@@ -138,52 +139,14 @@ public class WorkloadGroupMgr extends MasterDaemon implements Writable, GsonPost
public WorkloadGroupMgr() {
super("workload-group-thread", Config.query_queue_update_interval_ms);
- // if no fe image exist, we should append internal group here.
- appendInternalWorkloadGroup();
}
public static WorkloadGroupMgr read(DataInput in) throws IOException {
String json = Text.readString(in);
WorkloadGroupMgr ret = GsonUtils.GSON.fromJson(json, WorkloadGroupMgr.class);
- ret.appendInternalWorkloadGroup();
return ret;
}
- public void appendInternalWorkloadGroup() {
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.CPU_SHARE, "1024");
- properties.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- properties.put(WorkloadGroup.ENABLE_MEMORY_OVERCOMMIT, "true");
- WorkloadGroup defaultValWg = new WorkloadGroup(DEFAULT_GROUP_ID.longValue(), DEFAULT_GROUP_NAME,
- properties);
-
- // when doris version is 2.0, user create a normal group with id 12345
- // when doris upgrade from 2.0 to 2.1.2, Doris may create a workload id with 1
- // then doris could contain two normal workload group with id 12345 and 1
- // so we should check duplicate workload group when Fe starts
- // and remove invalid workload group.
- // case 1: no images exist or has an image but has no normal wg,
- // insert a normal group with id 1 and default value directly.
- // case 2: image exits and has a normal group, then do nothing.
- Set<Long> invalidNormalWg = new HashSet<>();
- for (WorkloadGroup curWg : idToWorkloadGroup.values()) {
- if (DEFAULT_GROUP_NAME.equals(curWg.getName()) && DEFAULT_GROUP_ID.longValue() != curWg.getId()) {
- invalidNormalWg.add(curWg.getId());
- }
- }
- for (Long wgId : invalidNormalWg) {
- idToWorkloadGroup.remove(wgId);
- }
-
- WorkloadGroup curNormalWg = idToWorkloadGroup.get(DEFAULT_GROUP_ID);
- if (curNormalWg == null) {
- curNormalWg = defaultValWg;
- idToWorkloadGroup.put(curNormalWg.getId(), curNormalWg);
- }
- nameToWorkloadGroup.put(curNormalWg.getName(), curNormalWg);
-
- }
-
private void readLock() {
lock.readLock().lock();
}
@@ -200,37 +163,34 @@ public class WorkloadGroupMgr extends MasterDaemon implements Writable, GsonPost
lock.writeLock().unlock();
}
+ private WorkloadGroup getWorkloadGroupByComputeGroupUnlock(String cgName, String wgName)
+ throws DdlException {
+ WorkloadGroup wg = nameToWorkloadGroup.get(Pair.of(cgName, wgName));
+ if (wg == null) {
+ throw new DdlException(
+ "Can not find workload group " + wgName + " in compute group " + cgName + ".");
+ }
+ return wg;
+ }
+
public List<TPipelineWorkloadGroup> getWorkloadGroup(ConnectContext context) throws UserException {
- String groupName = getWorkloadGroupNameAndCheckPriv(context);
+ String wgName = getWorkloadGroupNameAndCheckPriv(context);
+ Set<String> cgNames = context.getComputeGroup().getNames();
+
List<TPipelineWorkloadGroup> workloadGroups = Lists.newArrayList();
readLock();
try {
- WorkloadGroup workloadGroup = nameToWorkloadGroup.get(groupName);
- if (workloadGroup == null) {
- throw new UserException("Workload group " + groupName + " does not exist");
+ for (String cgName : cgNames) {
+ WorkloadGroup workloadGroup = getWorkloadGroupByComputeGroupUnlock(cgName, wgName);
+ workloadGroups.add(workloadGroup.toThrift());
}
- workloadGroups.add(workloadGroup.toThrift());
- context.setWorkloadGroupName(groupName);
+ context.setWorkloadGroupName(wgName);
} finally {
readUnlock();
}
return workloadGroups;
}
- public long getWorkloadGroup(UserIdentity currentUser, String groupName) throws UserException {
- Long workloadId = getWorkloadGroupIdByName(groupName);
- if (workloadId == null) {
- throw new UserException("Workload group " + groupName + " does not exist");
- }
- if (!Env.getCurrentEnv().getAccessManager()
- .checkWorkloadGroupPriv(currentUser, groupName, PrivPredicate.USAGE)) {
- ErrorReport.reportAnalysisException(
- "Access denied; you need (at least one of) the %s privilege(s) to use workload group '%s'.",
- ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "USAGE/ADMIN", groupName);
- }
- return workloadId.longValue();
- }
-
public List<TPipelineWorkloadGroup> getTWorkloadGroupById(long wgId) {
List<TPipelineWorkloadGroup> tWorkloadGroups = Lists.newArrayList();
readLock();
@@ -245,39 +205,6 @@ public class WorkloadGroupMgr extends MasterDaemon implements Writable, GsonPost
return tWorkloadGroups;
}
- public List<TPipelineWorkloadGroup> getWorkloadGroupByUser(UserIdentity user, boolean checkAuth)
- throws UserException {
- String groupName = Env.getCurrentEnv().getAuth().getWorkloadGroup(user.getQualifiedUser());
- List<TPipelineWorkloadGroup> ret = new ArrayList<>();
- WorkloadGroup wg = null;
- readLock();
- try {
- if (groupName == null || groupName.isEmpty()) {
- wg = nameToWorkloadGroup.get(DEFAULT_GROUP_NAME);
- if (wg == null) {
- throw new RuntimeException("can not find normal workload group for user " + user);
- }
- } else {
- wg = nameToWorkloadGroup.get(groupName);
- if (wg == null) {
- throw new UserException(
- "can not find workload group " + groupName + " for user " + user);
- }
- }
- if (checkAuth && !Env.getCurrentEnv().getAccessManager()
- .checkWorkloadGroupPriv(user, wg.getName(), PrivPredicate.USAGE)) {
- ErrorReport.reportAnalysisException(
- "Access denied; you need (at least one of) the %s privilege(s) to use workload group '%s'."
- + " used id=(%s)",
- ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "USAGE/ADMIN", wg.getName(), user.toString());
- }
- ret.add(wg.toThrift());
- } finally {
- readUnlock();
- }
- return ret;
- }
-
public List<TopicInfo> getPublishTopicInfo() {
List<TopicInfo> workloadGroups = new ArrayList();
readLock();
@@ -291,19 +218,35 @@ public class WorkloadGroupMgr extends MasterDaemon implements Writable, GsonPost
return workloadGroups;
}
- public QueryQueue getWorkloadGroupQueryQueue(ConnectContext context) throws UserException {
- String groupName = getWorkloadGroupNameAndCheckPriv(context);
+ public QueryQueue getWorkloadGroupQueryQueue(Set<Long> wgIdSet) throws UserException {
writeLock();
try {
- WorkloadGroup wg = nameToWorkloadGroup.get(groupName);
- if (wg == null) {
- throw new UserException("Workload group " + groupName + " does not exist");
+ QueryQueue queryQueue = null;
+ for (long wgId : wgIdSet) {
+ WorkloadGroup wg = idToWorkloadGroup.get(wgId);
+ if (wg == null) {
+ continue;
+ }
+ QueryQueue tmpQueue = idToQueryQueue.get(wg.getId());
+ if (tmpQueue == null) {
+ tmpQueue = new QueryQueue(wg.getId(), wg.getMaxConcurrency(), wg.getMaxQueueSize(),
+ wg.getQueueTimeout(), wg.getVersion());
+ idToQueryQueue.put(wg.getId(), tmpQueue);
+ queryQueue = tmpQueue;
+ break;
+ }
+ if (queryQueue == null) {
+ queryQueue = tmpQueue;
+ } else {
+ Pair<Integer, Integer> detail1 = queryQueue.getQueryQueueDetail();
+ Pair<Integer, Integer> detail2 = tmpQueue.getQueryQueueDetail();
+ if (detail2.first < detail1.first) {
+ queryQueue = tmpQueue;
+ }
+ }
}
- QueryQueue queryQueue = idToQueryQueue.get(wg.getId());
if (queryQueue == null) {
- queryQueue = new QueryQueue(wg.getId(), wg.getMaxConcurrency(), wg.getMaxQueueSize(),
- wg.getQueueTimeout(), wg.getVersion());
- idToQueryQueue.put(wg.getId(), queryQueue);
+ throw new DdlException("Can not find query queue for workload group: " + wgIdSet);
}
return queryQueue;
} finally {
@@ -352,190 +295,184 @@ public class WorkloadGroupMgr extends MasterDaemon implements Writable, GsonPost
return groupName;
}
- public void createWorkloadGroup(WorkloadGroup workloadGroup, boolean isIfNotExists) throws DdlException {
+ public void createWorkloadGroup(String computeGroup, WorkloadGroup workloadGroup, boolean isIfNotExists)
+ throws DdlException {
String workloadGroupName = workloadGroup.getName();
writeLock();
try {
- if (nameToWorkloadGroup.containsKey(workloadGroupName)) {
+ Pair<String, String> key = Pair.of(computeGroup, workloadGroupName);
+ if (nameToWorkloadGroup.containsKey(key)) {
if (isIfNotExists) {
return;
}
- throw new DdlException("workload group " + workloadGroupName + " already exist");
+ throw new DdlException(
+ "Compute group " + key.first + " already has workload group " + key.second + ".");
}
if (idToWorkloadGroup.size() >= Config.workload_group_max_num) {
throw new DdlException(
- "workload group number can not be exceed " + Config.workload_group_max_num);
+ "Workload group number can not be exceed " + Config.workload_group_max_num);
}
checkGlobalUnlock(workloadGroup, null);
- nameToWorkloadGroup.put(workloadGroupName, workloadGroup);
+ nameToWorkloadGroup.put(key, workloadGroup);
idToWorkloadGroup.put(workloadGroup.getId(), workloadGroup);
Env.getCurrentEnv().getEditLog().logCreateWorkloadGroup(workloadGroup);
} finally {
writeUnlock();
}
- LOG.info("Create workload group success: {}", workloadGroup);
+ LOG.info("Create workload group {} for compute group {} success.", workloadGroup, computeGroup);
}
public void createWorkloadGroup(CreateWorkloadGroupStmt stmt) throws DdlException {
- WorkloadGroup workloadGroup = WorkloadGroup.create(stmt.getWorkloadGroupName(), stmt.getProperties());
- createWorkloadGroup(workloadGroup, stmt.isIfNotExists());
+ throw new DdlException("Unsupported create statement");
}
// NOTE: used for checking sum value of 100% for cpu_hard_limit and memory_limit
// when create/alter workload group with same tag.
// when oldWg is null it means caller is an alter stmt.
private void checkGlobalUnlock(WorkloadGroup newWg, WorkloadGroup oldWg) throws DdlException {
- Optional<Set<String>> newWgTag = newWg.getTag();
- Set<String> newWgTagSet = null;
- if (newWgTag.isPresent()) {
- newWgTagSet = newWgTag.get();
- } else {
- newWgTagSet = new HashSet<>();
- newWgTagSet.add(null);
- }
+ String newWgCg = newWg.getComputeGroup();
- for (String newWgOneTag : newWgTagSet) {
- double sumOfAllMemLimit = 0;
- int sumOfAllCpuHardLimit = 0;
+ double sumOfAllMemLimit = 0;
+ int sumOfAllCpuHardLimit = 0;
- // 1 get sum value of all wg which has same tag without current wg
- for (Map.Entry<Long, WorkloadGroup> entry : idToWorkloadGroup.entrySet()) {
- WorkloadGroup wg = entry.getValue();
- Optional<Set<String>> wgTag = wg.getTag();
+ // 1 get sum value of all wg which has same tag without current wg
+ for (Map.Entry<Long, WorkloadGroup> entry : idToWorkloadGroup.entrySet()) {
+ WorkloadGroup wg = entry.getValue();
+ String curWgCg = wg.getComputeGroup();
- if (oldWg != null && entry.getKey() == oldWg.getId()) {
- continue;
- }
+ if (oldWg != null && entry.getKey() == oldWg.getId()) {
+ continue;
+ }
- if (newWgOneTag == null) {
- if (wgTag.isPresent()) {
- continue;
- }
- } else if (!wgTag.isPresent() || (!wgTag.get().contains(newWgOneTag))) {
- continue;
- }
+ if (!newWgCg.equals(curWgCg)) {
+ continue;
+ }
- if (wg.getCpuHardLimitWhenCalSum() > 0) {
- sumOfAllCpuHardLimit += wg.getCpuHardLimitWhenCalSum();
- }
- if (wg.getMemoryLimitPercentWhenCalSum() > 0) {
- sumOfAllMemLimit += wg.getMemoryLimitPercentWhenCalSum();
- }
+ if (wg.getCpuHardLimitWhenCalSum() > 0) {
+ sumOfAllCpuHardLimit += wg.getCpuHardLimitWhenCalSum();
+ }
+ if (wg.getMemoryLimitPercentWhenCalSum() > 0) {
+ sumOfAllMemLimit += wg.getMemoryLimitPercentWhenCalSum();
}
+ }
- // 2 sum current wg value
- sumOfAllMemLimit += newWg.getMemoryLimitPercentWhenCalSum();
- sumOfAllCpuHardLimit += newWg.getCpuHardLimitWhenCalSum();
+ // 2 sum current wg value
+ sumOfAllMemLimit += newWg.getMemoryLimitPercentWhenCalSum();
+ sumOfAllCpuHardLimit += newWg.getCpuHardLimitWhenCalSum();
- // 3 check total sum
- if (sumOfAllMemLimit > 100.0 + 1e-6) {
- throw new DdlException(
- "The sum of all workload group " + WorkloadGroup.MEMORY_LIMIT + " within tag " + (
- newWgTag.isPresent() ? newWgTag.get() : "")
- + " cannot be greater than 100.0%. current sum val:" + sumOfAllMemLimit);
- }
+ // 3 check total sum
+ if (sumOfAllMemLimit > 100.0 + 1e-6) {
+ throw new DdlException(
+ "The sum of all workload group " + WorkloadGroup.MEMORY_LIMIT + " within compute group " + (
+ newWgCg)
+ + " can not be greater than 100.0%. current sum val:" + sumOfAllMemLimit);
+ }
- if (sumOfAllCpuHardLimit > 100) {
- throw new DdlException(
- "sum of all workload group " + WorkloadGroup.CPU_HARD_LIMIT + " within tag " + (
- newWgTag.isPresent()
- ? newWgTag.get() : "") + " can not be greater than 100% ");
- }
+ if (sumOfAllCpuHardLimit > 100) {
+ throw new DdlException(
+ "The sum of all workload group " + WorkloadGroup.CPU_HARD_LIMIT + " within compute group " + newWgCg
+ + " can not be greater than 100%. current sum val:" + sumOfAllCpuHardLimit);
}
}
public void alterWorkloadGroup(AlterWorkloadGroupStmt stmt) throws DdlException {
- alterWorkloadGroup(stmt.getWorkloadGroupName(), stmt.getProperties());
+ throw new DdlException("Unsupported alter statement");
}
- public void alterWorkloadGroup(String workloadGroupName, Map<String, String> properties) throws DdlException {
+ public void alterWorkloadGroup(String computeGroup, String workloadGroupName, Map<String, String> properties)
+ throws DdlException {
if (properties.size() == 0) {
- throw new DdlException("alter workload group should contain at least one property");
+ throw new DdlException("Alter workload group should contain at least one property");
}
+
WorkloadGroup newWorkloadGroup;
writeLock();
try {
- if (!nameToWorkloadGroup.containsKey(workloadGroupName)) {
- throw new DdlException("workload group(" + workloadGroupName + ") does not exist.");
- }
- WorkloadGroup currentWorkloadGroup = nameToWorkloadGroup.get(workloadGroupName);
+ WorkloadGroup currentWorkloadGroup = getWorkloadGroupByComputeGroupUnlock(computeGroup, workloadGroupName);
newWorkloadGroup = WorkloadGroup.copyAndUpdate(currentWorkloadGroup, properties);
checkGlobalUnlock(newWorkloadGroup, currentWorkloadGroup);
- nameToWorkloadGroup.put(workloadGroupName, newWorkloadGroup);
+ nameToWorkloadGroup.put(Pair.of(computeGroup, workloadGroupName), newWorkloadGroup);
idToWorkloadGroup.put(newWorkloadGroup.getId(), newWorkloadGroup);
- // NOTE: used for regression test query queue
- if (Config.enable_alter_queue_prop_sync) {
- resetQueryQueueProp();
- }
Env.getCurrentEnv().getEditLog().logAlterWorkloadGroup(newWorkloadGroup);
} finally {
writeUnlock();
}
- LOG.info("Alter resource success: {}", newWorkloadGroup);
+ LOG.info("Alter workload group {} for compute group {} success: {}", newWorkloadGroup, computeGroup);
}
public void dropWorkloadGroup(DropWorkloadGroupStmt stmt) throws DdlException {
- dropWorkloadGroup(stmt.getWorkloadGroupName(), stmt.isIfExists());
+ throw new DdlException("Unsupported drop statement.");
}
- public void dropWorkloadGroup(String workloadGroupName, boolean ifExists) throws DdlException {
- if (DEFAULT_GROUP_NAME.equals(workloadGroupName)) {
- throw new DdlException("Dropping workload group " + workloadGroupName + " is not allowed");
+ public void dropWorkloadGroup(String computeGroup, String workloadGroupName, boolean ifExists)
+ throws DdlException {
+ if (Tag.DEFAULT_BACKEND_TAG.value.equals(computeGroup) && DEFAULT_GROUP_NAME.equals(workloadGroupName)) {
+ throw new DdlException(
+ "Dropping workload group " + workloadGroupName + " for " + computeGroup + "is not allowed.");
}
// if a workload group exists in user property, it should not be dropped
// user need to reset user property first
Pair<Boolean, String> ret = Env.getCurrentEnv().getAuth().isWorkloadGroupInUse(workloadGroupName);
if (ret.first) {
- throw new DdlException("workload group " + workloadGroupName + " is set for user " + ret.second
+ throw new DdlException("Workload group " + workloadGroupName + " is set for user " + ret.second
+ ", you can reset the user's property(eg, "
+ "set property for " + ret.second + " 'default_workload_group'='xxx'; ), "
+ "then you can drop the group.");
}
// A group with related policies should not be deleted.
- Long wgId = getWorkloadGroupIdByName(workloadGroupName);
+ Long wgId = null;
+ readLock();
+ try {
+ WorkloadGroup wg = nameToWorkloadGroup.get(Pair.of(computeGroup, workloadGroupName));
+ if (wg != null) {
+ wgId = wg.getId();
+ }
+ } finally {
+ readUnlock();
+ }
if (wgId != null) {
boolean groupHasPolicy = Env.getCurrentEnv().getWorkloadSchedPolicyMgr()
.checkWhetherGroupHasPolicy(wgId.longValue());
if (groupHasPolicy) {
throw new DdlException(
- "workload group " + workloadGroupName + " can't be dropped, because it has related policy");
+ "Workload group " + workloadGroupName + " can't be dropped, because it has related policy");
}
}
+ Pair<String, String> key = Pair.of(computeGroup, workloadGroupName);
writeLock();
try {
- if (!nameToWorkloadGroup.containsKey(workloadGroupName)) {
+ if (!nameToWorkloadGroup.containsKey(key)) {
if (ifExists) {
return;
}
- throw new DdlException("workload group " + workloadGroupName + " does not exist");
+ throw new DdlException(
+ "Can not find workload group " + workloadGroupName + " in compute group " + computeGroup + ".");
}
- WorkloadGroup workloadGroup = nameToWorkloadGroup.get(workloadGroupName);
+ WorkloadGroup workloadGroup = nameToWorkloadGroup.get(key);
long groupId = workloadGroup.getId();
idToWorkloadGroup.remove(groupId);
- nameToWorkloadGroup.remove(workloadGroupName);
+ nameToWorkloadGroup.remove(key);
idToQueryQueue.remove(groupId);
Env.getCurrentEnv().getEditLog().logDropWorkloadGroup(new DropWorkloadGroupOperationLog(groupId));
} finally {
writeUnlock();
}
- LOG.info("Drop workload group success: {}", workloadGroupName);
+ LOG.info("Drop workload group success: {} for compute group {}", workloadGroupName, computeGroup);
}
private void insertWorkloadGroup(WorkloadGroup workloadGroup) {
writeLock();
try {
- // when wg named normal but id is not DEFAULT_GROUP_ID,
- // then we should abort it to avoid duplicate normal group
- if (DEFAULT_GROUP_NAME.equals(workloadGroup.getName())
- && DEFAULT_GROUP_ID.longValue() != workloadGroup.getId()) {
- return;
- }
-
+ LOG.info("[init_wg] before, name map, {}, id map: {}, name map: {}", "replay",
+ idToWorkloadGroup, nameToWorkloadGroup);
idToWorkloadGroup.put(workloadGroup.getId(), workloadGroup);
- nameToWorkloadGroup.put(workloadGroup.getName(), workloadGroup);
+ Pair<String, String> key = Pair.of(workloadGroup.getComputeGroup(), workloadGroup.getName());
+ nameToWorkloadGroup.put(key, workloadGroup);
+ LOG.info("[init_wg] after, name map, {}, id map: {}, name map: {}", "replay",
+ idToWorkloadGroup, nameToWorkloadGroup);
} finally {
writeUnlock();
}
@@ -590,19 +527,6 @@ public class WorkloadGroupMgr extends MasterDaemon implements Writable, GsonPost
return procNode.fetchResult(currentUserIdentity).getRows();
}
- public Long getWorkloadGroupIdByName(String name) {
- readLock();
- try {
- WorkloadGroup wg = nameToWorkloadGroup.get(name);
- if (wg == null) {
- return null;
- }
- return wg.getId();
- } finally {
- readUnlock();
- }
- }
-
public Map<Long, String> getIdToNameMap() {
Map<Long, String> ret = Maps.newHashMap();
readLock();
@@ -630,7 +554,7 @@ public class WorkloadGroupMgr extends MasterDaemon implements Writable, GsonPost
}
// for ut
- public Map<String, WorkloadGroup> getNameToWorkloadGroup() {
+ public Map<Pair<String, String>, WorkloadGroup> getNameToWorkloadGroup() {
return nameToWorkloadGroup;
}
@@ -639,6 +563,11 @@ public class WorkloadGroupMgr extends MasterDaemon implements Writable, GsonPost
return idToWorkloadGroup;
}
+ // for ut
+ public Map<Long, QueryQueue> getIdToQueryQueue() {
+ return idToQueryQueue;
+ }
+
@Override
public void write(DataOutput out) throws IOException {
String json = GsonUtils.GSON.toJson(this);
@@ -647,8 +576,14 @@ public class WorkloadGroupMgr extends MasterDaemon implements Writable, GsonPost
@Override
public void gsonPostProcess() throws IOException {
- idToWorkloadGroup.forEach(
- (id, workloadGroup) -> nameToWorkloadGroup.put(workloadGroup.getName(), workloadGroup));
+ LOG.info("[init_wg] before, name map, {}, id map: {}, name map: {}", "gson",
+ idToWorkloadGroup, nameToWorkloadGroup);
+ for (Map.Entry<Long, WorkloadGroup> entry : idToWorkloadGroup.entrySet()) {
+ String computeGroupName = entry.getValue().getComputeGroup();
+ nameToWorkloadGroup.put(Pair.of(computeGroupName, entry.getValue().getName()), entry.getValue());
+ }
+ LOG.info("[init_wg] after, name map, {}, id map: {}, name map: {}", "gson",
+ idToWorkloadGroup, nameToWorkloadGroup);
}
public class ResourceProcNode {
@@ -670,4 +605,92 @@ public class WorkloadGroupMgr extends MasterDaemon implements Writable, GsonPost
return result;
}
}
+
+
+ public List<WorkloadGroup> getOldWorkloadGroup() {
+ List<WorkloadGroup> oldWgList = Lists.newArrayList();
+ readLock();
+ try {
+ for (Map.Entry<Pair<String, String>, WorkloadGroup> entry : nameToWorkloadGroup.entrySet()) {
+ if (EMPTY_COMPUTE_GROUP.equals(entry.getKey().first)) {
+ oldWgList.add(entry.getValue());
+ }
+ }
+ } finally {
+ readUnlock();
+ }
+ return oldWgList;
+ }
+
+ public void tryCreateNormalWorkloadGroup() {
+ writeLock();
+ try {
+ LOG.info("[init_wg] before create normal wg, id map: {}, name map: {}", idToWorkloadGroup,
+ nameToWorkloadGroup);
+ if (idToWorkloadGroup.isEmpty()) {
+ String defaultCgName = Config.isCloudMode() ? Tag.VALUE_DEFAULT_COMPUTE_GROUP_NAME
+ : Tag.VALUE_DEFAULT_TAG;
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(WorkloadGroup.ENABLE_MEMORY_OVERCOMMIT, "true");
+ properties.put(WorkloadGroup.COMPUTE_GROUP, defaultCgName);
+ WorkloadGroup defaultWg = new WorkloadGroup(Env.getCurrentEnv().getNextId(), DEFAULT_GROUP_NAME,
+ properties);
+ idToWorkloadGroup.put(defaultWg.getId(), defaultWg);
+ nameToWorkloadGroup.put(Pair.of(defaultWg.getComputeGroup(), defaultWg.getName()), defaultWg);
+ Env.getCurrentEnv().getEditLog().logCreateWorkloadGroup(defaultWg);
+ LOG.info("[init_wg]Create default workload group success: {}", defaultWg);
+ } else {
+ LOG.info("[init_wg]This is not a new cluster, skip create default wg");
+ }
+ LOG.info("[init_wg] after create normal wg, id map: {}, name map: {}", idToWorkloadGroup,
+ nameToWorkloadGroup);
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ public void bindWorkloadGroupToComputeGroup(Set<String> cgSet, WorkloadGroup oldWg) {
+ writeLock();
+ try {
+ Pair<String, String> oldKey = Pair.of(EMPTY_COMPUTE_GROUP, oldWg.getName());
+ // it means old compute group has been dropped, just return;
+ if (!nameToWorkloadGroup.containsKey(oldKey)) {
+ LOG.info("[init_wg]Old workload group {} has been dropped, skip it.", oldWg.getName());
+ return;
+ }
+ // create new workload group for all compute group.
+ for (String computeGroup : cgSet) {
+ Pair<String, String> newKey = Pair.of(computeGroup, oldWg.getName());
+ if (nameToWorkloadGroup.containsKey(newKey)) {
+ LOG.info("[init_wg]Workload group {} already exists in compute group {}, skip it.",
+ oldWg.getName(), computeGroup);
+ continue;
+ }
+ Map<String, String> newProp = Maps.newHashMap();
+ for (Map.Entry<String, String> entry : oldWg.getProperties().entrySet()) {
+ newProp.put(entry.getKey(), entry.getValue());
+ }
+ newProp.put(WorkloadGroup.COMPUTE_GROUP, computeGroup);
+ WorkloadGroup newWg = new WorkloadGroup(Env.getCurrentEnv().getNextId(), oldWg.getName(),
+ newProp);
+ nameToWorkloadGroup.put(newKey, newWg);
+ idToWorkloadGroup.put(newWg.getId(), newWg);
+ Env.getCurrentEnv().getEditLog().logCreateWorkloadGroup(newWg);
+ LOG.info("[init_wg]Create workload group {} for compute group {} success.", oldWg.getName(),
+ computeGroup);
+ }
+
+ // drop old workload group
+ nameToWorkloadGroup.remove(oldKey);
+ idToWorkloadGroup.remove(oldWg.getId());
+ idToQueryQueue.remove(oldWg.getId());
+ Env.getCurrentEnv().getEditLog().logDropWorkloadGroup(new DropWorkloadGroupOperationLog(oldWg.getId()));
+ LOG.info("[init_wg]Drop old workload group {} success.", oldWg);
+ } catch (Throwable t) {
+ LOG.error("[init_wg]Error happens when drop old workload group, {}, {}", cgSet, oldWg.getName(), t);
+ } finally {
+ writeUnlock();
+ }
+ }
+
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java
index 43f9cd2a794..c98af8b87f2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java
@@ -34,6 +34,7 @@ import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.persist.gson.GsonPostProcessable;
import org.apache.doris.persist.gson.GsonUtils;
import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.resource.Tag;
import org.apache.doris.service.ExecuteEnv;
import org.apache.doris.thrift.TCompareOperator;
import org.apache.doris.thrift.TUserIdentity;
@@ -447,10 +448,24 @@ public class WorkloadSchedPolicyMgr extends MasterDaemon implements Writable, Gs
String workloadGroupNameStr = properties.get(WorkloadSchedPolicy.WORKLOAD_GROUP);
if (workloadGroupNameStr != null && !workloadGroupNameStr.isEmpty()) {
- Long wgId = Env.getCurrentEnv().getWorkloadGroupMgr().getWorkloadGroupIdByName(workloadGroupNameStr);
- if (wgId == null) {
- throw new UserException("unknown workload group:" + workloadGroupNameStr);
+ String cg = Config.isCloudMode() ? Tag.VALUE_DEFAULT_COMPUTE_GROUP_NAME : Tag.VALUE_DEFAULT_TAG;
+ String wg = "";
+ String[] ss = workloadGroupNameStr.split("\\.");
+ if (ss.length == 1) {
+ wg = ss[0];
+ } else if (ss.length == 2) {
+ cg = ss[0];
+ wg = ss[1];
+ } else {
+ throw new UserException("invalid workload group format: " + workloadGroupNameStr);
}
+ ConnectContext tmpCtx = new ConnectContext();
+ tmpCtx.setComputeGroup(
+ Env.getCurrentEnv().getComputeGroupMgr().getComputeGroupByName(cg));
+ tmpCtx.getSessionVariable().setWorkloadGroup(wg);
+ tmpCtx.setCurrentUserIdentity(UserIdentity.ROOT);
+ Long wgId = Env.getCurrentEnv().getWorkloadGroupMgr().getWorkloadGroup(tmpCtx)
+ .get(0).getId();
wgIdList.add(wgId);
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index 0d1afbac39e..91d33a42bb1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -2034,8 +2034,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
boolean hasUserName = !StringUtils.isEmpty(request.getUser());
if (Config.enable_workload_group && hasUserName) {
tWorkloadGroupList = Env.getCurrentEnv().getWorkloadGroupMgr()
- .getWorkloadGroupByUser(ConnectContext.get()
- .getCurrentUserIdentity(), true);
+ .getWorkloadGroup(ConnectContext.get());
}
if (!Strings.isNullOrEmpty(request.getLoadSql())) {
httpStreamPutImpl(request, result);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java b/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java
index 0bd8377bddb..f0decf481d8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java
@@ -39,9 +39,7 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
import com.google.gson.annotations.SerializedName;
-import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -50,12 +48,9 @@ import java.io.DataOutput;
import java.io.IOException;
import java.security.SecureRandom;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
/**
@@ -1050,36 +1045,12 @@ public class Backend implements Writable {
this.lastPublishTaskAccumulatedNum = accumulatedNum;
}
- public Set<String> getBeWorkloadGroupTagSet() {
- Set<String> beTagSet = Sets.newHashSet();
- String beTagStr = this.tagMap.get(Tag.WORKLOAD_GROUP);
- if (StringUtils.isEmpty(beTagStr)) {
- return beTagSet;
- }
-
- String[] beTagArr = beTagStr.split(",");
- for (String beTag : beTagArr) {
- beTagSet.add(beTag.trim());
- }
-
- return beTagSet;
- }
-
- public static boolean isMatchWorkloadGroupTag(String wgTagStr, Set<String> beTagSet) {
- if (StringUtils.isEmpty(wgTagStr)) {
- return true;
- }
- if (beTagSet.isEmpty()) {
- return false;
- }
-
- String[] wgTagArr = wgTagStr.split(",");
- Set<String> wgTagSet = new HashSet<>();
- for (String wgTag : wgTagArr) {
- wgTagSet.add(wgTag.trim());
+ public String getComputeGroup() {
+ if (Config.isCloudMode()) {
+ return getCloudClusterId();
+ } else {
+ return getLocationTag().value;
}
-
- return !Collections.disjoint(wgTagSet, beTagSet);
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
index 4e7cd6b5116..4c94f8d1fe7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
@@ -625,7 +625,7 @@ public class MetadataGenerator {
trow.addToColumnValue(new TCell().setLongVal(Long.valueOf(rGroupsInfo.get(13))));
trow.addToColumnValue(new TCell().setStringVal(rGroupsInfo.get(14))); // spill low watermark
trow.addToColumnValue(new TCell().setStringVal(rGroupsInfo.get(15))); // spill high watermark
- trow.addToColumnValue(new TCell().setStringVal(rGroupsInfo.get(16))); // tag
+ trow.addToColumnValue(new TCell().setStringVal(rGroupsInfo.get(16))); // compute group
trow.addToColumnValue(new TCell().setLongVal(Long.valueOf(rGroupsInfo.get(17)))); // read bytes per second
trow.addToColumnValue(
new TCell().setLongVal(Long.valueOf(rGroupsInfo.get(18)))); // remote read bytes per second
diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateEncryptKeyTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateEncryptKeyTest.java
index a1e4a299a65..c062a52d17d 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateEncryptKeyTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/CreateEncryptKeyTest.java
@@ -58,6 +58,7 @@ public class CreateEncryptKeyTest {
@Test
public void test() throws Exception {
ConnectContext ctx = UtFrameUtils.createDefaultCtx();
+ Env.getCurrentEnv().getWorkloadGroupMgr().tryCreateNormalWorkloadGroup();
// create database db1
String createDbStmtStr = "create database db1;";
diff --git a/fe/fe-core/src/test/java/org/apache/doris/planner/ResourceTagQueryTest.java b/fe/fe-core/src/test/java/org/apache/doris/planner/ResourceTagQueryTest.java
index be08780ee91..711b5c9b150 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/planner/ResourceTagQueryTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/planner/ResourceTagQueryTest.java
@@ -207,7 +207,7 @@ public class ResourceTagQueryTest {
ComputeGroup cg2 = Env.getCurrentEnv().getAuth().getComputeGroup(Auth.ROOT_USER);
Assert.assertTrue(cg2 instanceof MergedComputeGroup);
MergedComputeGroup cg3 = (MergedComputeGroup) cg2;
- Set<String> cgNameSet1 = cg3.getComputeGroupNameSet();
+ Set<String> cgNameSet1 = cg3.getNames();
Assert.assertTrue(cgNameSet1.size() == 1);
Assert.assertTrue(cgNameSet1.contains("default"));
@@ -224,8 +224,8 @@ public class ResourceTagQueryTest {
ComputeGroup cg4 = Env.getCurrentEnv().getAuth().getComputeGroup(Auth.ROOT_USER);
Assert.assertTrue(cg4 instanceof MergedComputeGroup);
MergedComputeGroup cg5 = (MergedComputeGroup) cg4;
- Assert.assertTrue(cg5.getComputeGroupNameSet().size() == 1);
- Assert.assertTrue(cg5.getComputeGroupNameSet().contains("zone1"));
+ Assert.assertTrue(cg5.getNames().size() == 1);
+ Assert.assertTrue(cg5.getNames().contains("zone1"));
// update connection context and query, it will failed because no zone1 backend
connectContext.setComputeGroup(cg4);
diff --git a/fe/fe-core/src/test/java/org/apache/doris/resource/ComputeGroupTest.java b/fe/fe-core/src/test/java/org/apache/doris/resource/ComputeGroupTest.java
index 2ad93dfbf82..0242055078d 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/resource/ComputeGroupTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/resource/ComputeGroupTest.java
@@ -159,14 +159,14 @@ public class ComputeGroupTest {
auth.createUser(createNonAdminUser);
ComputeGroup cg = auth.getComputeGroup(nonAdminUserStr);
Assert.assertTrue(cg instanceof MergedComputeGroup);
- Assert.assertTrue(((MergedComputeGroup) cg).getComputeGroupNameSet().contains(Tag.VALUE_DEFAULT_TAG));
+ Assert.assertTrue(((MergedComputeGroup) cg).getNames().contains(Tag.VALUE_DEFAULT_TAG));
// 2.1 get a non-admin user with resource tag
String setPropStr = "set property for '" + nonAdminUserStr + "' 'resource_tags.location' = 'test_rg1';";
ExceptionChecker.expectThrowsNoException(() -> setProperty(setPropStr));
ComputeGroup cg2 = auth.getComputeGroup(nonAdminUserStr);
Assert.assertTrue(cg2 instanceof MergedComputeGroup);
- Assert.assertTrue(((MergedComputeGroup) cg2).getComputeGroupNameSet().contains("test_rg1"));
+ Assert.assertTrue(((MergedComputeGroup) cg2).getNames().contains("test_rg1"));
// 2.2 get a non-admin user with multi-resource tag
String setPropStr2 = "set property for '" + nonAdminUserStr
@@ -174,7 +174,7 @@ public class ComputeGroupTest {
ExceptionChecker.expectThrowsNoException(() -> setProperty(setPropStr2));
ComputeGroup cg3 = auth.getComputeGroup(nonAdminUserStr);
Assert.assertTrue(cg3 instanceof MergedComputeGroup);
- Set<String> cgNameSet = ((MergedComputeGroup) cg3).getComputeGroupNameSet();
+ Set<String> cgNameSet = ((MergedComputeGroup) cg3).getNames();
Assert.assertTrue(cgNameSet.contains("test_rg1"));
Assert.assertTrue(cgNameSet.contains("test_rg2"));
Assert.assertTrue(cgNameSet.size() == 2);
@@ -185,7 +185,7 @@ public class ComputeGroupTest {
ExceptionChecker.expectThrowsNoException(() -> setProperty(setPropStr3));
ComputeGroup cg4 = auth.getComputeGroup(nonAdminUserStr);
Assert.assertTrue(cg4 instanceof MergedComputeGroup);
- Set<String> cgNameSets = ((MergedComputeGroup) cg4).getComputeGroupNameSet();
+ Set<String> cgNameSets = ((MergedComputeGroup) cg4).getNames();
Assert.assertTrue(cgNameSets.size() == 1);
Assert.assertTrue(cgNameSets.contains("default"));
}
@@ -200,7 +200,7 @@ public class ComputeGroupTest {
ExceptionChecker.expectThrowsNoException(() -> setProperty(setPropStr));
ComputeGroup cg2 = auth.getComputeGroup("root");
Assert.assertTrue(cg2 instanceof MergedComputeGroup);
- Assert.assertTrue(((MergedComputeGroup) cg2).getComputeGroupNameSet().contains("test_rg2"));
+ Assert.assertTrue(((MergedComputeGroup) cg2).getNames().contains("test_rg2"));
// 4.2 get an admin user with an empty resource tag
@@ -572,7 +572,7 @@ public class ComputeGroupTest {
brokerLoadJob.setComputeGroup();
ComputeGroup cg = ConnectContext.get().getComputeGroupSafely();
Assert.assertTrue(cg instanceof MergedComputeGroup);
- Assert.assertTrue(((MergedComputeGroup) cg).getComputeGroupNameSet().contains(tagName));
+ Assert.assertTrue(((MergedComputeGroup) cg).getNames().contains(tagName));
}
}
@@ -620,7 +620,7 @@ public class ComputeGroupTest {
job.setComputeGroup();
ComputeGroup cg = ConnectContext.get().getComputeGroupSafely();
Assert.assertTrue(cg instanceof MergedComputeGroup);
- Assert.assertTrue(((MergedComputeGroup) cg).getComputeGroupNameSet().contains("tag_rg_1"));
+ Assert.assertTrue(((MergedComputeGroup) cg).getNames().contains("tag_rg_1"));
}
// 4 get a null job
diff --git a/fe/fe-core/src/test/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgrTest.java b/fe/fe-core/src/test/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgrTest.java
index d729881358e..323f95f1f15 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgrTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgrTest.java
@@ -17,8 +17,6 @@
package org.apache.doris.resource.workloadgroup;
-import org.apache.doris.analysis.AlterWorkloadGroupStmt;
-import org.apache.doris.analysis.CreateWorkloadGroupStmt;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
@@ -29,9 +27,12 @@ import org.apache.doris.mysql.privilege.Auth;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.persist.EditLog;
import org.apache.doris.qe.ConnectContext;
-import org.apache.doris.thrift.TopicInfo;
+import org.apache.doris.resource.computegroup.ComputeGroup;
+import org.apache.doris.resource.computegroup.MergedComputeGroup;
+import org.apache.doris.thrift.TPipelineWorkloadGroup;
import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
import mockit.Delegate;
import mockit.Expectations;
import mockit.Injectable;
@@ -42,6 +43,7 @@ import org.junit.Test;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
public class WorkloadGroupMgrTest {
@@ -111,350 +113,451 @@ public class WorkloadGroupMgrTest {
public void testCreateWorkloadGroup() throws DdlException {
Config.enable_workload_group = true;
WorkloadGroupMgr workloadGroupMgr = new WorkloadGroupMgr();
+ workloadGroupMgr.getIdToWorkloadGroup().clear();
+ workloadGroupMgr.getNameToWorkloadGroup().clear();
+
+ // 1 create workload group 1
Map<String, String> properties1 = Maps.newHashMap();
+ String cg1 = "cg1";
+ long wgId1 = 1;
+ String wgName1 = "wg1";
properties1.put(WorkloadGroup.CPU_SHARE, "10");
- properties1.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- String name1 = "g1";
- CreateWorkloadGroupStmt stmt1 = new CreateWorkloadGroupStmt(false, name1, properties1);
- workloadGroupMgr.createWorkloadGroup(stmt1);
-
- Map<String, WorkloadGroup> nameToRG = workloadGroupMgr.getNameToWorkloadGroup();
- Assert.assertEquals(2, nameToRG.size());
- Assert.assertTrue(nameToRG.containsKey(name1));
- WorkloadGroup group1 = nameToRG.get(name1);
- Assert.assertEquals(name1, group1.getName());
+ properties1.put(WorkloadGroup.MEMORY_LIMIT, "50%");
+ properties1.put(WorkloadGroup.COMPUTE_GROUP, cg1);
+ WorkloadGroup wg1 = new WorkloadGroup(wgId1, wgName1, properties1);
+ workloadGroupMgr.createWorkloadGroup(cg1, wg1, false);
+
+ Pair<String, String> key1 = Pair.of(cg1, wgName1);
+ Map<Pair<String, String>, WorkloadGroup> nameToRG = workloadGroupMgr.getNameToWorkloadGroup();
+ Assert.assertEquals(1, nameToRG.size());
+ Assert.assertTrue(nameToRG.containsKey(key1));
+ WorkloadGroup group1 = nameToRG.get(key1);
+ Assert.assertEquals(key1.second, group1.getName());
+ Assert.assertEquals(key1.first, group1.getComputeGroup());
Map<Long, WorkloadGroup> idToRG = workloadGroupMgr.getIdToWorkloadGroup();
- Assert.assertEquals(2, idToRG.size());
+ Assert.assertEquals(1, idToRG.size());
Assert.assertTrue(idToRG.containsKey(group1.getId()));
+ // 2 create workload group 2
+ long wgId2 = 2;
+ String cg2 = "cg2";
+ String wgName2 = "wg2";
Map<String, String> properties2 = Maps.newHashMap();
properties2.put(WorkloadGroup.CPU_SHARE, "20");
properties2.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- String name2 = "g2";
- CreateWorkloadGroupStmt stmt2 = new CreateWorkloadGroupStmt(false, name2, properties2);
- workloadGroupMgr.createWorkloadGroup(stmt2);
+ properties2.put(WorkloadGroup.COMPUTE_GROUP, cg2);
+ WorkloadGroup wg2 = new WorkloadGroup(wgId2, wgName2, properties2);
+ workloadGroupMgr.createWorkloadGroup(cg2, wg2, false);
+ Pair<String, String> key2 = Pair.of(cg2, wgName2);
nameToRG = workloadGroupMgr.getNameToWorkloadGroup();
- Assert.assertEquals(3, nameToRG.size());
- Assert.assertTrue(nameToRG.containsKey(name2));
- WorkloadGroup group2 = nameToRG.get(name2);
+ Assert.assertEquals(2, nameToRG.size());
+ Assert.assertTrue(nameToRG.containsKey(key2));
+ WorkloadGroup group2 = nameToRG.get(key2);
idToRG = workloadGroupMgr.getIdToWorkloadGroup();
- Assert.assertEquals(3, idToRG.size());
+ Assert.assertEquals(2, idToRG.size());
Assert.assertTrue(idToRG.containsKey(group2.getId()));
-
+ Assert.assertTrue(key2.first.equals(wg2.getComputeGroup()));
+
+ // 3 test memory limit exceeds
+ Map<String, String> properties3 = Maps.newHashMap();
+ properties3.put(WorkloadGroup.CPU_SHARE, "20");
+ properties3.put(WorkloadGroup.MEMORY_LIMIT, "90%");
+ properties3.put(WorkloadGroup.COMPUTE_GROUP, cg1);
+ String wgName3 = "wg3";
+ long wgId3 = 3;
try {
- workloadGroupMgr.createWorkloadGroup(stmt2);
+ workloadGroupMgr.createWorkloadGroup(cg1, new WorkloadGroup(wgId3, wgName3, properties3), false);
Assert.fail();
} catch (DdlException e) {
- Assert.assertTrue(e.getMessage().contains("already exist"));
+ Assert.assertTrue(e.getMessage().contains("can not be greater"));
}
+ properties3.put(WorkloadGroup.MEMORY_LIMIT, "1%");
+ workloadGroupMgr.createWorkloadGroup(cg1, new WorkloadGroup(wgId3, wgName3, properties3), false);
+
- CreateWorkloadGroupStmt stmt3 = new CreateWorkloadGroupStmt(true, name2, properties2);
- workloadGroupMgr.createWorkloadGroup(stmt3);
- Assert.assertEquals(3, workloadGroupMgr.getIdToWorkloadGroup().size());
- Assert.assertEquals(3, workloadGroupMgr.getNameToWorkloadGroup().size());
+ // 4 test create duplicate workload group error.
+ workloadGroupMgr.createWorkloadGroup(cg1, wg1, true);
+ try {
+ // create wg1 in cg1, it should fail
+ workloadGroupMgr.createWorkloadGroup(cg1, new WorkloadGroup(4, wgName1, properties1), false);
+ Assert.fail();
+ } catch (DdlException e) {
+ Assert.assertTrue(e.getMessage().contains("already has workload group"));
+ }
+ Map<String, String> properties4 = Maps.newHashMap();
+ properties4.put(WorkloadGroup.CPU_SHARE, "10");
+ properties4.put(WorkloadGroup.COMPUTE_GROUP, cg2);
+ // create wg1 in cg2, it should be success
+ workloadGroupMgr.createWorkloadGroup(cg2, new WorkloadGroup(4, wgName1, properties4), false);
}
@Test
public void testGetWorkloadGroup() throws UserException {
Config.enable_workload_group = true;
- ConnectContext context = new ConnectContext();
WorkloadGroupMgr workloadGroupMgr = new WorkloadGroupMgr();
+ long wgId1 = 1;
+ String wgName1 = "wg1";
+ String cgName1 = "cg1";
Map<String, String> properties1 = Maps.newHashMap();
properties1.put(WorkloadGroup.CPU_SHARE, "10");
- properties1.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- String name1 = "g1";
- CreateWorkloadGroupStmt stmt1 = new CreateWorkloadGroupStmt(false, name1, properties1);
- workloadGroupMgr.createWorkloadGroup(stmt1);
- context.getSessionVariable().setWorkloadGroup(name1);
- List<TopicInfo> tWorkloadGroups1 = workloadGroupMgr.getPublishTopicInfo();
- Assert.assertEquals(2, tWorkloadGroups1.size());
- TopicInfo tWorkloadGroup1 = null;
- for (int i = 0; i < 2; ++i) {
- if (tWorkloadGroups1.get(i).getWorkloadGroupInfo().getName().equals(name1)) {
- tWorkloadGroup1 = tWorkloadGroups1.get(i);
- break;
- }
+ properties1.put(WorkloadGroup.COMPUTE_GROUP, cgName1);
+ workloadGroupMgr.createWorkloadGroup(cgName1, new WorkloadGroup(wgId1, wgName1, properties1), false);
+
+ long wgId2 = 2;
+ String wgName2 = "wg2";
+ String cgName2 = "cg2";
+ Map<String, String> properties2 = Maps.newHashMap();
+ properties2.put(WorkloadGroup.CPU_SHARE, "20");
+ properties2.put(WorkloadGroup.COMPUTE_GROUP, cgName2);
+ workloadGroupMgr.createWorkloadGroup(cgName2, new WorkloadGroup(wgId2, wgName2, properties2), false);
+
+ workloadGroupMgr.createWorkloadGroup(cgName1, new WorkloadGroup(100, "normal", properties1), false);
+ workloadGroupMgr.createWorkloadGroup(cgName2, new WorkloadGroup(101, "normal", properties1), false);
+
+
+ // 1 test get workload group by ConnectContext
+ ConnectContext ctx = new ConnectContext();
+ // 1.1 not set wg, get normal
+ ctx.setComputeGroup(new ComputeGroup("1", cgName1, null));
+ List<TPipelineWorkloadGroup> ret = workloadGroupMgr.getWorkloadGroup(ctx);
+ Assert.assertTrue(ret.get(0).getId() == 100);
+
+ ctx.setComputeGroup(new ComputeGroup("2", cgName2, null));
+ Assert.assertTrue(workloadGroupMgr.getWorkloadGroup(ctx).get(0).getId() == 101);
+
+
+ // 1.2 get from user prop
+
+ // 1.3 get from session
+ ctx.getSessionVariable().setWorkloadGroup(wgName2);
+ Assert.assertTrue(workloadGroupMgr.getWorkloadGroup(ctx).size() == 1);
+ Assert.assertTrue(workloadGroupMgr.getWorkloadGroup(ctx).get(0).getId() == wgId2);
+
+ // 1.4 get multi workload group
+ Set<String> cgSet = Sets.newHashSet();
+ cgSet.add(cgName1);
+ cgSet.add(cgName2);
+ long wgId3 = 3;
+ ComputeGroup mergedComputeGroup = new MergedComputeGroup(cgSet, null);
+ ctx.setComputeGroup(mergedComputeGroup);
+ Map<String, String> prop3 = Maps.newHashMap();
+ prop3.put(WorkloadGroup.COMPUTE_GROUP, cgName1);
+ workloadGroupMgr.createWorkloadGroup(cgName1, new WorkloadGroup(wgId3, wgName2, prop3), false);
+ List<TPipelineWorkloadGroup> tPipWgList = workloadGroupMgr.getWorkloadGroup(ctx);
+ Set<Long> idSet = Sets.newHashSet();
+ for (TPipelineWorkloadGroup tpip : tPipWgList) {
+ idSet.add(tpip.getId());
}
- Assert.assertEquals(name1, tWorkloadGroup1.getWorkloadGroupInfo().getName());
- Assert.assertTrue(tWorkloadGroup1.getWorkloadGroupInfo().getCpuShare() == 10);
+ Assert.assertTrue(idSet.size() == 2);
+ Assert.assertTrue(idSet.contains(wgId2));
+ Assert.assertTrue(idSet.contains(wgId3));
+
+ // 1.5 test get failed
+ ctx.getSessionVariable().setWorkloadGroup("abc");
try {
- context.getSessionVariable().setWorkloadGroup("g2");
- workloadGroupMgr.getWorkloadGroup(context);
+ workloadGroupMgr.getWorkloadGroup(ctx);
Assert.fail();
- } catch (UserException e) {
- Assert.assertTrue(e.getMessage().contains("does not exist"));
+ } catch (DdlException e) {
+ Assert.assertTrue(e.getMessage().contains("Can not find workload group"));
}
}
@Test
public void testAlterWorkloadGroup() throws UserException {
Config.enable_workload_group = true;
- ConnectContext context = new ConnectContext();
WorkloadGroupMgr workloadGroupMgr = new WorkloadGroupMgr();
+ workloadGroupMgr.getIdToWorkloadGroup().clear();
+ workloadGroupMgr.getNameToWorkloadGroup().clear();
+
Map<String, String> p0 = Maps.newHashMap();
- String name = "g1";
try {
- AlterWorkloadGroupStmt stmt1 = new AlterWorkloadGroupStmt(name, p0);
- workloadGroupMgr.alterWorkloadGroup(stmt1);
+ workloadGroupMgr.alterWorkloadGroup("", "", p0);
} catch (DdlException e) {
- Assert.assertTrue(e.getMessage().contains("alter workload group should contain at least one property"));
+ Assert.assertTrue(e.getMessage().contains("should contain at least one property"));
}
p0.put(WorkloadGroup.CPU_SHARE, "10");
try {
- AlterWorkloadGroupStmt stmt1 = new AlterWorkloadGroupStmt(name, p0);
- workloadGroupMgr.alterWorkloadGroup(stmt1);
+ workloadGroupMgr.alterWorkloadGroup("", "abc", p0);
} catch (DdlException e) {
- Assert.assertTrue(e.getMessage().contains("does not exist"));
+ Assert.assertTrue(e.getMessage().contains("Can not find workload group"));
}
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.CPU_SHARE, "10");
- properties.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
-
- Map<String, String> newProperties = Maps.newHashMap();
- newProperties.put(WorkloadGroup.CPU_SHARE, "5");
- newProperties.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- AlterWorkloadGroupStmt stmt2 = new AlterWorkloadGroupStmt(name, newProperties);
- workloadGroupMgr.alterWorkloadGroup(stmt2);
-
- context.getSessionVariable().setWorkloadGroup(name);
- List<TopicInfo> tWorkloadGroups = workloadGroupMgr.getPublishTopicInfo();
- Assert.assertEquals(2, tWorkloadGroups.size());
- TopicInfo tWorkloadGroup1 = null;
- for (int i = 0; i < 2; ++i) {
- if (tWorkloadGroups.get(i).getWorkloadGroupInfo().getName().equals(name)) {
- tWorkloadGroup1 = tWorkloadGroups.get(i);
- break;
- }
+ long wgId1 = 1;
+ String wgName1 = "wg1";
+ String cgName1 = "cg1";
+ Map<String, String> prop1 = Maps.newHashMap();
+ prop1.put(WorkloadGroup.COMPUTE_GROUP, cgName1);
+ prop1.put(WorkloadGroup.CPU_SHARE, "10");
+ workloadGroupMgr.createWorkloadGroup(cgName1, new WorkloadGroup(wgId1, wgName1, prop1), false);
+ Assert.assertTrue(Long.valueOf(
+ workloadGroupMgr.getNameToWorkloadGroup().get(Pair.of(cgName1, wgName1)).getProperties()
+ .get(WorkloadGroup.CPU_SHARE)) == 10);
+
+ // test alter failed
+ String cgName2 = "cg2";
+ String wgName2 = "wg2";
+ Map<String, String> prop2 = Maps.newHashMap();
+ prop2.put(WorkloadGroup.CPU_SHARE, "20");
+ try {
+ workloadGroupMgr.alterWorkloadGroup(cgName2, wgName2, prop2);
+ Assert.fail();
+ } catch (DdlException e) {
+ Assert.assertTrue(e.getMessage().contains("Can not find workload group"));
}
- Assert.assertTrue(tWorkloadGroup1.getWorkloadGroupInfo().getCpuShare() == 5);
+
+ // test alter success
+ workloadGroupMgr.alterWorkloadGroup(cgName1, wgName1, prop2);
+ WorkloadGroup wg = workloadGroupMgr.getNameToWorkloadGroup().get(Pair.of(cgName1, wgName1));
+ Assert.assertTrue(Long.valueOf(wg.getProperties().get(WorkloadGroup.CPU_SHARE)) == 20);
}
+ // before:
+ // empty cg: wg1, wg2, wg3(id=3)
+ // cg1: wg3(id=4)
+ // after:
+ // cg1: wg1, wg2, wg3(id=4)
+ // cg2: wg1, wg2, wg3
@Test
- public void testMultiTagCreateWorkloadGroup() throws UserException {
- Config.enable_workload_group = true;
- WorkloadGroupMgr workloadGroupMgr = new WorkloadGroupMgr();
-
- {
- String name = "empty_g1";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "50%");
- properties.put(WorkloadGroup.TAG, "");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- }
-
- {
- String name = "empty_g2";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "10%");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- }
-
- {
- String name = "not_empty_g1";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- properties.put(WorkloadGroup.TAG, "cn1,cn2");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- }
-
- {
- String name = "not_empty_g2";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- properties.put(WorkloadGroup.TAG, "cn3,cn2");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- }
-
-
- {
- String name = "not_empty_g3";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "70%");
- properties.put(WorkloadGroup.TAG, "cn2,cn100");
- try {
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- } catch (DdlException e) {
- Assert.assertTrue(e.getMessage().contains("The sum of all workload group " + WorkloadGroup.MEMORY_LIMIT));
- }
- }
-
- {
- String name = "not_empty_g3";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "70%");
- properties.put(WorkloadGroup.TAG, "cn3,cn100");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- }
-
- {
- String name = "not_empty_g5";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "70%");
- properties.put(WorkloadGroup.TAG, "cn5");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- }
+ public void testBindWorkloadGroupToCg() throws DdlException {
+ WorkloadGroupMgr wgMgr = new WorkloadGroupMgr();
+ wgMgr.getIdToNameMap().clear();
+ wgMgr.getNameToWorkloadGroup().clear();
+
+ long wgId1 = 1;
+ String wgName1 = "wg1";
+ Map<String, String> prop1 = Maps.newHashMap();
+ prop1.put(WorkloadGroup.CPU_SHARE, "123");
+ WorkloadGroup wg1 = new WorkloadGroup(wgId1, wgName1, prop1);
+ wgMgr.getIdToWorkloadGroup().put(wgId1, wg1);
+ wgMgr.getNameToWorkloadGroup().put(Pair.of(WorkloadGroupMgr.EMPTY_COMPUTE_GROUP, wgName1), wg1);
+ wgMgr.getIdToQueryQueue().put(wgId1, new QueryQueue(0, 0, 0, 0, 0));
+
+ long wgId2 = 2;
+ String wgName2 = "wg2";
+ Map<String, String> prop2 = Maps.newHashMap();
+ prop2.put(WorkloadGroup.CPU_SHARE, "123");
+ WorkloadGroup wg2 = new WorkloadGroup(wgId2, wgName2, prop2);
+ wgMgr.getIdToWorkloadGroup().put(wgId2, wg2);
+ wgMgr.getNameToWorkloadGroup().put(Pair.of(WorkloadGroupMgr.EMPTY_COMPUTE_GROUP, wgName2), wg2);
+ wgMgr.getIdToQueryQueue().put(wgId2, new QueryQueue(0, 0, 0, 0, 0));
+
+ long wgId3 = 3;
+ String wgName3 = "wg3";
+ Map<String, String> prop3 = Maps.newHashMap();
+ prop3.put(WorkloadGroup.CPU_SHARE, "123");
+ WorkloadGroup wg3 = new WorkloadGroup(wgId3, wgName3, prop3);
+ wgMgr.getIdToWorkloadGroup().put(wgId3, wg3);
+ wgMgr.getNameToWorkloadGroup().put(Pair.of(WorkloadGroupMgr.EMPTY_COMPUTE_GROUP, wgName3), wg3);
+ wgMgr.getIdToQueryQueue().put(wgId3, new QueryQueue(0, 0, 0, 0, 0));
+
+
+ // create a duplicate wg3 which binds to a compute group
+ String cg1 = "cg1";
+
+ long wgId4 = 4;
+ String wgName4 = wgName3;
+ Map<String, String> prop4 = Maps.newHashMap();
+ prop4.put(WorkloadGroup.CPU_SHARE, "456");
+ prop4.put(WorkloadGroup.COMPUTE_GROUP, cg1);
+ WorkloadGroup wg4 = new WorkloadGroup(wgId4, wgName4, prop4);
+ wgMgr.createWorkloadGroup(cg1, wg4, false);
+ wgMgr.getIdToQueryQueue().put(wgId4, new QueryQueue(0, 0, 0, 0, 0));
+
+
+ Assert.assertTrue(wgMgr.getIdToWorkloadGroup().size() == 4);
+ Assert.assertTrue(wgMgr.getNameToWorkloadGroup().size() == 4);
+ Assert.assertTrue(wgMgr.getIdToQueryQueue().size() == 4);
+ Assert.assertTrue(wgMgr.getOldWorkloadGroup().size() == 3);
+
+ String cg2 = "cg2";
+ Set<String> cgSet = Sets.newHashSet();
+ cgSet.add(cg1);
+ cgSet.add(cg2);
+
+ for (WorkloadGroup oldWg : wgMgr.getOldWorkloadGroup()) {
+ wgMgr.bindWorkloadGroupToComputeGroup(cgSet, oldWg);
+ }
- {
- String name = "not_empty_g6";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- properties.put(WorkloadGroup.TAG, "cn5");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
+ Assert.assertTrue(wgMgr.getIdToWorkloadGroup().size() == 6);
+ Assert.assertTrue(wgMgr.getNameToWorkloadGroup().size() == 6);
+ Assert.assertTrue(wgMgr.getIdToQueryQueue().size() == 1);
+ Assert.assertTrue(wgMgr.getOldWorkloadGroup().size() == 0);
+ Assert.assertTrue(wgMgr.getIdToWorkloadGroup().get(wgId1) == null);
+ Assert.assertTrue(wgMgr.getIdToWorkloadGroup().get(wgId2) == null);
+ Assert.assertTrue(wgMgr.getIdToWorkloadGroup().get(wgId3) == null);
+ Assert.assertTrue(wgMgr.getIdToWorkloadGroup().get(wgId4).equals(wg4));
+
+ for (String cgName : cgSet) {
+ WorkloadGroup wg11 = wgMgr.getNameToWorkloadGroup().get(Pair.of(cgName, wgName1));
+ WorkloadGroup wg22 = wgMgr.getNameToWorkloadGroup().get(Pair.of(cgName, wgName2));
+ WorkloadGroup wg33 = wgMgr.getNameToWorkloadGroup().get(Pair.of(cgName, wgName3));
+
+ Assert.assertTrue(wgMgr.getIdToWorkloadGroup().containsKey(wg11.getId()));
+ Assert.assertTrue(wgMgr.getIdToWorkloadGroup().containsKey(wg22.getId()));
+ Assert.assertTrue(wgMgr.getIdToWorkloadGroup().containsKey(wg33.getId()));
+
+ Assert.assertTrue(wg11.getComputeGroup().equals(cgName));
+ Assert.assertTrue(wg22.getComputeGroup().equals(cgName));
+ Assert.assertTrue(wg33.getComputeGroup().equals(cgName));
+
+ Assert.assertTrue(wg11.getProperties().get(WorkloadGroup.CPU_SHARE).equals("123"));
+ Assert.assertTrue(wg22.getProperties().get(WorkloadGroup.CPU_SHARE).equals("123"));
+ if (cg1.equals(cgName)) {
+ Assert.assertTrue(wg33.getProperties().get(WorkloadGroup.CPU_SHARE).equals("456"));
+ } else {
+ Assert.assertTrue(wg33.getProperties().get(WorkloadGroup.CPU_SHARE).equals("123"));
}
- {
- String name = "not_empty_g7";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "70%");
- properties.put(WorkloadGroup.TAG, "cn5");
- try {
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- } catch (DdlException e) {
- Assert.assertTrue(e.getMessage().contains("The sum of all workload group " + WorkloadGroup.MEMORY_LIMIT));
- }
- }
+ }
}
-
@Test
- public void testMultiTagAlterWorkloadGroup() throws UserException {
+ public void testMultiTagCreateWorkloadGroup() throws UserException {
Config.enable_workload_group = true;
- WorkloadGroupMgr workloadGroupMgr = new WorkloadGroupMgr();
- {
- String name = "empty_g1";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "50%");
- properties.put(WorkloadGroup.TAG, "");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- }
-
- {
- String name = "empty_g2";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "10%");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- }
-
- {
- String name = "not_empty_g1";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- properties.put(WorkloadGroup.TAG, "cn1,cn2");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
+ String[] props = {WorkloadGroup.MEMORY_LIMIT, WorkloadGroup.CPU_HARD_LIMIT};
+ for (String propName : props) {
+ WorkloadGroupMgr workloadGroupMgr = new WorkloadGroupMgr();
+
+ // create cg1.wg1 with 30%
+ String cgName1 = "cg1";
+ String wgName11 = "wg1";
+ Map<String, String> prop1 = Maps.newHashMap();
+ prop1.put(WorkloadGroup.COMPUTE_GROUP, cgName1);
+ prop1.put(propName, "30%");
+ workloadGroupMgr.createWorkloadGroup(cgName1, new WorkloadGroup(1, wgName11, prop1), false);
+
+ // create cg2.wg1 with 71%
+ String cgName2 = "cg2";
+ String wgName21 = "wg1";
+ Map<String, String> prop2 = Maps.newHashMap();
+ prop2.put(WorkloadGroup.COMPUTE_GROUP, cgName2);
+ prop2.put(propName, "71%");
+ workloadGroupMgr.createWorkloadGroup(cgName2, new WorkloadGroup(2, wgName21, prop2), false);
+
+ // create cg1.wg2 with 71%, it should be failed.
+ String wgName12 = "wg2";
+ Map<String, String> prop3 = Maps.newHashMap();
+ prop3.put(WorkloadGroup.COMPUTE_GROUP, cgName1);
+ prop3.put(propName, "71%");
+ try {
+ workloadGroupMgr.createWorkloadGroup(cgName1, new WorkloadGroup(3, wgName12, prop3), false);
+ Assert.fail();
+ } catch (DdlException e) {
+ Assert.assertTrue(e.getMessage().contains("can not be greater"));
}
- {
- String name = "not_empty_g2";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- properties.put(WorkloadGroup.TAG, "cn3,cn2");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
+ // reset limit, then create should succ
+ prop3.put(propName, "70%");
+ workloadGroupMgr.createWorkloadGroup(cgName1, new WorkloadGroup(3, wgName12, prop3), false);
+
+ // create cg1.wg3, it should be failed and sum value should be 100
+ String wgName13 = "wg3";
+ Map<String, String> prop4 = Maps.newHashMap();
+ prop4.put(WorkloadGroup.COMPUTE_GROUP, cgName1);
+ prop4.put(propName, "1%");
+ try {
+ workloadGroupMgr.createWorkloadGroup(cgName1, new WorkloadGroup(4, wgName13, prop4), false);
+ Assert.fail();
+ } catch (DdlException e) {
+ Assert.assertTrue(e.getMessage().contains("current sum val:101"));
}
- {
- String name = "not_empty_g3";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- properties.put(WorkloadGroup.TAG, "cn2,cn100");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
+ // create cg2.wg2 with limit 20%, it should be succ.
+ String wgName22 = "wg2";
+ Map<String, String> prop5 = Maps.newHashMap();
+ prop5.put(WorkloadGroup.COMPUTE_GROUP, cgName2);
+ prop5.put(propName, "9%");
+ workloadGroupMgr.createWorkloadGroup(cgName2, new WorkloadGroup(5, wgName22, prop5), false);
+
+ // create cg2.wg3 with limit 60, it should be failed.
+ String wgName23 = "wg3";
+ Map<String, String> prop6 = Maps.newHashMap();
+ prop6.put(WorkloadGroup.COMPUTE_GROUP, cgName2);
+ prop6.put(propName, "30%");
+ try {
+ workloadGroupMgr.createWorkloadGroup(cgName2, new WorkloadGroup(6, wgName23, prop6), false);
+ } catch (DdlException e) {
+ Assert.assertTrue(e.getMessage().contains("current sum val:110"));
}
- {
- String name = "not_empty_g3";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "70%");
- properties.put(WorkloadGroup.TAG, "cn2,cn100");
- AlterWorkloadGroupStmt alterStmt = new AlterWorkloadGroupStmt(name, properties);
- try {
- workloadGroupMgr.alterWorkloadGroup(alterStmt);
- } catch (DdlException e) {
- Assert.assertTrue(e.getMessage().contains("The sum of all workload group " + WorkloadGroup.MEMORY_LIMIT));
- }
- }
+ }
}
@Test
- public void testMultiTagCreateWorkloadGroupWithNoTag() throws UserException {
+ public void testMultiTagAlterWorkloadGroup() throws UserException {
Config.enable_workload_group = true;
- WorkloadGroupMgr workloadGroupMgr = new WorkloadGroupMgr();
+ String[] props = {WorkloadGroup.MEMORY_LIMIT, WorkloadGroup.CPU_HARD_LIMIT};
+ String cg1 = "cg1";
+ String cg2 = "cg2";
+ for (String prop : props) {
+ WorkloadGroupMgr workloadGroupMgr = new WorkloadGroupMgr();
+ // create 3 wgs for cg1
+ {
+ String wgName = "wg1";
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(prop, "10%");
+ properties.put(WorkloadGroup.COMPUTE_GROUP, cg1);
+ workloadGroupMgr.createWorkloadGroup(cg1, new WorkloadGroup(1, wgName, properties), false);
+ }
- {
- String name = "not_empty_g1";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- properties.put(WorkloadGroup.TAG, "cn1,cn2");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- }
+ {
+ String wgName = "wg2";
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(prop, "10%");
+ properties.put(WorkloadGroup.COMPUTE_GROUP, cg1);
+ workloadGroupMgr.createWorkloadGroup(cg1, new WorkloadGroup(2, wgName, properties), false);
+ }
- {
- String name = "not_empty_g2";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- properties.put(WorkloadGroup.TAG, "cn3,cn2");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- }
+ {
+ String wgName = "wg3";
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(prop, "10%");
+ properties.put(WorkloadGroup.COMPUTE_GROUP, cg1);
+ workloadGroupMgr.createWorkloadGroup(cg1, new WorkloadGroup(3, wgName, properties), false);
+ }
- // create not tag workload group
- {
- String name = "no_tag_g1";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "10%");
- properties.put(WorkloadGroup.TAG, "");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- }
+ // create 2 wgs for cg2
+ {
+ String wgName = "wg1";
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(prop, "10%");
+ properties.put(WorkloadGroup.COMPUTE_GROUP, cg2);
+ workloadGroupMgr.createWorkloadGroup(cg2, new WorkloadGroup(4, wgName, properties), false);
+ }
- {
- String name = "no_tag_g2";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- }
+ {
+ String wgName = "wg2";
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(prop, "10%");
+ properties.put(WorkloadGroup.COMPUTE_GROUP, cg2);
+ workloadGroupMgr.createWorkloadGroup(cg2, new WorkloadGroup(5, wgName, properties), false);
+ }
- {
- String name = "no_tag_g3";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "70%");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- try {
- workloadGroupMgr.createWorkloadGroup(createStmt);
- } catch (DdlException e) {
- Assert.assertTrue(e.getMessage().contains("The sum of all workload group " + WorkloadGroup.MEMORY_LIMIT));
+ // alter cg1.wg1 failed
+ {
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(prop, "90%");
+ try {
+ workloadGroupMgr.alterWorkloadGroup(cg1, "wg1", properties);
+ } catch (DdlException e) {
+ Assert.assertTrue(e.getMessage().contains("current sum val:110"));
+ Assert.assertTrue(e.getMessage().contains("cg1"));
+ Assert.assertFalse(e.getMessage().contains("cg2"));
+ }
}
- }
- {
- String name = "no_tag_g3";
- Map<String, String> properties = Maps.newHashMap();
- properties.put(WorkloadGroup.MEMORY_LIMIT, "30%");
- CreateWorkloadGroupStmt createStmt = new CreateWorkloadGroupStmt(false, name, properties);
- workloadGroupMgr.createWorkloadGroup(createStmt);
- }
+ // alter cg1.wg1 succ
+ {
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(prop, "20%");
+ workloadGroupMgr.alterWorkloadGroup(cg1, "wg1", properties);
+ }
+ }
}
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/resource/workloadgroup/WorkloadGroupTest.java b/fe/fe-core/src/test/java/org/apache/doris/resource/workloadgroup/WorkloadGroupTest.java
index 729cbc5c886..a7176f2d50a 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/resource/workloadgroup/WorkloadGroupTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/resource/workloadgroup/WorkloadGroupTest.java
@@ -35,10 +35,11 @@ public class WorkloadGroupTest {
Map<String, String> properties1 = Maps.newHashMap();
properties1.put(WorkloadGroup.CPU_SHARE, "10");
properties1.put(WorkloadGroup.MEMORY_LIMIT, "30%");
+ properties1.put(WorkloadGroup.COMPUTE_GROUP, "default");
String name1 = "g1";
WorkloadGroup group1 = WorkloadGroup.create(name1, properties1);
Assert.assertEquals(name1, group1.getName());
- Assert.assertEquals(7, group1.getProperties().size());
+ Assert.assertEquals(8, group1.getProperties().size());
Assert.assertTrue(group1.getProperties().containsKey(WorkloadGroup.CPU_SHARE));
Assert.assertTrue(Math.abs(group1.getMemoryLimitPercent() - 30) < 1e-6);
}
@@ -80,6 +81,7 @@ public class WorkloadGroupTest {
Map<String, String> properties1 = Maps.newHashMap();
properties1.put(WorkloadGroup.CPU_SHARE, "10");
properties1.put(WorkloadGroup.MEMORY_LIMIT, "30%");
+ properties1.put(WorkloadGroup.COMPUTE_GROUP, "default");
String name1 = "g1";
WorkloadGroup group1 = WorkloadGroup.create(name1, properties1);
diff --git a/fe/fe-core/src/test/java/org/apache/doris/utframe/TestWithFeService.java b/fe/fe-core/src/test/java/org/apache/doris/utframe/TestWithFeService.java
index 9b51382cb9f..c81ca8a206b 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/utframe/TestWithFeService.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/utframe/TestWithFeService.java
@@ -158,12 +158,13 @@ public abstract class TestWithFeService {
// this.enableAdvanceNextId may be reset by children classes
Config.enable_advance_next_id = this.enableAdvanceNextId;
FeConstants.enableInternalSchemaDb = false;
- FeConstants.shouldCreateInternalWorkloadGroup = false;
+ FeConstants.bindWgToComputeGroup = false;
beforeCreatingConnectContext();
connectContext = createDefaultCtx();
beforeCluster();
createDorisCluster();
runBeforeAll();
+ Env.getCurrentEnv().getWorkloadGroupMgr().tryCreateNormalWorkloadGroup();
}
protected void beforeCluster() {
diff --git a/regression-test/data/workload_manager_p0/test_curd_wlg.out b/regression-test/data/workload_manager_p0/test_curd_wlg.out
index 68c7f735096..bdeca8ccf9d 100644
Binary files a/regression-test/data/workload_manager_p0/test_curd_wlg.out and b/regression-test/data/workload_manager_p0/test_curd_wlg.out differ
diff --git a/regression-test/pipeline/cloud_p0/deploy.sh b/regression-test/pipeline/cloud_p0/deploy.sh
index 5a5186b16ee..dc570526b88 100644
--- a/regression-test/pipeline/cloud_p0/deploy.sh
+++ b/regression-test/pipeline/cloud_p0/deploy.sh
@@ -84,6 +84,8 @@ exit_flag=0
fi
if ! check_doris_ready; then exit 1; fi
+ create_normal_workload_group
+
echo "#### 5. set session variables"
if ! reset_doris_session_variables; then exit 1; fi
session_variables_file="${teamcity_build_checkoutDir}/regression-test/pipeline/cloud_p0/conf/session_variables.sql"
diff --git a/regression-test/pipeline/common/doris-utils.sh b/regression-test/pipeline/common/doris-utils.sh
index 8226a27cb19..2834f276952 100644
--- a/regression-test/pipeline/common/doris-utils.sh
+++ b/regression-test/pipeline/common/doris-utils.sh
@@ -460,6 +460,16 @@ show_session_variables() {
fi
}
+create_normal_workload_group() {
+ if [[ ! -d "${DORIS_HOME:-}" ]]; then return 1; fi
+ query_port=$(get_doris_conf_value "${DORIS_HOME}"/fe/conf/fe.conf query_port)
+ if mysql -h127.0.0.1 -P"${query_port}" -uroot -e"create workload group normal for cluster_name0 properties('cpu_share'='1024');"; then
+ return
+ else
+ return 1
+ fi
+}
+
set_session_variable() {
if [[ ! -d "${DORIS_HOME:-}" ]]; then return 1; fi
k="$1"
diff --git a/regression-test/pipeline/vault_p0/deploy.sh b/regression-test/pipeline/vault_p0/deploy.sh
index ec81bc7fb9d..9b9c3ab9005 100644
--- a/regression-test/pipeline/vault_p0/deploy.sh
+++ b/regression-test/pipeline/vault_p0/deploy.sh
@@ -85,6 +85,8 @@ exit_flag=0
# fi
if ! check_doris_ready; then exit 1; fi
+ create_normal_workload_group
+
echo "#### 5. set session variables"
if ! reset_doris_session_variables; then exit 1; fi
session_variables_file="${teamcity_build_checkoutDir}/regression-test/pipeline/vault_p0/conf/session_variables.sql"
diff --git a/regression-test/suites/auth_p0/test_master_slave_consistency_auth.groovy b/regression-test/suites/auth_p0/test_master_slave_consistency_auth.groovy
index 9cc6836412a..cf8896150b0 100644
--- a/regression-test/suites/auth_p0/test_master_slave_consistency_auth.groovy
+++ b/regression-test/suites/auth_p0/test_master_slave_consistency_auth.groovy
@@ -17,6 +17,16 @@
suite ("test_follower_consistent_auth","p0,auth") {
+ def forComputeGroupStr = "";
+
+ //cloud-mode
+ if (isCloudMode()) {
+ def clusters = sql " SHOW CLUSTERS; "
+ assertTrue(!clusters.isEmpty())
+ def validCluster = clusters[0][0]
+ forComputeGroupStr = " for $validCluster "
+ }
+
def get_follower_ip = {
def result = sql """show frontends;"""
logger.info("result:" + result)
@@ -50,8 +60,8 @@ suite ("test_follower_consistent_auth","p0,auth") {
String catalog_name = 'test_follower_consistent_catalog'
try_sql("DROP role ${role}")
sql """CREATE ROLE ${role}"""
- sql """drop WORKLOAD GROUP if exists '${wg}'"""
- sql """CREATE WORKLOAD GROUP "${wg}"
+ sql """drop WORKLOAD GROUP if exists '${wg}' $forComputeGroupStr """
+ sql """CREATE WORKLOAD GROUP "${wg} $forComputeGroupStr "
PROPERTIES (
"cpu_share"="10"
);"""
@@ -376,7 +386,7 @@ suite ("test_follower_consistent_auth","p0,auth") {
}
try_sql("DROP USER ${user}")
- try_sql("drop workload group if exists ${wg};")
+ try_sql("drop workload group if exists ${wg} $forComputeGroupStr;")
}
diff --git a/regression-test/suites/auth_up_down_p0/load.groovy b/regression-test/suites/auth_up_down_p0/load.groovy
index 73429318589..3bb9f1134b6 100644
--- a/regression-test/suites/auth_up_down_p0/load.groovy
+++ b/regression-test/suites/auth_up_down_p0/load.groovy
@@ -17,6 +17,16 @@
suite("test_upgrade_downgrade_prepare_auth","p0,auth,restart_fe") {
+ def forComputeGroupStr = "";
+
+ //cloud-mode
+ if (isCloudMode()) {
+ def clusters = sql " SHOW CLUSTERS; "
+ assertTrue(!clusters.isEmpty())
+ def validCluster = clusters[0][0]
+ forComputeGroupStr = " for $validCluster "
+ }
+
String user1 = 'test_upgrade_downgrade_compatibility_auth_user1'
String user2 = 'test_upgrade_downgrade_compatibility_auth_user2'
String role1 = 'test_upgrade_downgrade_compatibility_auth_role1'
@@ -80,13 +90,13 @@ suite("test_upgrade_downgrade_prepare_auth","p0,auth,restart_fe") {
);
"""
- sql """drop WORKLOAD GROUP if exists '${wg1}'"""
- sql """drop WORKLOAD GROUP if exists '${wg2}'"""
- sql """CREATE WORKLOAD GROUP "${wg1}"
+ sql """drop WORKLOAD GROUP if exists '${wg1}' $forComputeGroupStr """
+ sql """drop WORKLOAD GROUP if exists '${wg2}' $forComputeGroupStr """
+ sql """CREATE WORKLOAD GROUP "${wg1}" $forComputeGroupStr
PROPERTIES (
"cpu_share"="10"
);"""
- sql """CREATE WORKLOAD GROUP "${wg2}"
+ sql """CREATE WORKLOAD GROUP "${wg2}" $forComputeGroupStr
PROPERTIES (
"cpu_share"="10"
);"""
diff --git a/regression-test/suites/load_p0/broker_load/test_s3_load.groovy b/regression-test/suites/load_p0/broker_load/test_s3_load.groovy
index 7e57af5aabb..027ff246202 100644
--- a/regression-test/suites/load_p0/broker_load/test_s3_load.groovy
+++ b/regression-test/suites/load_p0/broker_load/test_s3_load.groovy
@@ -19,7 +19,18 @@ suite("test_s3_load", "load_p0") {
def s3BucketName = getS3BucketName()
def s3Endpoint = getS3Endpoint()
def s3Region = getS3Region()
- sql "create workload group if not exists broker_load_test properties ( 'cpu_share'='1024'); "
+
+ def forComputeGroupStr = "";
+
+ //cloud-mode
+ if (isCloudMode()) {
+ def clusters = sql " SHOW CLUSTERS; "
+ assertTrue(!clusters.isEmpty())
+ def validCluster = clusters[0][0]
+ forComputeGroupStr = " for $validCluster "
+ }
+
+ sql "create workload group if not exists broker_load_test $forComputeGroupStr properties ( 'cpu_share'='1024'); "
sql "set workload_group=broker_load_test;"
diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load.groovy
index 75b8ddce80d..5603b9dd25a 100644
--- a/regression-test/suites/load_p0/routine_load/test_routine_load.groovy
+++ b/regression-test/suites/load_p0/routine_load/test_routine_load.groovy
@@ -22,8 +22,18 @@ import org.apache.kafka.clients.producer.ProducerConfig
suite("test_routine_load","p0") {
- sql "create workload group if not exists create_routine_load_group properties ( 'cpu_share'='123');"
- sql "create workload group if not exists alter_routine_load_group properties ( 'cpu_share'='123');"
+ def forComputeGroupStr = "";
+
+ //cloud-mode
+ if (isCloudMode()) {
+ def clusters = sql " SHOW CLUSTERS; "
+ assertTrue(!clusters.isEmpty())
+ def validCluster = clusters[0][0]
+ forComputeGroupStr = " for $validCluster "
+ }
+
+ sql "create workload group if not exists create_routine_load_group $forComputeGroupStr properties ( 'cpu_share'='123');"
+ sql "create workload group if not exists alter_routine_load_group $forComputeGroupStr properties ( 'cpu_share'='123');"
Thread.sleep(5000) // wait publish workload group to be
def tables = [
diff --git a/regression-test/suites/load_p2/broker_load/test_s3_load_properties.groovy b/regression-test/suites/load_p2/broker_load/test_s3_load_properties.groovy
index d0bef21f226..721fdda27f8 100644
--- a/regression-test/suites/load_p2/broker_load/test_s3_load_properties.groovy
+++ b/regression-test/suites/load_p2/broker_load/test_s3_load_properties.groovy
@@ -19,7 +19,18 @@ suite("test_s3_load_properties", "p2") {
def s3BucketName = getS3BucketName()
def s3Endpoint = getS3Endpoint()
def s3Region = getS3Region()
- sql "create workload group if not exists broker_load_test properties ( 'cpu_share'='1024'); "
+
+ def forComputeGroupStr = "";
+
+ //cloud-mode
+ if (isCloudMode()) {
+ def clusters = sql " SHOW CLUSTERS; "
+ assertTrue(!clusters.isEmpty())
+ def validCluster = clusters[0][0]
+ forComputeGroupStr = " for $validCluster "
+ }
+
+ sql "create workload group if not exists broker_load_test $forComputeGroupStr properties ( 'cpu_share'='1024'); "
sql "set workload_group=broker_load_test;"
diff --git a/regression-test/suites/load_p2/routine_load/test_routine_load.groovy b/regression-test/suites/load_p2/routine_load/test_routine_load.groovy
index 24cb5627f7f..867eac5e8a8 100644
--- a/regression-test/suites/load_p2/routine_load/test_routine_load.groovy
+++ b/regression-test/suites/load_p2/routine_load/test_routine_load.groovy
@@ -22,8 +22,18 @@ import org.apache.kafka.clients.producer.ProducerConfig
suite("test_routine_load_p2","p2,nonConcurrent") {
- sql "create workload group if not exists create_routine_load_group properties ( 'cpu_share'='123');"
- sql "create workload group if not exists alter_routine_load_group properties ( 'cpu_share'='123');"
+ def forComputeGroupStr = "";
+
+ //cloud-mode
+ if (isCloudMode()) {
+ def clusters = sql " SHOW CLUSTERS; "
+ assertTrue(!clusters.isEmpty())
+ def validCluster = clusters[0][0]
+ forComputeGroupStr = " for $validCluster "
+ }
+
+ sql "create workload group if not exists create_routine_load_group $forComputeGroupStr properties ( 'cpu_share'='123');"
+ sql "create workload group if not exists alter_routine_load_group $forComputeGroupStr properties ( 'cpu_share'='123');"
Thread.sleep(5000) // wait publish workload group to be
def tables = [
diff --git a/regression-test/suites/workload_manager_p0/test_curd_wlg.groovy b/regression-test/suites/workload_manager_p0/test_curd_wlg.groovy
index 4dce40ac137..53d800bb4e8 100644
--- a/regression-test/suites/workload_manager_p0/test_curd_wlg.groovy
+++ b/regression-test/suites/workload_manager_p0/test_curd_wlg.groovy
@@ -34,15 +34,17 @@ suite("test_crud_wlg") {
sql "drop table if exists ${table_name2}"
sql "drop table if exists ${table_name3}"
- sql "drop workload group if exists tag1_wg1;"
- sql "drop workload group if exists tag1_wg2;"
- sql "drop workload group if exists tag2_wg1;"
- sql "drop workload group if exists tag1_wg3;"
- sql "drop workload group if exists tag1_mem_wg1;"
- sql "drop workload group if exists tag1_mem_wg2;"
- sql "drop workload group if exists tag1_mem_wg3;"
- sql "drop workload group if exists tag1_mem_wg4;"
- sql "drop workload group if exists bypass_group;"
+ def forComputeGroupStr = "";
+
+ //cloud-mode
+ if (isCloudMode()) {
+ def clusters = sql " SHOW CLUSTERS; "
+ assertTrue(!clusters.isEmpty())
+ def validCluster = clusters[0][0]
+ forComputeGroupStr = " for $validCluster "
+ }
+
+ sql "drop workload group if exists bypass_group $forComputeGroupStr;"
sql """
CREATE TABLE IF NOT EXISTS `${table_name}` (
@@ -84,10 +86,9 @@ suite("test_crud_wlg") {
"""
sql "ADMIN SET FRONTEND CONFIG ('enable_workload_group' = 'true');"
- sql "ADMIN SET FRONTEND CONFIG ('enable_alter_queue_prop_sync' = 'true');"
sql "ADMIN SET FRONTEND CONFIG ('query_queue_update_interval_ms' = '100');"
- sql "create workload group if not exists normal " +
+ sql "create workload group if not exists normal $forComputeGroupStr " +
"properties ( " +
" 'cpu_share'='1024', " +
" 'memory_limit'='50%', " +
@@ -95,44 +96,44 @@ suite("test_crud_wlg") {
");"
// reset normal group property
- sql "alter workload group normal properties ( 'cpu_share'='1024' );"
- sql "alter workload group normal properties ( 'memory_limit'='50%' );"
- sql "alter workload group normal properties ( 'enable_memory_overcommit'='true' );"
- sql "alter workload group normal properties ( 'max_concurrency'='2147483647' );"
- sql "alter workload group normal properties ( 'max_queue_size'='0' );"
- sql "alter workload group normal properties ( 'queue_timeout'='0' );"
- sql "alter workload group normal properties ( 'cpu_hard_limit'='1%' );"
- sql "alter workload group normal properties ( 'scan_thread_num'='-1' );"
- sql "alter workload group normal properties ( 'scan_thread_num'='-1' );"
- sql "alter workload group normal properties ( 'remote_read_bytes_per_second'='-1' );"
- sql "alter workload group normal properties ( 'read_bytes_per_second'='-1' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'cpu_share'='1024' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'memory_limit'='50%' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'enable_memory_overcommit'='true' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'max_concurrency'='2147483647' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'max_queue_size'='0' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'queue_timeout'='0' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'cpu_hard_limit'='1%' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'scan_thread_num'='-1' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'scan_thread_num'='-1' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'remote_read_bytes_per_second'='-1' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'read_bytes_per_second'='-1' );"
sql "set workload_group=normal;"
// test cpu_share
qt_cpu_share """ select count(1) from ${table_name} """
- sql "alter workload group normal properties ( 'scan_thread_num'='16' );"
- sql "alter workload group normal properties ( 'cpu_share'='20' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'scan_thread_num'='16' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'cpu_share'='20' );"
qt_cpu_share_2 """ select count(1) from ${table_name} """
test {
- sql "alter workload group normal properties ( 'cpu_share'='-2' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'cpu_share'='-2' );"
exception "The allowed cpu_share value is -1 or a positive integer"
}
test {
- sql "alter workload group normal properties ( 'scan_thread_num'='0' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'scan_thread_num'='0' );"
exception "The allowed scan_thread_num value is -1 or a positive integer"
}
- sql "drop workload group if exists test_group;"
+ sql "drop workload group if exists test_group $forComputeGroupStr;"
// test create group
- sql "create workload group if not exists test_group " +
+ sql "create workload group if not exists test_group $forComputeGroupStr " +
"properties ( " +
" 'cpu_share'='10', " +
" 'memory_limit'='10%', " +
@@ -140,106 +141,106 @@ suite("test_crud_wlg") {
");"
sql "set workload_group=test_group;"
- qt_show_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,tag,read_bytes_per_second,remote_read_bytes_per_second from information_schema.workload_groups where name in ('normal','test_group') order by name;"
+ qt_show_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,compute_group,read_bytes_per_second,remote_read_bytes_per_second from information_schema.workload_groups where name in ('normal','test_group') order by name;"
// test drop workload group
- sql "create workload group if not exists test_drop_wg properties ('cpu_share'='10')"
- qt_show_del_wg_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,tag from information_schema.workload_groups where name in ('normal','test_group','test_drop_wg') order by name;"
- sql "drop workload group test_drop_wg"
- qt_show_del_wg_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,tag from information_schema.workload_groups where name in ('normal','test_group','test_drop_wg') order by name;"
+ sql "create workload group if not exists test_drop_wg $forComputeGroupStr properties ('cpu_share'='10')"
+ qt_show_del_wg_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,compute_group from information_schema.workload_groups where name in ('normal','test_group','test_drop_wg') order by name;"
+ sql "drop workload group test_drop_wg $forComputeGroupStr"
+ qt_show_del_wg_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,compute_group from information_schema.workload_groups where name in ('normal','test_group','test_drop_wg') order by name;"
// test memory_limit
test {
- sql "alter workload group test_group properties ( 'memory_limit'='100%' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'memory_limit'='100%' );"
- exception "cannot be greater than"
+ exception "can not be greater than"
}
- sql "alter workload group test_group properties ( 'memory_limit'='11%' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'memory_limit'='11%' );"
qt_mem_limit_1 """ select count(1) from ${table_name} """
qt_mem_limit_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from information_schema.workload_groups where name in ('normal','test_group') order by name;"
// test enable_memory_overcommit
test {
- sql "alter workload group test_group properties ( 'enable_memory_overcommit'='1' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'enable_memory_overcommit'='1' );"
exception "must be true or false"
}
- sql "alter workload group test_group properties ( 'enable_memory_overcommit'='true' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'enable_memory_overcommit'='true' );"
qt_mem_overcommit_1 """ select count(1) from ${table_name} """
- sql "alter workload group test_group properties ( 'enable_memory_overcommit'='false' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'enable_memory_overcommit'='false' );"
qt_mem_overcommit_2 """ select count(1) from ${table_name} """
qt_mem_overcommit_3 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from information_schema.workload_groups where name in ('normal','test_group') order by name;"
// test cpu_hard_limit
test {
- sql "alter workload group test_group properties ( 'cpu_hard_limit'='101%' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'cpu_hard_limit'='101%' );"
exception "The allowed cpu_hard_limit value is -1 or a positive integer"
}
- sql "alter workload group test_group properties ( 'cpu_hard_limit'='99%' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'cpu_hard_limit'='99%' );"
test {
- sql "alter workload group normal properties ( 'cpu_hard_limit'='2%' );"
+ sql "alter workload group normal $forComputeGroupStr properties ( 'cpu_hard_limit'='2%' );"
exception "can not be greater than 100%"
}
- sql "alter workload group test_group properties ( 'cpu_hard_limit'='20%' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'cpu_hard_limit'='20%' );"
qt_cpu_hard_limit_1 """ select count(1) from ${table_name} """
qt_cpu_hard_limit_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from information_schema.workload_groups where name in ('normal','test_group') order by name;"
// test query queue
test {
- sql "alter workload group test_group properties ( 'max_concurrency'='-1' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'max_concurrency'='-1' );"
exception "The allowed max_concurrency value is an integer greater than or equal to 0"
}
test {
- sql "alter workload group test_group properties ( 'max_queue_size'='-1' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'max_queue_size'='-1' );"
exception "The allowed max_queue_size value is an integer greater than or equal to 0"
}
test {
- sql "alter workload group test_group properties ( 'queue_timeout'='-1' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'queue_timeout'='-1' );"
exception "The allowed queue_timeout value is an integer greater than or equal to 0"
}
test {
- sql "alter workload group test_group properties('read_bytes_per_second'='0')"
+ sql "alter workload group test_group $forComputeGroupStr properties('read_bytes_per_second'='0')"
exception "The allowed read_bytes_per_second value should be -1 or an positive integer"
}
test {
- sql "alter workload group test_group properties('read_bytes_per_second'='-2')"
+ sql "alter workload group test_group $forComputeGroupStr properties('read_bytes_per_second'='-2')"
exception "The allowed read_bytes_per_second value should be -1 or an positive integer"
}
test {
- sql "alter workload group test_group properties('remote_read_bytes_per_second'='0')"
+ sql "alter workload group test_group $forComputeGroupStr properties('remote_read_bytes_per_second'='0')"
exception "The allowed remote_read_bytes_per_second value should be -1 or an positive integer"
}
test {
- sql "alter workload group test_group properties('remote_read_bytes_per_second'='-2')"
+ sql "alter workload group test_group $forComputeGroupStr properties('remote_read_bytes_per_second'='-2')"
exception "The allowed remote_read_bytes_per_second value should be -1 or an positive integer"
}
- sql "alter workload group test_group properties ( 'max_concurrency'='100' );"
- sql "alter workload group test_group properties('remote_read_bytes_per_second'='104857600')"
- sql "alter workload group test_group properties('read_bytes_per_second'='104857600')"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'max_concurrency'='100' );"
+ sql "alter workload group test_group $forComputeGroupStr properties('remote_read_bytes_per_second'='104857600')"
+ sql "alter workload group test_group $forComputeGroupStr properties('read_bytes_per_second'='104857600')"
qt_queue_1 """ select count(1) from ${table_name} """
qt_show_queue "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,read_bytes_per_second,remote_read_bytes_per_second from information_schema.workload_groups where name in ('normal','test_group') order by name;"
// test create group failed
// failed for cpu_share
test {
- sql "create workload group if not exists test_group2 " +
+ sql "create workload group if not exists test_group2 $forComputeGroupStr " +
"properties ( " +
" 'cpu_share'='-2', " +
" 'memory_limit'='1%', " +
@@ -251,31 +252,31 @@ suite("test_crud_wlg") {
// failed for mem_limit
test {
- sql "create workload group if not exists test_group2 " +
+ sql "create workload group if not exists test_group2 $forComputeGroupStr " +
"properties ( " +
" 'cpu_share'='10', " +
" 'memory_limit'='200%', " +
" 'enable_memory_overcommit'='true' " +
");"
- exception "cannot be greater than"
+ exception "can not be greater than"
}
test {
- sql "create workload group if not exists test_group2 " +
+ sql "create workload group if not exists test_group2 $forComputeGroupStr " +
"properties ( " +
" 'cpu_share'='10', " +
" 'memory_limit'='99%', " +
" 'enable_memory_overcommit'='true' " +
");"
- exception "cannot be greater than"
+ exception "can not be greater than"
}
// failed for mem_overcommit
test {
- sql "create workload group if not exists test_group2 " +
+ sql "create workload group if not exists test_group2 $forComputeGroupStr " +
"properties ( " +
" 'cpu_share'='10', " +
" 'memory_limit'='3%', " +
@@ -288,7 +289,7 @@ suite("test_crud_wlg") {
// failed for cpu_hard_limit
test {
- sql "create workload group if not exists test_group2 " +
+ sql "create workload group if not exists test_group2 $forComputeGroupStr " +
"properties ( " +
" 'cpu_share'='10', " +
" 'memory_limit'='3%', " +
@@ -300,7 +301,7 @@ suite("test_crud_wlg") {
}
test {
- sql "create workload group if not exists test_group2 " +
+ sql "create workload group if not exists test_group2 $forComputeGroupStr " +
"properties ( " +
" 'cpu_share'='10', " +
" 'memory_limit'='3%', " +
@@ -348,8 +349,8 @@ suite("test_crud_wlg") {
// test query queue limit
sql "set workload_group=test_group;"
- sql "alter workload group test_group properties ( 'max_concurrency'='0' );"
- sql "alter workload group test_group properties ( 'max_queue_size'='0' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'max_concurrency'='0' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'max_queue_size'='0' );"
Thread.sleep(10000)
test {
sql "select /*+SET_VAR(workload_group=test_group)*/ * from ${table_name};"
@@ -371,8 +372,8 @@ suite("test_crud_wlg") {
exception "query waiting queue is full"
}
- sql "alter workload group test_group properties ( 'max_queue_size'='1' );"
- sql "alter workload group test_group properties ( 'queue_timeout'='500' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'max_queue_size'='1' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'queue_timeout'='500' );"
Thread.sleep(10000)
test {
sql "select /*+SET_VAR(parallel_pipeline_task_num=1)*/ * from ${table_name};"
@@ -381,8 +382,8 @@ suite("test_crud_wlg") {
}
// test query queue running query/waiting num
- sql "drop workload group if exists test_query_num_wg;"
- sql "create workload group if not exists test_query_num_wg properties ('max_concurrency'='1');"
+ sql "drop workload group if exists test_query_num_wg $forComputeGroupStr;"
+ sql "create workload group if not exists test_query_num_wg $forComputeGroupStr properties ('max_concurrency'='1');"
sql "set workload_group=test_query_num_wg;"
sql """insert into ${table_name} values
@@ -417,169 +418,90 @@ suite("test_crud_wlg") {
}
assertTrue(is_checked)
- sql "drop workload group test_query_num_wg;"
+ sql "drop workload group test_query_num_wg $forComputeGroupStr;"
sql "set workload_group=normal;"
- sql "alter workload group test_group properties ( 'max_concurrency'='10' );"
+ sql "alter workload group test_group $forComputeGroupStr properties ( 'max_concurrency'='10' );"
Thread.sleep(10000)
sql "select /*+SET_VAR(parallel_pipeline_task_num=1)*/ * from ${table_name};"
// test workload spill property
// 1 create group
test {
- sql "create workload group if not exists spill_group_test_failed properties ( 'memory_low_watermark'='96%');"
+ sql "create workload group if not exists spill_group_test_failed $forComputeGroupStr properties ( 'memory_low_watermark'='96%');"
exception "should bigger than memory_low_watermark"
}
- sql "create workload group if not exists spill_group_test properties ( 'memory_low_watermark'='10%','memory_high_watermark'='10%');"
+ sql "create workload group if not exists spill_group_test $forComputeGroupStr properties ( 'memory_low_watermark'='10%','memory_high_watermark'='10%');"
qt_show_spill_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,memory_low_watermark,memory_high_watermark from information_schema.workload_groups where name in ('spill_group_test');"
test {
- sql "create workload group if not exists spill_group_test properties ( 'memory_low_watermark'='20%','memory_high_watermark'='10%');"
+ sql "create workload group if not exists spill_group_test $forComputeGroupStr properties ( 'memory_low_watermark'='20%','memory_high_watermark'='10%');"
exception "should bigger than memory_low_watermark"
}
// 2 alter low
- sql "alter workload group spill_group_test properties ( 'memory_low_watermark'='5%' );"
+ sql "alter workload group spill_group_test $forComputeGroupStr properties ( 'memory_low_watermark'='5%' );"
qt_show_spill_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,memory_low_watermark,memory_high_watermark from information_schema.workload_groups where name in ('spill_group_test');"
test {
- sql "alter workload group spill_group_test properties ( 'memory_low_watermark'='20%' );"
+ sql "alter workload group spill_group_test $forComputeGroupStr properties ( 'memory_low_watermark'='20%' );"
exception "should bigger than memory_low_watermark"
}
test {
- sql "alter workload group spill_group_test properties ( 'memory_low_watermark'='0%' );"
+ sql "alter workload group spill_group_test $forComputeGroupStr properties ( 'memory_low_watermark'='0%' );"
exception "value is an integer value between 1 and 100"
}
test {
- sql "alter workload group spill_group_test properties ( 'memory_low_watermark'='101%' );"
+ sql "alter workload group spill_group_test $forComputeGroupStr properties ( 'memory_low_watermark'='101%' );"
exception "value is an integer value between 1 and 100"
}
test {
- sql "create workload group if not exists spill_group_test2 properties ( 'memory_low_watermark'='0%')"
+ sql "create workload group if not exists spill_group_test2 $forComputeGroupStr properties ( 'memory_low_watermark'='0%')"
exception "value is an integer value between 1 and 100"
}
test {
- sql "create workload group if not exists spill_group_test2 properties ( 'memory_low_watermark'='101%')"
+ sql "create workload group if not exists spill_group_test2 $forComputeGroupStr properties ( 'memory_low_watermark'='101%')"
exception "value is an integer value between 1 and 100"
}
// 3 alter high
- sql "alter workload group spill_group_test properties ( 'memory_high_watermark'='40%' );"
+ sql "alter workload group spill_group_test $forComputeGroupStr properties ( 'memory_high_watermark'='40%' );"
qt_show_spill_3 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,memory_low_watermark,memory_high_watermark from information_schema.workload_groups where name in ('spill_group_test');"
test {
- sql "alter workload group spill_group_test properties ( 'memory_high_watermark'='1%' );"
+ sql "alter workload group spill_group_test $forComputeGroupStr properties ( 'memory_high_watermark'='1%' );"
exception "should bigger than memory_low_watermark"
}
test {
- sql "alter workload group spill_group_test properties ( 'memory_high_watermark'='0%' );"
+ sql "alter workload group spill_group_test $forComputeGroupStr properties ( 'memory_high_watermark'='0%' );"
exception "value is an integer value between 1 and 100"
}
test {
- sql "alter workload group spill_group_test properties ( 'memory_high_watermark'='101%' );"
+ sql "alter workload group spill_group_test $forComputeGroupStr properties ( 'memory_high_watermark'='101%' );"
exception "value is an integer value between 1 and 100"
}
test {
- sql "create workload group if not exists spill_group_test2 properties ( 'memory_high_watermark'='0%')"
+ sql "create workload group if not exists spill_group_test2 $forComputeGroupStr properties ( 'memory_high_watermark'='0%')"
exception "value is an integer value between 1 and 100"
}
test {
- sql "create workload group if not exists spill_group_test2 properties ( 'memory_high_watermark'='101%')"
+ sql "create workload group if not exists spill_group_test2 $forComputeGroupStr properties ( 'memory_high_watermark'='101%')"
exception "value is an integer value between 1 and 100"
}
- sql "drop workload group test_group;"
- sql "drop workload group spill_group_test;"
-
+ sql "drop workload group test_group $forComputeGroupStr;"
+ sql "drop workload group spill_group_test $forComputeGroupStr;"
- // test workload group's tag property, cpu_hard_limit
- test {
- sql "create workload group tag_test properties('tag'=' a, b , c ');"
- exception "tag format is illegal"
- }
-
- test {
- sql "create workload group if not exists tag1_wg1 properties ( 'cpu_hard_limit'='101%', 'tag'='tag1')"
- exception "a positive integer between 1 and 100"
- }
-
- test {
- sql "create workload group if not exists tag1_wg1 properties ( 'cpu_hard_limit'='-2%', 'tag'='tag1')"
- exception "a positive integer between 1 and 100"
- }
-
- test {
- sql "create workload group if not exists tag1_wg1 properties ( 'cpu_hard_limit'='-1%', 'tag'='tag1')"
- exception "a positive integer between 1 and 100"
- }
-
- sql "create workload group if not exists tag1_wg1 properties ( 'cpu_hard_limit'='10%', 'tag'='tag1');"
-
- test {
- sql "create workload group if not exists tag1_wg2 properties ( 'cpu_hard_limit'='91%', 'tag'='tag1');"
- exception "can not be greater than 100%"
- }
-
- sql "create workload group if not exists tag1_wg2 properties ( 'cpu_hard_limit'='10%', 'tag'='tag1');"
-
- sql "create workload group if not exists tag2_wg1 properties ( 'cpu_hard_limit'='91%', 'tag'='tag2');"
-
- test {
- sql "alter workload group tag2_wg1 properties ( 'tag'='tag1' );"
- exception "can not be greater than 100% "
- }
-
- sql "alter workload group tag2_wg1 properties ( 'cpu_hard_limit'='10%' );"
- sql "alter workload group tag2_wg1 properties ( 'tag'='tag1' );"
-
- test {
- sql "create workload group if not exists tag1_wg3 properties ( 'cpu_hard_limit'='80%', 'tag'='tag1');"
- exception "can not be greater than 100% "
- }
-
- sql "drop workload group tag2_wg1;"
- sql "create workload group if not exists tag1_wg3 properties ( 'cpu_hard_limit'='80%', 'tag'='tag1');"
-
- // test workload group's tag property, memory_limit
- sql "create workload group if not exists tag1_mem_wg1 properties ( 'memory_limit'='50%', 'tag'='mem_tag1');"
-
- test {
- sql "create workload group if not exists tag1_mem_wg2 properties ( 'memory_limit'='60%', 'tag'='mem_tag1');"
- exception "cannot be greater than 100.0%"
- }
-
- sql "create workload group if not exists tag1_mem_wg2 properties ('memory_limit'='49%', 'tag'='mem_tag1');"
-
- sql "create workload group if not exists tag1_mem_wg3 properties ( 'memory_limit'='2%');"
-
- test {
- sql "alter workload group tag1_mem_wg3 properties ( 'tag'='mem_tag1' );"
- exception "cannot be greater than 100.0%"
- }
-
- sql "alter workload group tag1_mem_wg3 properties ( 'memory_limit'='1%' );"
-
- sql "alter workload group tag1_mem_wg3 properties ( 'tag'='mem_tag1' );"
-
- sql "create workload group tag1_mem_wg4 properties('memory_limit'='-1','tag'='mem_tag1');"
-
- test {
- sql "alter workload group tag1_mem_wg4 properties ( 'memory_limit'='1%' );"
- exception "cannot be greater than 100.0%"
- }
-
-
- qt_show_wg_tag "select name,MEMORY_LIMIT,CPU_HARD_LIMIT,TAG from information_schema.workload_groups where name in('tag1_wg1','tag1_wg2','tag2_wg1','tag1_wg3','tag1_mem_wg1','tag1_mem_wg2','tag1_mem_wg3') order by tag,name;"
// test bypass
- sql "create workload group if not exists bypass_group properties ( 'max_concurrency'='0','max_queue_size'='0','queue_timeout'='0');"
+ sql "create workload group if not exists bypass_group $forComputeGroupStr properties ( 'max_concurrency'='0','max_queue_size'='0','queue_timeout'='0');"
sql "set workload_group=bypass_group;"
test {
sql "select count(1) from ${table_name};"
@@ -590,87 +512,79 @@ suite("test_crud_wlg") {
sql "select count(1) from information_schema.active_queries;"
// test set remote scan pool
- sql "drop workload group if exists test_remote_scan_wg;"
+ sql "drop workload group if exists test_remote_scan_wg $forComputeGroupStr;"
test {
- sql "create workload group test_remote_scan_wg properties('min_remote_scan_thread_num'='123');"
+ sql "create workload group test_remote_scan_wg $forComputeGroupStr properties('min_remote_scan_thread_num'='123');"
exception "must be specified simultaneously"
}
test {
- sql "create workload group test_remote_scan_wg properties('max_remote_scan_thread_num'='123');"
+ sql "create workload group test_remote_scan_wg $forComputeGroupStr properties('max_remote_scan_thread_num'='123');"
exception "must be specified simultaneously"
}
test {
- sql "create workload group test_remote_scan_wg properties('max_remote_scan_thread_num'='10', 'min_remote_scan_thread_num'='123');"
+ sql "create workload group test_remote_scan_wg $forComputeGroupStr properties('max_remote_scan_thread_num'='10', 'min_remote_scan_thread_num'='123');"
exception "must bigger or equal "
}
- sql "create workload group test_remote_scan_wg properties('max_remote_scan_thread_num'='20', 'min_remote_scan_thread_num'='10');"
+ sql "create workload group test_remote_scan_wg $forComputeGroupStr properties('max_remote_scan_thread_num'='20', 'min_remote_scan_thread_num'='10');"
qt_select_remote_scan_num "select MAX_REMOTE_SCAN_THREAD_NUM,MIN_REMOTE_SCAN_THREAD_NUM from information_schema.workload_groups where name='test_remote_scan_wg';"
- sql "alter workload group test_remote_scan_wg properties('max_remote_scan_thread_num'='21')"
+ sql "alter workload group test_remote_scan_wg $forComputeGroupStr properties('max_remote_scan_thread_num'='21')"
qt_select_remote_scan_num_2 "select MAX_REMOTE_SCAN_THREAD_NUM,MIN_REMOTE_SCAN_THREAD_NUM from information_schema.workload_groups where name='test_remote_scan_wg';"
test {
- sql "alter workload group test_remote_scan_wg properties('max_remote_scan_thread_num'='5')"
+ sql "alter workload group test_remote_scan_wg $forComputeGroupStr properties('max_remote_scan_thread_num'='5')"
exception "must bigger or equal"
}
- sql "alter workload group test_remote_scan_wg properties('min_remote_scan_thread_num'='2')"
+ sql "alter workload group test_remote_scan_wg $forComputeGroupStr properties('min_remote_scan_thread_num'='2')"
qt_select_remote_scan_num_3 "select MAX_REMOTE_SCAN_THREAD_NUM,MIN_REMOTE_SCAN_THREAD_NUM from information_schema.workload_groups where name='test_remote_scan_wg';"
test {
- sql "alter workload group test_remote_scan_wg properties('min_remote_scan_thread_num'='30')"
+ sql "alter workload group test_remote_scan_wg $forComputeGroupStr properties('min_remote_scan_thread_num'='30')"
exception "must bigger or equal"
}
- sql "alter workload group test_remote_scan_wg properties('max_remote_scan_thread_num'='40', 'min_remote_scan_thread_num'='20')"
+ sql "alter workload group test_remote_scan_wg $forComputeGroupStr properties('max_remote_scan_thread_num'='40', 'min_remote_scan_thread_num'='20')"
qt_select_remote_scan_num_4 "select MAX_REMOTE_SCAN_THREAD_NUM,MIN_REMOTE_SCAN_THREAD_NUM from information_schema.workload_groups where name='test_remote_scan_wg';"
- sql "alter workload group test_remote_scan_wg properties('max_remote_scan_thread_num'='10', 'min_remote_scan_thread_num'='5')"
+ sql "alter workload group test_remote_scan_wg $forComputeGroupStr properties('max_remote_scan_thread_num'='10', 'min_remote_scan_thread_num'='5')"
qt_select_remote_scan_num_5 "select MAX_REMOTE_SCAN_THREAD_NUM,MIN_REMOTE_SCAN_THREAD_NUM from information_schema.workload_groups where name='test_remote_scan_wg';"
- sql "alter workload group test_remote_scan_wg properties('max_remote_scan_thread_num'='3', 'min_remote_scan_thread_num'='3')"
+ sql "alter workload group test_remote_scan_wg $forComputeGroupStr properties('max_remote_scan_thread_num'='3', 'min_remote_scan_thread_num'='3')"
qt_select_remote_scan_num_6 "select MAX_REMOTE_SCAN_THREAD_NUM,MIN_REMOTE_SCAN_THREAD_NUM from information_schema.workload_groups where name='test_remote_scan_wg';"
- sql "drop workload group test_remote_scan_wg;"
- sql "create workload group test_remote_scan_wg properties('cpu_share'='1024');"
+ sql "drop workload group test_remote_scan_wg $forComputeGroupStr;"
+ sql "create workload group test_remote_scan_wg $forComputeGroupStr properties('cpu_share'='1024');"
test {
- sql "alter workload group test_remote_scan_wg properties('min_remote_scan_thread_num'='30')"
+ sql "alter workload group test_remote_scan_wg $forComputeGroupStr properties('min_remote_scan_thread_num'='30')"
exception "must be specified simultaneously"
}
test {
- sql "alter workload group test_remote_scan_wg properties('max_remote_scan_thread_num'='30')"
+ sql "alter workload group test_remote_scan_wg $forComputeGroupStr properties('max_remote_scan_thread_num'='30')"
exception "must be specified simultaneously"
}
- sql "alter workload group test_remote_scan_wg properties('max_remote_scan_thread_num'='10', 'min_remote_scan_thread_num'='5')"
+ sql "alter workload group test_remote_scan_wg $forComputeGroupStr properties('max_remote_scan_thread_num'='10', 'min_remote_scan_thread_num'='5')"
qt_select_remote_scan_num_7 "select MAX_REMOTE_SCAN_THREAD_NUM,MIN_REMOTE_SCAN_THREAD_NUM from information_schema.workload_groups where name='test_remote_scan_wg';"
- sql "alter workload group test_remote_scan_wg properties('max_remote_scan_thread_num'='-1', 'min_remote_scan_thread_num'='-1')"
+ sql "alter workload group test_remote_scan_wg $forComputeGroupStr properties('max_remote_scan_thread_num'='-1', 'min_remote_scan_thread_num'='-1')"
qt_select_remote_scan_num_8 "select MAX_REMOTE_SCAN_THREAD_NUM,MIN_REMOTE_SCAN_THREAD_NUM from information_schema.workload_groups where name='test_remote_scan_wg';"
- sql "drop workload group test_remote_scan_wg"
-
- sql "drop workload group tag1_wg1;"
- sql "drop workload group tag1_wg2;"
- sql "drop workload group if exists tag2_wg1;"
- sql "drop workload group tag1_wg3;"
- sql "drop workload group tag1_mem_wg1;"
- sql "drop workload group tag1_mem_wg2;"
- sql "drop workload group tag1_mem_wg3;"
- sql "drop workload group tag1_mem_wg4;"
- sql "drop workload group bypass_group;"
+ sql "drop workload group test_remote_scan_wg $forComputeGroupStr"
+
+ sql "drop workload group bypass_group $forComputeGroupStr;"
// test workload group privilege table
sql "set workload_group=normal;"
sql "drop user if exists test_wg_priv_user1"
sql "drop user if exists test_wg_priv_user2"
sql "drop role if exists test_wg_priv_role1"
- sql "drop workload group if exists test_wg_priv_g1;"
+ sql "drop workload group if exists test_wg_priv_g1 $forComputeGroupStr;"
// 1 test grant user
- sql "create workload group test_wg_priv_g1 properties('cpu_share'='1024')"
+ sql "create workload group test_wg_priv_g1 $forComputeGroupStr properties('cpu_share'='1024')"
sql "create user test_wg_priv_user1"
qt_select_wgp_1 "select GRANTEE,WORKLOAD_GROUP_NAME,PRIVILEGE_TYPE,IS_GRANTABLE from information_schema.workload_group_privileges where grantee like '%test_wg_priv%' order by GRANTEE,WORKLOAD_GROUP_NAME,PRIVILEGE_TYPE,IS_GRANTABLE; "
@@ -725,15 +639,15 @@ suite("test_crud_wlg") {
sql "drop user test_wg_priv_user2"
sql "drop role test_wg_priv_role1"
qt_select_wgp_12 "select GRANTEE,WORKLOAD_GROUP_NAME,PRIVILEGE_TYPE,IS_GRANTABLE from information_schema.workload_group_privileges where grantee like '%test_wg_priv%' order by GRANTEE,WORKLOAD_GROUP_NAME,PRIVILEGE_TYPE,IS_GRANTABLE; "
- sql "drop workload group test_wg_priv_g1"
+ sql "drop workload group test_wg_priv_g1 $forComputeGroupStr"
// test default value
- sql "drop workload group if exists default_val_wg"
- sql "create workload group default_val_wg properties('enable_memory_overcommit'='true');"
- qt_select_default_val_wg_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,max_remote_scan_thread_num,min_remote_scan_thread_num,memory_low_watermark,memory_high_watermark,tag,read_bytes_per_second,remote_read_bytes_per_second from information_schema.workload_groups where name = 'default_val_wg'"
+ sql "drop workload group if exists default_val_wg $forComputeGroupStr"
+ sql "create workload group default_val_wg $forComputeGroupStr properties('enable_memory_overcommit'='true');"
+ qt_select_default_val_wg_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,max_remote_scan_thread_num,min_remote_scan_thread_num,memory_low_watermark,memory_high_watermark,compute_group,read_bytes_per_second,remote_read_bytes_per_second from information_schema.workload_groups where name = 'default_val_wg'"
sql """
- alter workload group default_val_wg properties(
+ alter workload group default_val_wg $forComputeGroupStr properties(
'cpu_share'='1024',
'memory_limit'='1%',
'enable_memory_overcommit'='true',
@@ -744,15 +658,14 @@ suite("test_crud_wlg") {
'scan_thread_num'='1',
'max_remote_scan_thread_num'='12',
'min_remote_scan_thread_num'='10',
- 'tag'='abc',
'read_bytes_per_second'='123',
'remote_read_bytes_per_second'='10');
"""
- qt_select_default_val_wg_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,max_remote_scan_thread_num,min_remote_scan_thread_num,memory_low_watermark,memory_high_watermark,tag,read_bytes_per_second,remote_read_bytes_per_second from information_schema.workload_groups where name = 'default_val_wg'"
+ qt_select_default_val_wg_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,max_remote_scan_thread_num,min_remote_scan_thread_num,memory_low_watermark,memory_high_watermark,compute_group,read_bytes_per_second,remote_read_bytes_per_second from information_schema.workload_groups where name = 'default_val_wg'"
sql """
- alter workload group default_val_wg properties(
+ alter workload group default_val_wg $forComputeGroupStr properties(
'cpu_share'='-1',
'memory_limit'='-1',
'enable_memory_overcommit'='true',
@@ -763,60 +676,59 @@ suite("test_crud_wlg") {
'scan_thread_num'='-1',
'max_remote_scan_thread_num'='-1',
'min_remote_scan_thread_num'='-1',
- 'tag'='',
'read_bytes_per_second'='-1',
'remote_read_bytes_per_second'='-1'
);
"""
- qt_select_default_val_wg_3 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,max_remote_scan_thread_num,min_remote_scan_thread_num,memory_low_watermark,memory_high_watermark,tag,read_bytes_per_second,remote_read_bytes_per_second from information_schema.workload_groups where name = 'default_val_wg'"
-
- sql "drop workload group if exists default_val_wg"
+ qt_select_default_val_wg_3 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,max_remote_scan_thread_num,min_remote_scan_thread_num,memory_low_watermark,memory_high_watermark,compute_group,read_bytes_per_second,remote_read_bytes_per_second from information_schema.workload_groups where name = 'default_val_wg'"
- for (int i = 0; i < 20; i++) {
- // 1. SHOW BACKENDS get be ip and http port
- Map<String, String> backendId_to_backendIP = new HashMap<>();
- Map<String, String> backendId_to_backendHttpPort = new HashMap<>();
- getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort);
- // Print above maps in logger.
- logger.info("backendId_to_backendIP: " + backendId_to_backendIP);
- logger.info("backendId_to_backendHttpPort: " + backendId_to_backendHttpPort);
+ sql "drop workload group if exists default_val_wg $forComputeGroupStr"
- // 2. CREATE WORKLOAD GROUP
- sql "drop workload group if exists test_wg_metrics;"
- sql "create workload group if not exists test_wg_metrics " +
- "properties ( " +
- " 'cpu_share'='10', " +
- " 'memory_limit'='10%', " +
- " 'enable_memory_overcommit'='true' " +
- ");"
- sql "set workload_group=test_wg_metrics;"
- wg = sql("select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,tag,read_bytes_per_second,remote_read_bytes_per_second from information_schema.workload_groups where name = 'test_wg_metrics' order by name;");
- logger.info("wg: " + wg);
-
- // 3. EXECUTE A QUERY SO THAT THE WORKLOAD GROUP IS USED
- sql "select count(*) from numbers(\"number\"=\"100\");"
-
- // curl backend http port to get metrics
- // get first backendId
- backendId = backendId_to_backendIP.keySet().iterator().next();
- backendIP = backendId_to_backendIP.get(backendId);
- backendHttpPort = backendId_to_backendHttpPort.get(backendId);
- logger.info("backendId: " + backendId + ", backendIP: " + backendIP + ", backendHttpPort: " + backendHttpPort);
-
- // Create a for loop to get metrics 5 times
- for (int j = 0; j < 5; j++) {
- String metrics = getMetrics(backendIP, backendHttpPort);
- String filteredMetrics = metrics.split('\n').findAll { line ->
- line.startsWith('doris_be_thread_pool') && line.contains('workload_group="test_wg_metrics"') && line.contains('thread_pool_name="Scan_test_wg_metrics"')
- }.join('\n')
- // Filter metrics with name test_wg_metrics
- logger.info("filteredMetrics: " + filteredMetrics);
- List<String> lines = filteredMetrics.split('\n').findAll { it.trim() }
- assert lines.size() == 5
- }
-
- sql "drop workload group if exists test_wg_metrics;"
- }
- sql "drop workload group if exists test_wg_metrics;"
+// for (int i = 0; i < 20; i++) {
+// // 1. SHOW BACKENDS get be ip and http port
+// Map<String, String> backendId_to_backendIP = new HashMap<>();
+// Map<String, String> backendId_to_backendHttpPort = new HashMap<>();
+// getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort);
+// // Print above maps in logger.
+// logger.info("backendId_to_backendIP: " + backendId_to_backendIP);
+// logger.info("backendId_to_backendHttpPort: " + backendId_to_backendHttpPort);
+//
+// // 2. CREATE WORKLOAD GROUP
+// sql "drop workload group if exists test_wg_metrics;"
+// sql "create workload group if not exists $forComputeGroupStr test_wg_metrics " +
+// "properties ( " +
+// " 'cpu_share'='10', " +
+// " 'memory_limit'='10%', " +
+// " 'enable_memory_overcommit'='true' " +
+// ");"
+// sql "set workload_group=test_wg_metrics;"
+// def wg = sql "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num,compute_group,read_bytes_per_second,remote_read_bytes_per_second from information_schema.workload_groups where name = 'test_wg_metrics' order by name;"
+// logger.info("wg: $wg");
+//
+// // 3. EXECUTE A QUERY SO THAT THE WORKLOAD GROUP IS USED
+// sql "select count(*) from numbers(\"number\"=\"100\");"
+//
+// // curl backend http port to get metrics
+// // get first backendId
+// def backendId = backendId_to_backendIP.keySet().iterator().next();
+// def backendIP = backendId_to_backendIP.get(backendId);
+// def backendHttpPort = backendId_to_backendHttpPort.get(backendId);
+// logger.info("backendId: " + backendId + ", backendIP: " + backendIP + ", backendHttpPort: " + backendHttpPort);
+//
+// // Create a for loop to get metrics 5 times
+// for (int j = 0; j < 5; j++) {
+// String metrics = getMetrics(backendIP, backendHttpPort);
+// String filteredMetrics = metrics.split('\n').findAll { line ->
+// line.startsWith('doris_be_thread_pool') && line.contains('workload_group="test_wg_metrics"') && line.contains('thread_pool_name="Scan_test_wg_metrics"')
+// }.join('\n')
+// // Filter metrics with name test_wg_metrics
+// logger.info("filteredMetrics: " + filteredMetrics);
+// List<String> lines = filteredMetrics.split('\n').findAll { it.trim() }
+// assert lines.size() == 5
+// }
+//
+// sql "drop workload group if exists test_wg_metrics;"
+// }
+// sql "drop workload group if exists test_wg_metrics;"
}
diff --git a/regression-test/suites/workload_manager_p0/test_workload_sched_policy.groovy b/regression-test/suites/workload_manager_p0/test_workload_sched_policy.groovy
index ca18c5a17e0..513c05cc8ac 100644
--- a/regression-test/suites/workload_manager_p0/test_workload_sched_policy.groovy
+++ b/regression-test/suites/workload_manager_p0/test_workload_sched_policy.groovy
@@ -17,6 +17,18 @@
suite("test_workload_sched_policy") {
+ def forComputeGroupStr = ""
+ def currentCgName = ""
+
+ //cloud-mode
+ if (isCloudMode()) {
+ def clusters = sql " SHOW CLUSTERS; "
+ assertTrue(!clusters.isEmpty())
+ def validCluster = clusters[0][0]
+ currentCgName = "${validCluster}.";
+ forComputeGroupStr = " for $validCluster "
+ }
+
sql "drop workload policy if exists test_cancel_policy;"
sql "drop workload policy if exists set_action_policy;"
sql "drop workload policy if exists fe_policy;"
@@ -146,10 +158,10 @@ suite("test_workload_sched_policy") {
sql """drop user if exists test_workload_sched_user"""
sql """create user test_workload_sched_user identified by '12345'"""
sql """grant ADMIN_PRIV on *.*.* to test_workload_sched_user"""
- sql "drop workload group if exists test_set_session_wg;"
- sql "drop workload group if exists test_set_session_wg2;"
- sql "create workload group test_set_session_wg properties('cpu_share'='1024');"
- sql "create workload group test_set_session_wg2 properties('cpu_share'='1024');"
+ sql "drop workload group if exists test_set_session_wg $forComputeGroupStr;"
+ sql "drop workload group if exists test_set_session_wg2 $forComputeGroupStr;"
+ sql "create workload group test_set_session_wg $forComputeGroupStr properties('cpu_share'='1024');"
+ sql "create workload group test_set_session_wg2 $forComputeGroupStr properties('cpu_share'='1024');"
sql "drop workload policy if exists test_set_var_policy;"
sql "drop workload policy if exists test_set_var_policy2;"
@@ -183,8 +195,8 @@ suite("test_workload_sched_policy") {
}
assertEquals("workload_group", result3[0][0])
assertEquals("test_set_session_wg", result3[0][1])
- sql "drop workload group if exists test_set_session_wg;"
- sql "drop workload group if exists test_set_session_wg2;"
+ sql "drop workload group if exists test_set_session_wg $forComputeGroupStr;"
+ sql "drop workload group if exists test_set_session_wg2 $forComputeGroupStr;"
sql "drop workload policy if exists test_set_var_policy;"
sql "drop workload policy if exists test_set_var_policy2;"
@@ -193,32 +205,32 @@ suite("test_workload_sched_policy") {
sql "drop workload policy if exists test_cancel_query_policy"
sql "drop workload policy if exists test_cancel_query_policy2"
sql "drop workload policy if exists test_set_session"
- sql "drop workload group if exists policy_group;"
+ sql "drop workload group if exists policy_group $forComputeGroupStr;"
sql "CREATE USER 'test_policy_user'@'%' IDENTIFIED BY '12345';"
sql """grant SELECT_PRIV on *.*.* to test_policy_user;"""
- sql "create workload group if not exists policy_group properties ('cpu_share'='1024');"
- sql "create workload group if not exists policy_group2 properties ('cpu_share'='1024');"
+ sql "create workload group if not exists policy_group $forComputeGroupStr properties ('cpu_share'='1024');"
+ sql "create workload group if not exists policy_group2 $forComputeGroupStr properties ('cpu_share'='1024');"
sql "GRANT USAGE_PRIV ON WORKLOAD GROUP 'policy_group' TO 'test_policy_user'@'%';"
sql "GRANT USAGE_PRIV ON WORKLOAD GROUP 'policy_group2' TO 'test_policy_user'@'%';"
- sql "create workload policy test_cancel_query_policy conditions(query_time > 1000) actions(cancel_query) properties('workload_group'='policy_group')"
- sql "create workload policy test_cancel_query_policy2 conditions(query_time > 0, be_scan_rows>1) actions(cancel_query) properties('workload_group'='policy_group')"
+ sql "create workload policy test_cancel_query_policy conditions(query_time > 1000) actions(cancel_query) properties('workload_group'='${currentCgName}policy_group')"
+ sql "create workload policy test_cancel_query_policy2 conditions(query_time > 0, be_scan_rows>1) actions(cancel_query) properties('workload_group'='${currentCgName}policy_group')"
sql "create workload policy test_set_session conditions(username='test_policy_user') actions(set_session_variable 'parallel_pipeline_task_num=1')"
test {
- sql "drop workload group policy_group;"
+ sql "drop workload group policy_group $forComputeGroupStr;"
exception "because it has related policy"
}
test {
- sql "alter workload policy test_cancel_query_policy properties('workload_group'='invalid_gorup');"
- exception "unknown workload group"
+ sql "alter workload policy test_cancel_query_policy properties('workload_group'='${currentCgName}invalid_gorup');"
+ exception "Can not find workload group"
}
// test alter policy property
sql "drop user if exists test_alter_policy_user"
sql "CREATE USER 'test_alter_policy_user'@'%' IDENTIFIED BY '12345';"
sql "drop workload policy if exists test_alter_policy;"
- sql "create workload policy test_alter_policy conditions(username='test_alter_policy_user') actions(set_session_variable 'parallel_pipeline_task_num=0') properties('workload_group'='normal');"
+ sql "create workload policy test_alter_policy conditions(username='test_alter_policy_user') actions(set_session_variable 'parallel_pipeline_task_num=0') properties('workload_group'='${currentCgName}normal');"
qt_select_alter_1 "select name,condition,action,PRIORITY,ENABLED,VERSION,WORKLOAD_GROUP from information_schema.workload_policy where name='test_alter_policy'"
sql "alter workload policy test_alter_policy properties('workload_group'='');"
@@ -230,7 +242,7 @@ suite("test_workload_sched_policy") {
sql "alter workload policy test_alter_policy properties('priority'='9');"
qt_select_alter_4 "select name,condition,action,PRIORITY,ENABLED,VERSION,WORKLOAD_GROUP from information_schema.workload_policy where name='test_alter_policy'"
- sql "alter workload policy test_alter_policy properties('workload_group'='normal');"
+ sql "alter workload policy test_alter_policy properties('workload_group'='${currentCgName}normal');"
qt_select_alter_5 "select name,condition,action,PRIORITY,ENABLED,VERSION,WORKLOAD_GROUP from information_schema.workload_policy where name='test_alter_policy'"
sql "drop user test_alter_policy_user"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org