database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[improvement](executor) Add tvf and regression test for Workload Scheduler (#28733)

Browse Source 
1 Add select workload schedule policy tvf
2 Add reg test
This commit is contained in:
wangbo
2023-12-22 12:09:51 +08:00
committed by GitHub
parent c72ad9b673
commit 012e66729a
18 changed files with 637 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
fe/fe-core/src/main/java/org/apache/doris/analysis/CreateWorkloadSchedPolicyStmt.java
View File

@ -88,4 +88,38 @@ public class CreateWorkloadSchedPolicyStmt extends DdlStmt {
public Map<String, String> getProperties() {
return properties;
}
@Override
public String toSql() {
String str = "";
str = str + "CREAYE ";
str = str + "WORKLOAD SCHEDULE POLICY " + policyName + " ";
str = str + " CONDITIONS( ";
if (conditions != null) {
for (WorkloadConditionMeta wcm : conditions) {
str += wcm.toString() + ",";
}
}
str = str.substring(0, str.length() - 1);
str = str + ")";
str = str + " ACTIONS( ";
if (actions != null) {
for (WorkloadActionMeta wam : actions) {
str = str + wam.toString() + ",";
}
}
str = str.substring(0, str.length() - 1);
str = str + ")";
str = str + " PROPERTIES(";
for (Map.Entry<String, String> entry : properties.entrySet()) {
str = str + "\"" + entry.getKey() + "\"" + "=" + "\"" + entry.getValue() + "\",";
}
str = str.substring(0, str.length() - 1);
str = str + ")";
return str;
}
}

13
fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java
View File

@ -445,6 +445,19 @@ public class WorkloadGroupMgr implements Writable, GsonPostProcessable {
}
}
public String getWorkloadGroupNameById(Long id) {
readLock();
try {
WorkloadGroup wg = idToWorkloadGroup.get(id);
if (wg == null) {
return null;
}
return wg.getName();
} finally {
readUnlock();
}
}
// for ut
public Map<String, WorkloadGroup> getNameToWorkloadGroup() {
return nameToWorkloadGroup;

16
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadActionMeta.java
View File

@ -17,9 +17,11 @@
package org.apache.doris.resource.workloadschedpolicy;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.UserException;
import com.google.gson.annotations.SerializedName;
import org.apache.commons.lang3.StringUtils;
public class WorkloadActionMeta {
@ -44,4 +46,18 @@ public class WorkloadActionMeta {
}
throw new UserException("invalid action type " + strType);
}
public String toString() {
if (StringUtils.isEmpty(actionArgs)) {
return action.toString();
} else {
String retActionArgs = actionArgs;
if (WorkloadActionType.MOVE_QUERY_TO_GROUP.equals(action)) {
retActionArgs = Env.getCurrentEnv().getWorkloadGroupMgr()
.getWorkloadGroupNameById(Long.valueOf(actionArgs));
}
retActionArgs = retActionArgs == null ? "-1" : retActionArgs;
return action + " \"" + retActionArgs + "\"";
}
}
}

18
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadConditionCompareUtils.java
View File

@ -39,6 +39,24 @@ public class WorkloadConditionCompareUtils {
}
}
// used for select tvf
static String getOperatorStr(WorkloadConditionOperator op) {
switch (op) {
case EQUAL:
return "=";
case GREATER:
return ">";
case GREATER_EQUAL:
return ">=";
case LESS:
return "<";
case LESS_EQUAl:
return "<=";
default:
throw new RuntimeException("unexpected compare operator " + op);
}
}
static boolean compareInteger(WorkloadConditionOperator operator, long firstArgs, long secondArgs) {
switch (operator) {
case EQUAL:

2
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadConditionMeta.java
View File

@ -49,6 +49,6 @@ public class WorkloadConditionMeta {
}
public String toString() {
return metricName + " " + op + " " + value;
return metricName + " " + WorkloadConditionCompareUtils.getOperatorStr(op) + " " + value;
}
}

2
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadQueryInfo.java
View File

@ -26,5 +26,5 @@ public class WorkloadQueryInfo {
String queryId = null;
TUniqueId tUniqueId = null;
ConnectContext context = null;
Map<WorkloadMetricType, String> metricMap;
public Map<WorkloadMetricType, String> metricMap;
}

11
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicy.java
View File

@ -62,6 +62,15 @@ public class WorkloadSchedPolicy implements Writable, GsonPostProcessable {
private List<WorkloadCondition> workloadConditionList;
private List<WorkloadAction> workloadActionList;
// for ut
public WorkloadSchedPolicy() {
}
// for ut
public void setWorkloadConditionList(List<WorkloadCondition> workloadConditionList) {
this.workloadConditionList = workloadConditionList;
}
public WorkloadSchedPolicy(long id, String name, List<WorkloadCondition> workloadConditionList,
List<WorkloadAction> workloadActionList, Map<String, String> properties) throws UserException {
this.id = id;
@ -77,7 +86,7 @@ public class WorkloadSchedPolicy implements Writable, GsonPostProcessable {
// return false,
// 1 metric not match
// 2 condition value not match query info's value
boolean isMatch(WorkloadQueryInfo queryInfo) {
public boolean isMatch(WorkloadQueryInfo queryInfo) {
for (WorkloadCondition condition : workloadConditionList) {
WorkloadMetricType metricType = condition.getMetricType();
String value = queryInfo.metricMap.get(metricType);

67
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java
View File

@ -34,12 +34,12 @@ import org.apache.doris.persist.gson.GsonPostProcessable;
import org.apache.doris.persist.gson.GsonUtils;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.service.ExecuteEnv;
import org.apache.doris.thrift.TUserIdentity;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Maps;
import com.google.gson.annotations.SerializedName;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@ -69,7 +69,7 @@ public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
public static final ImmutableList<String> WORKLOAD_SCHED_POLICY_NODE_TITLE_NAMES
= new ImmutableList.Builder<String>()
.add("Id").add("Name").add("ItemName").add("ItemValue")
.add("Id").add("Name").add("Condition").add("Action").add("Priority").add("Enabled").add("Version")
.build();
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
@ -349,7 +349,8 @@ public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
throw new UserException("policy's priority can only between 0 ~ 100");
}
} catch (NumberFormatException e) {
throw new UserException("policy's priority must be a number, input value=" + priorityStr);
throw new UserException(
"invalid priority property value, it must be a number, input value=" + priorityStr);
}
}
}
@ -448,6 +449,11 @@ public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
return policyProcNode.fetchResult(currentUserIdentity).getRows();
}
public List<List<String>> getWorkloadSchedPolicyTvfInfo(TUserIdentity tcurrentUserIdentity) {
UserIdentity currentUserIdentity = UserIdentity.fromThrift(tcurrentUserIdentity);
return policyProcNode.fetchResult(currentUserIdentity).getRows();
}
public class PolicyProcNode {
public ProcResult fetchResult(UserIdentity currentUserIdentity) {
BaseProcResult result = new BaseProcResult();
@ -460,54 +466,31 @@ public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
continue;
}
String pId = String.valueOf(policy.getId());
List<String> row = new ArrayList<>();
String pName = policy.getName();
row.add(String.valueOf(policy.getId()));
row.add(pName);
List<WorkloadConditionMeta> conditionList = policy.getConditionMetaList();
StringBuilder cmStr = new StringBuilder();
for (WorkloadConditionMeta cm : conditionList) {
List<String> condRow = new ArrayList<>();
condRow.add(pId);
condRow.add(pName);
condRow.add("condition");
condRow.add(cm.toString());
result.addRow(condRow);
cmStr.append(cm.toString()).append(";");
}
String retStr = cmStr.toString().toLowerCase();
row.add(retStr.substring(0, retStr.length() - 1));
List<WorkloadActionMeta> actionList = policy.getActionMetaList();
for (WorkloadActionMeta workloadActionMeta : actionList) {
List<String> actionRow = new ArrayList<>();
actionRow.add(pId);
actionRow.add(pName);
actionRow.add("action");
if (StringUtils.isEmpty(workloadActionMeta.actionArgs)) {
actionRow.add(workloadActionMeta.action.toString());
} else {
actionRow.add(workloadActionMeta.action + " " + workloadActionMeta.actionArgs);
}
result.addRow(actionRow);
StringBuilder actionStr = new StringBuilder();
for (WorkloadActionMeta am : actionList) {
actionStr.append(am.toString()).append(";");
}
String retStr2 = actionStr.toString().toLowerCase();
row.add(retStr2.substring(0, retStr2.length() - 1));
List<String> prioRow = new ArrayList<>();
prioRow.add(pId);
prioRow.add(pName);
prioRow.add("priority");
prioRow.add(String.valueOf(policy.getPriority()));
result.addRow(prioRow);
List<String> enabledRow = new ArrayList<>();
enabledRow.add(pId);
enabledRow.add(pName);
enabledRow.add("enabled");
enabledRow.add(String.valueOf(policy.isEnabled()));
result.addRow(enabledRow);
List<String> versionRow = new ArrayList<>();
versionRow.add(pId);
versionRow.add(pName);
versionRow.add("version");
versionRow.add(String.valueOf(policy.getVersion()));
result.addRow(versionRow);
row.add(String.valueOf(policy.getPriority()));
row.add(String.valueOf(policy.isEnabled()));
row.add(String.valueOf(policy.getVersion()));
result.addRow(row);
}
} finally {
readUnlock();

30
fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
View File

@ -116,6 +116,9 @@ public class MetadataGenerator {
case QUERIES:
result = queriesMetadataResult(params, request);
break;
case WORKLOAD_SCHED_POLICY:
result = workloadSchedPolicyMetadataResult(params);
break;
default:
return errorResult("Metadata table params is not set.");
}
@ -383,6 +386,33 @@ public class MetadataGenerator {
return result;
}
private static TFetchSchemaTableDataResult workloadSchedPolicyMetadataResult(TMetadataTableRequestParams params) {
if (!params.isSetCurrentUserIdent()) {
return errorResult("current user ident is not set.");
}
TUserIdentity tcurrentUserIdentity = params.getCurrentUserIdent();
List<List<String>> workloadPolicyList = Env.getCurrentEnv().getWorkloadSchedPolicyMgr()
.getWorkloadSchedPolicyTvfInfo(tcurrentUserIdentity);
TFetchSchemaTableDataResult result = new TFetchSchemaTableDataResult();
List<TRow> dataBatch = Lists.newArrayList();
for (List<String> policyRow : workloadPolicyList) {
TRow trow = new TRow();
trow.addToColumnValue(new TCell().setLongVal(Long.valueOf(policyRow.get(0)))); // id
trow.addToColumnValue(new TCell().setStringVal(policyRow.get(1))); // name
trow.addToColumnValue(new TCell().setStringVal(policyRow.get(2))); // condition
trow.addToColumnValue(new TCell().setStringVal(policyRow.get(3))); // action
trow.addToColumnValue(new TCell().setIntVal(Integer.valueOf(policyRow.get(4)))); // priority
trow.addToColumnValue(new TCell().setBoolVal(Boolean.valueOf(policyRow.get(5)))); // enabled
trow.addToColumnValue(new TCell().setIntVal(Integer.valueOf(policyRow.get(6)))); // version
dataBatch.add(trow);
}
result.setDataBatch(dataBatch);
result.setStatus(new TStatus(TStatusCode.OK));
return result;
}
private static TFetchSchemaTableDataResult queriesMetadataResult(TMetadataTableRequestParams params,
TFetchSchemaTableDataRequest parentRequest) {
if (!params.isSetQueriesMetadataParams()) {

2
fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataTableValuedFunction.java
View File

@ -51,6 +51,8 @@ public abstract class MetadataTableValuedFunction extends TableValuedFunctionIf
return TasksTableValuedFunction.getColumnIndexFromColumnName(columnName, params);
case QUERIES:
return QueriesTableValuedFunction.getColumnIndexFromColumnName(columnName);
case WORKLOAD_SCHED_POLICY:
return WorkloadSchedPolicyTableValuedFunction.getColumnIndexFromColumnName(columnName);
default:
throw new AnalysisException("Unknown Metadata TableValuedFunction type");
}

2
fe/fe-core/src/main/java/org/apache/doris/tablefunction/TableValuedFunctionIf.java
View File

@ -76,6 +76,8 @@ public abstract class TableValuedFunctionIf {
return new GroupCommitTableValuedFunction(params);
case QueriesTableValuedFunction.NAME:
return new QueriesTableValuedFunction(params);
case WorkloadSchedPolicyTableValuedFunction.NAME:
return new WorkloadSchedPolicyTableValuedFunction(params);
default:
throw new AnalysisException("Could not find table function " + funcName);
}

88
fe/fe-core/src/main/java/org/apache/doris/tablefunction/WorkloadSchedPolicyTableValuedFunction.java Normal file
View File

@ -0,0 +1,88 @@
// 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.tablefunction;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.catalog.ScalarType;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.thrift.TMetaScanRange;
import org.apache.doris.thrift.TMetadataType;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import java.util.List;
import java.util.Map;
public class WorkloadSchedPolicyTableValuedFunction extends MetadataTableValuedFunction {
public static final String NAME = "workload_schedule_policy";
private static final ImmutableList<Column> SCHEMA = ImmutableList.of(
new Column("Id", ScalarType.createType(PrimitiveType.BIGINT)),
new Column("Name", ScalarType.createStringType()),
new Column("Condition", ScalarType.createType(PrimitiveType.STRING)),
new Column("Action", ScalarType.createType(PrimitiveType.STRING)),
new Column("Priority", ScalarType.createType(PrimitiveType.INT)),
new Column("Enabled", ScalarType.createType(PrimitiveType.BOOLEAN)),
new Column("Version", ScalarType.createType(PrimitiveType.INT)));
private static final ImmutableMap<String, Integer> COLUMN_TO_INDEX;
static {
ImmutableMap.Builder<String, Integer> builder = new ImmutableMap.Builder();
for (int i = 0; i < SCHEMA.size(); i++) {
builder.put(SCHEMA.get(i).getName().toLowerCase(), i);
}
COLUMN_TO_INDEX = builder.build();
}
public static Integer getColumnIndexFromColumnName(String columnName) {
return COLUMN_TO_INDEX.get(columnName.toLowerCase());
}
public WorkloadSchedPolicyTableValuedFunction(Map<String, String> params) {
if (params.size() > 0) {
throw new org.apache.doris.nereids.exceptions.AnalysisException(
"workload schedule policy table-valued-function does not support any params");
}
}
@Override
public TMetadataType getMetadataType() {
return TMetadataType.WORKLOAD_SCHED_POLICY;
}
@Override
public TMetaScanRange getMetaScanRange() {
TMetaScanRange metaScanRange = new TMetaScanRange();
metaScanRange.setMetadataType(TMetadataType.WORKLOAD_SCHED_POLICY);
return metaScanRange;
}
@Override
public String getTableName() {
return "WorkloadSchedPolicyTableValuedFunction";
}
@Override
public List<Column> getTableColumns() throws AnalysisException {
return SCHEMA;
}
}