You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "wangbo (via GitHub)" <gi...@apache.org> on 2023/12/15 03:04:48 UTC
[PR] [feature][execturo]support query schedule policy [doris]
wangbo opened a new pull request, #28443:
URL: https://github.com/apache/doris/pull/28443
support query schedule policy
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1427632597
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/PolicyCondition.java:
##########
@@ -0,0 +1,39 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.common.UserException;
+
+
+public interface PolicyCondition {
Review Comment:
WorkloadSchedCondition，要不还是简写吧，这也太长了
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429617971
##########
fe/fe-core/src/main/java/org/apache/doris/common/publish/WorkloadMoveActionPublisherThread.java:
##########
@@ -0,0 +1,122 @@
+// 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.common.publish;
+
+import org.apache.doris.common.ClientPool;
+import org.apache.doris.common.ThreadPoolManager;
+import org.apache.doris.common.util.Daemon;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.BackendService;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TPublishTopicRequest;
+import org.apache.doris.thrift.TTopicInfoType;
+import org.apache.doris.thrift.TWorkloadMoveQueryToGroupAction;
+import org.apache.doris.thrift.TopicInfo;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+public class WorkloadMoveActionPublisherThread extends Daemon {
Review Comment:
独立线程主要考虑是两点，第一个是action的数量不会太多。第二个是，publishworkload group的那个线程比较特殊，那个只能master启动，而action这个是每个fe都要起的。其次线程活跃的间隔也不同，moveaction的间隔肯定是越短越好。当然这里可以考虑的设计是，所有action用一个线程
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1430885182
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java:
##########
@@ -506,6 +509,8 @@ public class Env {
private TopicPublisherThread topicPublisherThread;
+ private WorkloadActionPublishThread workloadMoveActionPublisherThread;
Review Comment:
workloadMoveActionPublisherThread --》 workloadActionPublishThread
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1427651703
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicy.java:
##########
@@ -0,0 +1,188 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonPostProcessable;
+import org.apache.doris.persist.gson.GsonUtils;
+
+import com.esotericsoftware.minlog.Log;
+import com.google.common.collect.ImmutableSet;
+import com.google.gson.annotations.SerializedName;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class WorkloadSchedPolicy implements Writable, GsonPostProcessable {
Review Comment:
policy应该只是封装了condition和action的元数据，可能把那个线程改成叫WorkloadSChedulerThread会好点，那个线程本身是负责调度的
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28443:
URL: https://github.com/apache/doris/pull/28443#issuecomment-1862281058
PR approved by anyone and no changes requested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429572910
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadAction.java:
##########
@@ -0,0 +1,41 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.common.UserException;
+
+public interface WorkloadAction {
+
+ void exec(WorkloadQueryInfo queryInfo);
+
+ WorkloadActionType getWorkloadActionType();
+
+ // NOTE(wb) currently createPolicyAction is also used when replay meta, it better not contains heavy check
+ static WorkloadAction createWorkloadAction(String actionCmd, String actionCmdArgs)
+ throws UserException {
+ if (WorkloadActionType.cancel_query.toString().equals(actionCmd)) {
+ return WorkloadActionCancelQuery.createWorkloadAction();
+ } else if (WorkloadActionType.move_query_to_group.toString().equals(actionCmd)) {
Review Comment:
这里不要to string 比较， 要把createWorkloadAction 的第一个参数输入变成 enum，在解析阶段就做，不要把string 带到非常靠后的流程处理。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1430951537
##########
fe/fe-core/src/main/java/org/apache/doris/common/publish/WorkloadActionPublishThread.java:
##########
@@ -0,0 +1,123 @@
+// 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.common.publish;
+
+import org.apache.doris.common.ClientPool;
+import org.apache.doris.common.ThreadPoolManager;
+import org.apache.doris.common.util.Daemon;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.BackendService;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TPublishTopicRequest;
+import org.apache.doris.thrift.TTopicInfoType;
+import org.apache.doris.thrift.TopicInfo;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+
+public class WorkloadActionPublishThread extends Daemon {
+
+ private ExecutorService executor = ThreadPoolManager
+ .newDaemonFixedThreadPool(4, 256, "workload-action-publish-thread", true);
+
+ private static final Logger LOG = LogManager.getLogger(WorkloadActionPublishThread.class);
+
+ public static Map<TTopicInfoType, List<TopicInfo>> workloadActionToplicInfoMap
+ = new HashMap<TTopicInfoType, List<TopicInfo>>();
+
+ public static synchronized void putWorkloadAction(TTopicInfoType type, TopicInfo topicInfo) {
+ List<TopicInfo> list = workloadActionToplicInfoMap.get(type);
+ if (list == null) {
+ list = new ArrayList<TopicInfo>();
+ workloadActionToplicInfoMap.put(type, list);
+ }
+ list.add(topicInfo);
+ }
+
+ public static synchronized Map<TTopicInfoType, List<TopicInfo>> getCurrentWorkloadActionMap() {
+ Map<TTopicInfoType, List<TopicInfo>> retMap = workloadActionToplicInfoMap;
+ workloadActionToplicInfoMap = new HashMap<TTopicInfoType, List<TopicInfo>>();
+ return retMap;
+ }
+
+ private SystemInfoService clusterInfoService;
+
+ public WorkloadActionPublishThread(String name, long intervalMs,
+ SystemInfoService clusterInfoService) {
+ super(name, intervalMs);
+ this.clusterInfoService = clusterInfoService;
+ }
+
+ @Override
+ protected final void runOneCycle() {
+ Map<TTopicInfoType, List<TopicInfo>> actionMap
+ = WorkloadActionPublishThread.getCurrentWorkloadActionMap();
+ if (actionMap.size() == 0) {
+ LOG.info("no workload action found, skip publish");
+ return;
+ }
+ Collection<Backend> currentBeToPublish = clusterInfoService.getIdToBackend().values();
+ AckResponseHandler handler = new AckResponseHandler(currentBeToPublish);
+ TPublishTopicRequest request = new TPublishTopicRequest();
+ request.setTopicMap(actionMap);
+ for (Backend be : currentBeToPublish) {
+ executor.submit(new WorkloadMoveActionTask(request, be, handler));
Review Comment:
这里其实已经是批量了，每10s一次拿所有的topicinfo，构造一个rpcrequest，然后发送给所有be。也就是每次轮询发送rpc的个数就是be的个数
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][executor]support query schedule policy [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #28443:
URL: https://github.com/apache/doris/pull/28443#issuecomment-1862390648
(From new machine)TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 43.95 seconds
stream load tsv: 563 seconds loaded 74807831229 Bytes, about 126 MB/s
stream load json: 19 seconds loaded 2358488459 Bytes, about 118 MB/s
stream load orc: 66 seconds loaded 1101869774 Bytes, about 15 MB/s
stream load parquet: 32 seconds loaded 861443392 Bytes, about 25 MB/s
insert into select: 28.9 seconds inserted 10000000 Rows, about 346K ops/s
storage size: 17184034761 Bytes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1430892239
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java:
##########
@@ -0,0 +1,534 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.analysis.AlterWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.CreateWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.DropWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.proc.BaseProcResult;
+import org.apache.doris.common.proc.ProcResult;
+import org.apache.doris.common.util.DebugUtil;
+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.service.ExecuteEnv;
+
+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;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
+
+ private static final Logger LOG = LogManager.getLogger(WorkloadSchedPolicyMgr.class);
+
+ @SerializedName(value = "idToPolicy")
+ private Map<Long, WorkloadSchedPolicy> idToPolicy = Maps.newConcurrentMap();
+ private Map<String, WorkloadSchedPolicy> nameToPolicy = Maps.newHashMap();
+
+ private PolicyProcNode policyProcNode = new PolicyProcNode();
+
+ public static final ImmutableList<String> WORKLOAD_SCHED_POLICY_NODE_TITLE_NAMES
+ = new ImmutableList.Builder<String>()
+ .add("Id").add("Name").add("ItemName").add("ItemValue")
+ .build();
+
+ private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public static Comparator<WorkloadSchedPolicy> policyComparator = new Comparator<WorkloadSchedPolicy>() {
+ @Override
+ public int compare(WorkloadSchedPolicy p1, WorkloadSchedPolicy p2) {
+ return p2.getPriority() - p1.getPriority();
+ }
+ };
+
+ private Thread policyExecThread = new Thread() {
+
+ @Override
+ public void run() {
+ while (true) {
+ try {
+ // 1 get query info map
+ Map<Integer, ConnectContext> connectMap = ExecuteEnv.getInstance().getScheduler()
+ .getConnectionMap();
Review Comment:
这个可能不对。 对于统计信息收集的query，或者构建mtmv的query，可能没有connection map？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1430051181
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -2268,6 +2268,12 @@ public class Config extends ConfigBase {
@ConfField(mutable = true, masterOnly = true)
public static int publish_topic_info_interval_ms = 30000; // 30s
+ @ConfField(mutable = true)
+ public static int workload_sched_policy_interval_ms = 10000; // 10s
+
+ @ConfField(mutable = true)
+ public static int workload_move_action_interval_ms = 10000; // 10s
Review Comment:
所有需要通知be的action收敛到一个线程，然后用一个间隔吧
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1430891191
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java:
##########
@@ -0,0 +1,534 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.analysis.AlterWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.CreateWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.DropWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.proc.BaseProcResult;
+import org.apache.doris.common.proc.ProcResult;
+import org.apache.doris.common.util.DebugUtil;
+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.service.ExecuteEnv;
+
+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;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
+
+ private static final Logger LOG = LogManager.getLogger(WorkloadSchedPolicyMgr.class);
+
+ @SerializedName(value = "idToPolicy")
+ private Map<Long, WorkloadSchedPolicy> idToPolicy = Maps.newConcurrentMap();
Review Comment:
我们得设定一下最多的policy 数量，否则数量很大，我们匹配的时间也很久
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java:
##########
@@ -0,0 +1,534 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.analysis.AlterWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.CreateWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.DropWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.proc.BaseProcResult;
+import org.apache.doris.common.proc.ProcResult;
+import org.apache.doris.common.util.DebugUtil;
+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.service.ExecuteEnv;
+
+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;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
+
+ private static final Logger LOG = LogManager.getLogger(WorkloadSchedPolicyMgr.class);
+
+ @SerializedName(value = "idToPolicy")
+ private Map<Long, WorkloadSchedPolicy> idToPolicy = Maps.newConcurrentMap();
+ private Map<String, WorkloadSchedPolicy> nameToPolicy = Maps.newHashMap();
+
+ private PolicyProcNode policyProcNode = new PolicyProcNode();
+
+ public static final ImmutableList<String> WORKLOAD_SCHED_POLICY_NODE_TITLE_NAMES
+ = new ImmutableList.Builder<String>()
+ .add("Id").add("Name").add("ItemName").add("ItemValue")
+ .build();
+
+ private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public static Comparator<WorkloadSchedPolicy> policyComparator = new Comparator<WorkloadSchedPolicy>() {
+ @Override
+ public int compare(WorkloadSchedPolicy p1, WorkloadSchedPolicy p2) {
+ return p2.getPriority() - p1.getPriority();
+ }
+ };
+
+ private Thread policyExecThread = new Thread() {
+
+ @Override
+ public void run() {
+ while (true) {
+ try {
+ // 1 get query info map
+ Map<Integer, ConnectContext> connectMap = ExecuteEnv.getInstance().getScheduler()
+ .getConnectionMap();
+ List<WorkloadQueryInfo> queryInfoList = new ArrayList<>();
+
+ // a snapshot for connect context
+ Set<Integer> keySet = new HashSet<>();
+ keySet.addAll(connectMap.keySet());
+
+ for (Integer connectId : keySet) {
+ ConnectContext cctx = connectMap.get(connectId);
+ if (cctx == null || cctx.isKilled()) {
+ continue;
+ }
+
+ String username = cctx.getQualifiedUser();
+ long queryTime = System.currentTimeMillis() - cctx.getStartTime();
+
+ WorkloadQueryInfo policyQueryInfo = new WorkloadQueryInfo();
+ policyQueryInfo.queryId = cctx.queryId() == null ? null : DebugUtil.printId(cctx.queryId());
+ policyQueryInfo.tUniqueId = cctx.queryId();
+ policyQueryInfo.context = cctx;
+ policyQueryInfo.metricMap = new HashMap<>();
+ policyQueryInfo.metricMap.put(WorkloadMetricType.USERNAME, username);
+ policyQueryInfo.metricMap.put(WorkloadMetricType.QUERY_TIME, String.valueOf(queryTime));
+
+ queryInfoList.add(policyQueryInfo);
+ }
+
+ // 2 exec policy
+ if (queryInfoList.size() > 0) {
+ execPolicy(queryInfoList);
+ }
+ } catch (Throwable t) {
+ LOG.error("[policy thread]error happens when exec policy");
+ }
+
+ // 3 sleep
+ try {
+ Thread.sleep(Config.workload_sched_policy_interval_ms);
+ } catch (InterruptedException e) {
+ LOG.error("error happends when policy exec thread sleep");
+ }
+ }
+ }
+ };
+
+ public void start() {
+ policyExecThread.setName("workload-auto-scheduler-thread");
+ policyExecThread.start();
+ }
+
+ public void createWorkloadSchedPolicy(CreateWorkloadSchedPolicyStmt createStmt) throws UserException {
+ String policyName = createStmt.getPolicyName();
+
+ // 1 create condition
Review Comment:
我们得设定一下最多的policy 数量，否则数量很大，我们匹配的时间也很久
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1430895609
##########
fe/fe-core/src/main/java/org/apache/doris/common/publish/WorkloadActionPublishThread.java:
##########
@@ -0,0 +1,123 @@
+// 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.common.publish;
+
+import org.apache.doris.common.ClientPool;
+import org.apache.doris.common.ThreadPoolManager;
+import org.apache.doris.common.util.Daemon;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.BackendService;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TPublishTopicRequest;
+import org.apache.doris.thrift.TTopicInfoType;
+import org.apache.doris.thrift.TopicInfo;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+
+public class WorkloadActionPublishThread extends Daemon {
+
+ private ExecutorService executor = ThreadPoolManager
+ .newDaemonFixedThreadPool(4, 256, "workload-action-publish-thread", true);
+
+ private static final Logger LOG = LogManager.getLogger(WorkloadActionPublishThread.class);
+
+ public static Map<TTopicInfoType, List<TopicInfo>> workloadActionToplicInfoMap
+ = new HashMap<TTopicInfoType, List<TopicInfo>>();
+
+ public static synchronized void putWorkloadAction(TTopicInfoType type, TopicInfo topicInfo) {
+ List<TopicInfo> list = workloadActionToplicInfoMap.get(type);
+ if (list == null) {
+ list = new ArrayList<TopicInfo>();
+ workloadActionToplicInfoMap.put(type, list);
+ }
+ list.add(topicInfo);
+ }
+
+ public static synchronized Map<TTopicInfoType, List<TopicInfo>> getCurrentWorkloadActionMap() {
+ Map<TTopicInfoType, List<TopicInfo>> retMap = workloadActionToplicInfoMap;
+ workloadActionToplicInfoMap = new HashMap<TTopicInfoType, List<TopicInfo>>();
+ return retMap;
+ }
+
+ private SystemInfoService clusterInfoService;
+
+ public WorkloadActionPublishThread(String name, long intervalMs,
+ SystemInfoService clusterInfoService) {
+ super(name, intervalMs);
+ this.clusterInfoService = clusterInfoService;
+ }
+
+ @Override
+ protected final void runOneCycle() {
+ Map<TTopicInfoType, List<TopicInfo>> actionMap
+ = WorkloadActionPublishThread.getCurrentWorkloadActionMap();
+ if (actionMap.size() == 0) {
+ LOG.info("no workload action found, skip publish");
+ return;
+ }
+ Collection<Backend> currentBeToPublish = clusterInfoService.getIdToBackend().values();
+ AckResponseHandler handler = new AckResponseHandler(currentBeToPublish);
+ TPublishTopicRequest request = new TPublishTopicRequest();
+ request.setTopicMap(actionMap);
+ for (Backend be : currentBeToPublish) {
+ executor.submit(new WorkloadMoveActionTask(request, be, handler));
Review Comment:
这里设计成批量的接口吧，一个个的publish 太慢了，可能rpc 交互太多
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28443:
URL: https://github.com/apache/doris/pull/28443#issuecomment-1859522395
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1427632105
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicy.java:
##########
@@ -0,0 +1,188 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonPostProcessable;
+import org.apache.doris.persist.gson.GsonUtils;
+
+import com.esotericsoftware.minlog.Log;
+import com.google.common.collect.ImmutableSet;
+import com.google.gson.annotations.SerializedName;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class WorkloadSchedPolicy implements Writable, GsonPostProcessable {
+
+ public static final String ENABLED = "enabled";
+ public static final String PRIORITY = "priority";
+ public static final ImmutableSet<String> POLICY_PROPERTIES = new ImmutableSet.Builder<String>()
+ .add(ENABLED).add(PRIORITY).build();
+
+ @SerializedName(value = "id")
+ long id;
+ @SerializedName(value = "name")
+ String name;
+ @SerializedName(value = "version")
+ int version;
+
+ @SerializedName(value = "enabled")
+ private volatile boolean enabled;
+ @SerializedName(value = "priority")
+ private volatile int priority;
+
+ @SerializedName(value = "conditionMetaList")
+ List<ConditionMeta> conditionMetaList;
+ // we regard action as a command, map's key is command, map's value is args, so it's a command list
+ @SerializedName(value = "actionMetaList")
+ Map<String, String> actionMetaList;
+
+ private List<PolicyCondition> policyConditionList;
+ private List<PolicyAction> policyActionList;
+
+ public WorkloadSchedPolicy(long id, String name, List<PolicyCondition> policyConditionList,
+ List<PolicyAction> policyActionList, Map<String, String> properties) throws UserException {
+ this.id = id;
+ this.name = name;
+ this.policyConditionList = policyConditionList;
+ this.policyActionList = policyActionList;
+ // set enable and priority
+ parseAndSetProperties(properties);
+ this.version = 0;
+ }
+
+ // return true, this means all conditions in policy can match queryInfo's data
+ // return false,
+ // 1 metric not match
+ // 2 condition value not match query info's value
+ boolean isMatch(PolicyQueryInfo queryInfo) {
Review Comment:
policy本身是分fe和be的，所以你说的这个我感觉应该是在定时线程调用里区分比较好。比如fe的定时线程就只取fe的policy。目前由于缺乏统一的统计信息收集渠道，所以暂不支持混合型的policy，这个得很靠后了。后续再改就行。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1427563698
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicy.java:
##########
@@ -0,0 +1,188 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonPostProcessable;
+import org.apache.doris.persist.gson.GsonUtils;
+
+import com.esotericsoftware.minlog.Log;
+import com.google.common.collect.ImmutableSet;
+import com.google.gson.annotations.SerializedName;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class WorkloadSchedPolicy implements Writable, GsonPostProcessable {
Review Comment:
我们要不直接叫 WorkloadScheduler 感觉名字更好，不仅仅是一个policy
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429565219
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -2268,6 +2268,12 @@ public class Config extends ConfigBase {
@ConfField(mutable = true, masterOnly = true)
public static int publish_topic_info_interval_ms = 30000; // 30s
+ @ConfField(mutable = true)
+ public static int workload_sched_policy_interval_ms = 10000; // 10s
+
+ @ConfField(mutable = true)
+ public static int workload_move_action_interval_ms = 10000; // 10s
Review Comment:
这种的就不要放config了，否则以后越来越多的策略，每个策略都搞一个config就崩溃了。
可能得把这个时间放到action的定义上。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429645572
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java:
##########
@@ -0,0 +1,509 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.analysis.AlterWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.CreateWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.DropWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.proc.BaseProcResult;
+import org.apache.doris.common.proc.ProcResult;
+import org.apache.doris.common.util.DebugUtil;
+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.service.ExecuteEnv;
+
+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.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
+
+ private static final Logger LOG = LogManager.getLogger(WorkloadSchedPolicyMgr.class);
+
+ @SerializedName(value = "idToPolicy")
+ private Map<Long, WorkloadSchedPolicy> idToPolicy = Maps.newConcurrentMap();
+ private Map<String, WorkloadSchedPolicy> nameToPolicy = Maps.newHashMap();
+
+ private PolicyProcNode policyProcNode = new PolicyProcNode();
+
+ public static final ImmutableList<String> WORKLOAD_SCHED_POLICY_NODE_TITLE_NAMES
+ = new ImmutableList.Builder<String>()
+ .add("Id").add("Name").add("ItemName").add("ItemValue")
+ .build();
+
+ private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public static Comparator<WorkloadSchedPolicy> policyComparator = new Comparator<WorkloadSchedPolicy>() {
+ @Override
+ public int compare(WorkloadSchedPolicy p1, WorkloadSchedPolicy p2) {
+ return p2.getPriority() - p1.getPriority();
+ }
+ };
+
+ private Thread policyExecThread = new Thread() {
+ @Override
+ public void run() {
+ while (true) {
+ try {
+ // 1 get query info map
+ Map<Integer, ConnectContext> connectMap = ExecuteEnv.getInstance().getScheduler()
+ .getConnectionMap();
+ List<WorkloadQueryInfo> queryInfoList = new ArrayList<>();
+
+ // a snapshot for connect context
+ Set<Integer> keySet = new HashSet<>();
+ keySet.addAll(connectMap.keySet());
+
+ for (Integer connectId : keySet) {
+ ConnectContext cctx = connectMap.get(connectId);
+ if (cctx == null) {
+ continue;
+ }
+
+ String username = cctx.getQualifiedUser();
+ long queryTime = System.currentTimeMillis() - cctx.getStartTime();
+
+ WorkloadQueryInfo policyQueryInfo = new WorkloadQueryInfo();
+ policyQueryInfo.queryId = DebugUtil.printId(cctx.queryId());
+ policyQueryInfo.tUniqueId = cctx.queryId();
+ policyQueryInfo.context = cctx;
+ policyQueryInfo.metricMap = new HashMap<>();
+ policyQueryInfo.metricMap.put(WorkloadMetricType.username, username);
+ policyQueryInfo.metricMap.put(WorkloadMetricType.query_time, String.valueOf(queryTime));
+
+ queryInfoList.add(policyQueryInfo);
+ }
+
+ // 2 exec policy
+ if (queryInfoList.size() > 0) {
+ execPolicy(queryInfoList);
+ }
+ } catch (Throwable t) {
+ LOG.error("[policy thread]error happens when exec policy");
+ }
+
+ // 3 sleep
+ try {
+ Thread.sleep(Config.workload_sched_policy_interval_ms);
+ } catch (InterruptedException e) {
+ LOG.error("error happends when policy exec thread sleep");
+ }
+ }
+ }
+ };
+
+ public void start() {
+ policyExecThread.start();
+ }
+
+ public void createWorkloadSchedPolicy(CreateWorkloadSchedPolicyStmt createStmt) throws UserException {
+ String policyName = createStmt.getPolicyName();
+
+ // 1 create condition
+ List<WorkloadConditionMeta> originConditions = createStmt.getConditions();
+ List<WorkloadCondition> policyConditionList = new ArrayList<>();
+ for (WorkloadConditionMeta cm : originConditions) {
+ WorkloadCondition cond = WorkloadCondition.createWorkloadCondition(cm.metricName, cm.op,
+ cm.value);
+ policyConditionList.add(cond);
+ }
+
+ // 2 create action
+ Map<String, String> originActions = createStmt.getActions();
+ List<WorkloadAction> policyActionList = new ArrayList<>();
+ for (Map.Entry<String, String> actionEntry : originActions.entrySet()) {
+ String actionName = actionEntry.getKey();
+ String actionArgs = actionEntry.getValue();
+
+ // we need convert wgName to wgId, because wgName may change
+ if (WorkloadActionType.move_query_to_group.toString().equals(actionName)) {
+ Long wgId = Env.getCurrentEnv().getWorkloadGroupMgr().getWorkloadGroupIdByName(actionArgs);
+ if (wgId == null) {
+ throw new UserException(
+ "can not find workload group " + actionArgs + " when set workload sched policy");
+ }
+ actionEntry.setValue(wgId.toString());
+ }
+
+ WorkloadAction ret = WorkloadAction.createWorkloadAction(actionEntry.getKey(), actionEntry.getValue());
+ policyActionList.add(ret);
+ }
+ checkPolicyActionConflicts(policyActionList);
+
+ // 3 create policy
+ Map<String, String> propMap = createStmt.getProperties();
+ if (propMap == null) {
+ propMap = new HashMap<>();
+ }
+ if (propMap.size() != 0) {
+ checkProperties(propMap);
+ }
+ writeLock();
+ try {
+ if (nameToPolicy.containsKey(createStmt.getPolicyName())) {
+ if (createStmt.isIfNotExists()) {
+ return;
+ } else {
+ throw new UserException("workload schedule policy " + policyName + " already exists ");
+ }
+ }
+ long id = Env.getCurrentEnv().getNextId();
+ WorkloadSchedPolicy policy = new WorkloadSchedPolicy(id, policyName,
+ policyConditionList, policyActionList, propMap);
+ policy.setConditionMeta(originConditions);
+ policy.setActionMeta(originActions);
+ // Env.getCurrentEnv().getEditLog().logCreateWorkloadSchedPolicy(policy);
+ idToPolicy.put(id, policy);
+ nameToPolicy.put(policyName, policy);
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ private void checkPolicyActionConflicts(List<WorkloadAction> actionList) throws UserException {
+ Set<WorkloadActionType> actionTypeSet = new HashSet<>();
+ for (WorkloadAction action : actionList) {
+ if (!actionTypeSet.add(action.getWorkloadActionType())) {
+ throw new UserException("duplicate action in one policy");
+ }
+ }
+ if (actionTypeSet.contains(WorkloadActionType.cancel_query) && actionTypeSet.contains(
+ WorkloadActionType.move_query_to_group)) {
+ throw new UserException(String.format("%s and %s can not exist in one policy at same time",
+ WorkloadActionType.cancel_query, WorkloadActionType.move_query_to_group));
+ }
+ }
+
+ public void execPolicy(List<WorkloadQueryInfo> queryInfoList) {
+ // 1 get a snapshot of policy
+ Set<Long> policyIdSet = new HashSet<>();
+ readLock();
+ try {
+ policyIdSet.addAll(idToPolicy.keySet());
+ } finally {
+ readUnlock();
+ }
+
+ for (WorkloadQueryInfo queryInfo : queryInfoList) {
+ try {
+ // 1 check policy is match
+ Map<WorkloadActionType, Queue<WorkloadSchedPolicy>> matchedPolicyMap = Maps.newHashMap();
+ for (Long policyId : policyIdSet) {
+ WorkloadSchedPolicy policy = idToPolicy.get(policyId);
+ if (policy == null) {
+ continue;
+ }
+ if (policy.isEnabled() && policy.isMatch(queryInfo)) {
+ WorkloadActionType actionType = policy.getFirstActionType();
+ // add to priority queue
+ Queue<WorkloadSchedPolicy> queue = matchedPolicyMap.get(actionType);
+ if (queue == null) {
+ queue = new PriorityQueue<>(policyComparator);
+ matchedPolicyMap.put(actionType, queue);
+ }
+ queue.offer(policy);
Review Comment:
这个queue 有意义吗？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1427562634
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/CancelQueryAction.java:
##########
@@ -0,0 +1,36 @@
+// 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.workloadschedpolicy;
+
+public class CancelQueryAction implements PolicyAction {
+ @Override
+ public void exec(PolicyQueryInfo queryInfo) {
+ if (queryInfo.context != null) {
+ queryInfo.context.cancelQuery();
+ }
+ }
+
+ public static CancelQueryAction createCancelQueryAction() {
Review Comment:
Java 里一般不这么写，如果需要类似工厂的模式，都会写一个单独的factory class
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429635537
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadCondition.java:
##########
@@ -0,0 +1,42 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.common.UserException;
+
+
+public interface WorkloadCondition {
+
+ boolean eval(String strValue);
+
+ WorkloadMetricType getMetricType();
+
+ // NOTE(wb) currently createPolicyCondition is also used when replay meta, it better not contains heavy check
+ static WorkloadCondition createWorkloadCondition(String metricName, String op, String value)
+ throws UserException {
+ if (WorkloadMetricType.username.toString().equals(metricName)) {
Review Comment:
这里也是，不要tostring 比较
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1430891191
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java:
##########
@@ -0,0 +1,534 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.analysis.AlterWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.CreateWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.DropWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.proc.BaseProcResult;
+import org.apache.doris.common.proc.ProcResult;
+import org.apache.doris.common.util.DebugUtil;
+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.service.ExecuteEnv;
+
+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;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
+
+ private static final Logger LOG = LogManager.getLogger(WorkloadSchedPolicyMgr.class);
+
+ @SerializedName(value = "idToPolicy")
+ private Map<Long, WorkloadSchedPolicy> idToPolicy = Maps.newConcurrentMap();
Review Comment:
我们得设定一下最多的policy 数量，否则数量很大，我们匹配的时间也很久
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][executor]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo merged PR #28443:
URL: https://github.com/apache/doris/pull/28443
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][executor]support query schedule policy [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #28443:
URL: https://github.com/apache/doris/pull/28443#issuecomment-1862483429
<details>
<summary>TPC-H test result on machine: 'aliyun_ecs.c7a.8xlarge_32C64G'</summary>
```
Tpch sf100 test result on commit 65f9e0cb5bb152608d1685a941d60004dec9cba2, data reload: false
run tpch-sf100 query with default conf and session variables
q1 4679 4429 4451 4429
q2 359 153 199 153
q3 1467 1287 1196 1196
q4 1118 914 914 914
q5 3173 3171 3176 3171
q6 246 128 126 126
q7 997 497 483 483
q8 2204 2223 2203 2203
q9 6682 6683 6687 6683
q10 3218 3281 3282 3281
q11 303 187 181 181
q12 350 204 205 204
q13 4559 3839 3856 3839
q14 240 210 217 210
q15 568 516 536 516
q16 445 381 382 381
q17 998 683 538 538
q18 7271 7011 7082 7011
q19 1541 1481 1351 1351
q20 562 324 308 308
q21 3030 2678 2728 2678
q22 354 286 280 280
Total cold run time: 44364 ms
Total hot run time: 40136 ms
run tpch-sf100 query with default conf and set session variable runtime_filter_mode=off
q1 4377 4360 4386 4360
q2 273 162 167 162
q3 3530 3527 3537 3527
q4 2400 2384 2377 2377
q5 5731 5758 5760 5758
q6 240 121 122 121
q7 2403 1876 1856 1856
q8 3536 3523 3547 3523
q9 9127 9022 9019 9019
q10 3930 4014 4007 4007
q11 496 366 365 365
q12 760 609 611 609
q13 4286 3596 3563 3563
q14 281 256 253 253
q15 571 528 524 524
q16 500 461 472 461
q17 1889 1869 1862 1862
q18 8899 8255 8367 8255
q19 1751 1754 1793 1754
q20 2262 1942 1936 1936
q21 6544 6183 6265 6183
q22 516 419 424 419
Total cold run time: 64302 ms
Total hot run time: 60894 ms
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1427563174
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/CancelQueryAction.java:
##########
@@ -0,0 +1,36 @@
+// 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.workloadschedpolicy;
+
+public class CancelQueryAction implements PolicyAction {
Review Comment:
所有的类命名前面加前缀 Workload， 要不我们的名字太通用了，容易重名
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1427568188
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/PolicyCondition.java:
##########
@@ -0,0 +1,39 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.common.UserException;
+
+
+public interface PolicyCondition {
Review Comment:
WorkloadSchedulerTriggerCondition
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1430894338
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java:
##########
@@ -0,0 +1,534 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.analysis.AlterWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.CreateWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.DropWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.proc.BaseProcResult;
+import org.apache.doris.common.proc.ProcResult;
+import org.apache.doris.common.util.DebugUtil;
+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.service.ExecuteEnv;
+
+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;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
+
+ private static final Logger LOG = LogManager.getLogger(WorkloadSchedPolicyMgr.class);
+
+ @SerializedName(value = "idToPolicy")
+ private Map<Long, WorkloadSchedPolicy> idToPolicy = Maps.newConcurrentMap();
+ private Map<String, WorkloadSchedPolicy> nameToPolicy = Maps.newHashMap();
+
+ private PolicyProcNode policyProcNode = new PolicyProcNode();
+
+ public static final ImmutableList<String> WORKLOAD_SCHED_POLICY_NODE_TITLE_NAMES
+ = new ImmutableList.Builder<String>()
+ .add("Id").add("Name").add("ItemName").add("ItemValue")
+ .build();
+
+ private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public static Comparator<WorkloadSchedPolicy> policyComparator = new Comparator<WorkloadSchedPolicy>() {
+ @Override
+ public int compare(WorkloadSchedPolicy p1, WorkloadSchedPolicy p2) {
+ return p2.getPriority() - p1.getPriority();
+ }
+ };
+
+ private Thread policyExecThread = new Thread() {
+
+ @Override
+ public void run() {
+ while (true) {
+ try {
+ // 1 get query info map
+ Map<Integer, ConnectContext> connectMap = ExecuteEnv.getInstance().getScheduler()
+ .getConnectionMap();
+ List<WorkloadQueryInfo> queryInfoList = new ArrayList<>();
+
+ // a snapshot for connect context
+ Set<Integer> keySet = new HashSet<>();
+ keySet.addAll(connectMap.keySet());
+
+ for (Integer connectId : keySet) {
+ ConnectContext cctx = connectMap.get(connectId);
+ if (cctx == null || cctx.isKilled()) {
+ continue;
+ }
+
+ String username = cctx.getQualifiedUser();
+ long queryTime = System.currentTimeMillis() - cctx.getStartTime();
+
+ WorkloadQueryInfo policyQueryInfo = new WorkloadQueryInfo();
+ policyQueryInfo.queryId = cctx.queryId() == null ? null : DebugUtil.printId(cctx.queryId());
+ policyQueryInfo.tUniqueId = cctx.queryId();
+ policyQueryInfo.context = cctx;
+ policyQueryInfo.metricMap = new HashMap<>();
+ policyQueryInfo.metricMap.put(WorkloadMetricType.USERNAME, username);
+ policyQueryInfo.metricMap.put(WorkloadMetricType.QUERY_TIME, String.valueOf(queryTime));
+
+ queryInfoList.add(policyQueryInfo);
+ }
+
+ // 2 exec policy
+ if (queryInfoList.size() > 0) {
+ execPolicy(queryInfoList);
+ }
+ } catch (Throwable t) {
+ LOG.error("[policy thread]error happens when exec policy");
+ }
+
+ // 3 sleep
+ try {
+ Thread.sleep(Config.workload_sched_policy_interval_ms);
+ } catch (InterruptedException e) {
+ LOG.error("error happends when policy exec thread sleep");
+ }
+ }
+ }
+ };
+
+ public void start() {
+ policyExecThread.setName("workload-auto-scheduler-thread");
+ policyExecThread.start();
+ }
+
+ public void createWorkloadSchedPolicy(CreateWorkloadSchedPolicyStmt createStmt) throws UserException {
+ String policyName = createStmt.getPolicyName();
+
+ // 1 create condition
+ List<WorkloadConditionMeta> originConditions = createStmt.getConditions();
+ List<WorkloadCondition> policyConditionList = new ArrayList<>();
+ for (WorkloadConditionMeta cm : originConditions) {
+ WorkloadCondition cond = WorkloadCondition.createWorkloadCondition(cm);
+ policyConditionList.add(cond);
+ }
+
+ // 2 create action
+ List<WorkloadActionMeta> originActions = createStmt.getActions();
+ List<WorkloadAction> policyActionList = new ArrayList<>();
+ for (WorkloadActionMeta workloadActionMeta : originActions) {
+ WorkloadActionType actionName = workloadActionMeta.action;
+ String actionArgs = workloadActionMeta.actionArgs;
+
+ // we need convert wgName to wgId, because wgName may change
+ if (WorkloadActionType.MOVE_QUERY_TO_GROUP.equals(actionName)) {
+ Long wgId = Env.getCurrentEnv().getWorkloadGroupMgr().getWorkloadGroupIdByName(actionArgs);
+ if (wgId == null) {
+ throw new UserException(
+ "can not find workload group " + actionArgs + " when set workload sched policy");
+ }
+ workloadActionMeta.actionArgs = wgId.toString();
Review Comment:
如果，你再这里改了，那么show policy的时候是不是看到的就是id了？ 此时用户是不是就看不懂了？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28443:
URL: https://github.com/apache/doris/pull/28443#issuecomment-1862279522
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1430896708
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java:
##########
@@ -0,0 +1,534 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.analysis.AlterWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.CreateWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.DropWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.proc.BaseProcResult;
+import org.apache.doris.common.proc.ProcResult;
+import org.apache.doris.common.util.DebugUtil;
+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.service.ExecuteEnv;
+
+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;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
+
+ private static final Logger LOG = LogManager.getLogger(WorkloadSchedPolicyMgr.class);
+
+ @SerializedName(value = "idToPolicy")
+ private Map<Long, WorkloadSchedPolicy> idToPolicy = Maps.newConcurrentMap();
+ private Map<String, WorkloadSchedPolicy> nameToPolicy = Maps.newHashMap();
+
+ private PolicyProcNode policyProcNode = new PolicyProcNode();
+
+ public static final ImmutableList<String> WORKLOAD_SCHED_POLICY_NODE_TITLE_NAMES
+ = new ImmutableList.Builder<String>()
+ .add("Id").add("Name").add("ItemName").add("ItemValue")
+ .build();
+
+ private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public static Comparator<WorkloadSchedPolicy> policyComparator = new Comparator<WorkloadSchedPolicy>() {
+ @Override
+ public int compare(WorkloadSchedPolicy p1, WorkloadSchedPolicy p2) {
+ return p2.getPriority() - p1.getPriority();
+ }
+ };
+
+ private Thread policyExecThread = new Thread() {
+
+ @Override
+ public void run() {
+ while (true) {
+ try {
+ // 1 get query info map
+ Map<Integer, ConnectContext> connectMap = ExecuteEnv.getInstance().getScheduler()
+ .getConnectionMap();
+ List<WorkloadQueryInfo> queryInfoList = new ArrayList<>();
+
+ // a snapshot for connect context
+ Set<Integer> keySet = new HashSet<>();
+ keySet.addAll(connectMap.keySet());
+
+ for (Integer connectId : keySet) {
+ ConnectContext cctx = connectMap.get(connectId);
+ if (cctx == null || cctx.isKilled()) {
+ continue;
+ }
+
+ String username = cctx.getQualifiedUser();
+ long queryTime = System.currentTimeMillis() - cctx.getStartTime();
+
+ WorkloadQueryInfo policyQueryInfo = new WorkloadQueryInfo();
+ policyQueryInfo.queryId = cctx.queryId() == null ? null : DebugUtil.printId(cctx.queryId());
+ policyQueryInfo.tUniqueId = cctx.queryId();
+ policyQueryInfo.context = cctx;
+ policyQueryInfo.metricMap = new HashMap<>();
+ policyQueryInfo.metricMap.put(WorkloadMetricType.USERNAME, username);
+ policyQueryInfo.metricMap.put(WorkloadMetricType.QUERY_TIME, String.valueOf(queryTime));
+
+ queryInfoList.add(policyQueryInfo);
+ }
+
+ // 2 exec policy
+ if (queryInfoList.size() > 0) {
+ execPolicy(queryInfoList);
+ }
+ } catch (Throwable t) {
+ LOG.error("[policy thread]error happens when exec policy");
+ }
+
+ // 3 sleep
+ try {
+ Thread.sleep(Config.workload_sched_policy_interval_ms);
+ } catch (InterruptedException e) {
+ LOG.error("error happends when policy exec thread sleep");
+ }
+ }
+ }
+ };
+
+ public void start() {
+ policyExecThread.setName("workload-auto-scheduler-thread");
+ policyExecThread.start();
+ }
+
+ public void createWorkloadSchedPolicy(CreateWorkloadSchedPolicyStmt createStmt) throws UserException {
+ String policyName = createStmt.getPolicyName();
+
+ // 1 create condition
Review Comment:
这个我记录到jira了
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1427569466
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicy.java:
##########
@@ -0,0 +1,188 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonPostProcessable;
+import org.apache.doris.persist.gson.GsonUtils;
+
+import com.esotericsoftware.minlog.Log;
+import com.google.common.collect.ImmutableSet;
+import com.google.gson.annotations.SerializedName;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class WorkloadSchedPolicy implements Writable, GsonPostProcessable {
+
+ public static final String ENABLED = "enabled";
+ public static final String PRIORITY = "priority";
+ public static final ImmutableSet<String> POLICY_PROPERTIES = new ImmutableSet.Builder<String>()
+ .add(ENABLED).add(PRIORITY).build();
+
+ @SerializedName(value = "id")
+ long id;
+ @SerializedName(value = "name")
+ String name;
+ @SerializedName(value = "version")
+ int version;
+
+ @SerializedName(value = "enabled")
+ private volatile boolean enabled;
+ @SerializedName(value = "priority")
+ private volatile int priority;
+
+ @SerializedName(value = "conditionMetaList")
+ List<ConditionMeta> conditionMetaList;
+ // we regard action as a command, map's key is command, map's value is args, so it's a command list
+ @SerializedName(value = "actionMetaList")
+ Map<String, String> actionMetaList;
+
+ private List<PolicyCondition> policyConditionList;
+ private List<PolicyAction> policyActionList;
+
+ public WorkloadSchedPolicy(long id, String name, List<PolicyCondition> policyConditionList,
+ List<PolicyAction> policyActionList, Map<String, String> properties) throws UserException {
+ this.id = id;
+ this.name = name;
+ this.policyConditionList = policyConditionList;
+ this.policyActionList = policyActionList;
+ // set enable and priority
+ parseAndSetProperties(properties);
+ this.version = 0;
+ }
+
+ // return true, this means all conditions in policy can match queryInfo's data
+ // return false,
+ // 1 metric not match
+ // 2 condition value not match query info's value
+ boolean isMatch(PolicyQueryInfo queryInfo) {
Review Comment:
这里面需要多加一个参数， QueryStage 我们不会一下子匹配所有的condition，得分阶段匹配一个scheduler 下的condition。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429639152
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java:
##########
@@ -0,0 +1,509 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.analysis.AlterWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.CreateWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.DropWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.proc.BaseProcResult;
+import org.apache.doris.common.proc.ProcResult;
+import org.apache.doris.common.util.DebugUtil;
+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.service.ExecuteEnv;
+
+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.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
+
+ private static final Logger LOG = LogManager.getLogger(WorkloadSchedPolicyMgr.class);
+
+ @SerializedName(value = "idToPolicy")
+ private Map<Long, WorkloadSchedPolicy> idToPolicy = Maps.newConcurrentMap();
+ private Map<String, WorkloadSchedPolicy> nameToPolicy = Maps.newHashMap();
+
+ private PolicyProcNode policyProcNode = new PolicyProcNode();
+
+ public static final ImmutableList<String> WORKLOAD_SCHED_POLICY_NODE_TITLE_NAMES
+ = new ImmutableList.Builder<String>()
+ .add("Id").add("Name").add("ItemName").add("ItemValue")
+ .build();
+
+ private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public static Comparator<WorkloadSchedPolicy> policyComparator = new Comparator<WorkloadSchedPolicy>() {
+ @Override
+ public int compare(WorkloadSchedPolicy p1, WorkloadSchedPolicy p2) {
+ return p2.getPriority() - p1.getPriority();
+ }
+ };
+
+ private Thread policyExecThread = new Thread() {
Review Comment:
每一个thread 都需要有一个名字，当定位问题的时候可以看到名字
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429569171
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/ShowWorkloadSchedPolicyStmt.java:
##########
@@ -0,0 +1,59 @@
+// 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.analysis;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.common.UserException;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.ShowResultSetMetaData;
+import org.apache.doris.resource.workloadschedpolicy.WorkloadSchedPolicyMgr;
+
+public class ShowWorkloadSchedPolicyStmt extends ShowStmt {
Review Comment:
这个感觉可以没有，我们走tvf 系统表的思路比较好
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1430896553
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java:
##########
@@ -0,0 +1,534 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.analysis.AlterWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.CreateWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.DropWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.proc.BaseProcResult;
+import org.apache.doris.common.proc.ProcResult;
+import org.apache.doris.common.util.DebugUtil;
+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.service.ExecuteEnv;
+
+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;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
+
+ private static final Logger LOG = LogManager.getLogger(WorkloadSchedPolicyMgr.class);
+
+ @SerializedName(value = "idToPolicy")
+ private Map<Long, WorkloadSchedPolicy> idToPolicy = Maps.newConcurrentMap();
+ private Map<String, WorkloadSchedPolicy> nameToPolicy = Maps.newHashMap();
+
+ private PolicyProcNode policyProcNode = new PolicyProcNode();
+
+ public static final ImmutableList<String> WORKLOAD_SCHED_POLICY_NODE_TITLE_NAMES
+ = new ImmutableList.Builder<String>()
+ .add("Id").add("Name").add("ItemName").add("ItemValue")
+ .build();
+
+ private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public static Comparator<WorkloadSchedPolicy> policyComparator = new Comparator<WorkloadSchedPolicy>() {
+ @Override
+ public int compare(WorkloadSchedPolicy p1, WorkloadSchedPolicy p2) {
+ return p2.getPriority() - p1.getPriority();
+ }
+ };
+
+ private Thread policyExecThread = new Thread() {
+
+ @Override
+ public void run() {
+ while (true) {
+ try {
+ // 1 get query info map
+ Map<Integer, ConnectContext> connectMap = ExecuteEnv.getInstance().getScheduler()
+ .getConnectionMap();
+ List<WorkloadQueryInfo> queryInfoList = new ArrayList<>();
+
+ // a snapshot for connect context
+ Set<Integer> keySet = new HashSet<>();
+ keySet.addAll(connectMap.keySet());
+
+ for (Integer connectId : keySet) {
+ ConnectContext cctx = connectMap.get(connectId);
+ if (cctx == null || cctx.isKilled()) {
+ continue;
+ }
+
+ String username = cctx.getQualifiedUser();
+ long queryTime = System.currentTimeMillis() - cctx.getStartTime();
+
+ WorkloadQueryInfo policyQueryInfo = new WorkloadQueryInfo();
+ policyQueryInfo.queryId = cctx.queryId() == null ? null : DebugUtil.printId(cctx.queryId());
+ policyQueryInfo.tUniqueId = cctx.queryId();
+ policyQueryInfo.context = cctx;
+ policyQueryInfo.metricMap = new HashMap<>();
+ policyQueryInfo.metricMap.put(WorkloadMetricType.USERNAME, username);
+ policyQueryInfo.metricMap.put(WorkloadMetricType.QUERY_TIME, String.valueOf(queryTime));
+
+ queryInfoList.add(policyQueryInfo);
+ }
+
+ // 2 exec policy
+ if (queryInfoList.size() > 0) {
+ execPolicy(queryInfoList);
+ }
+ } catch (Throwable t) {
+ LOG.error("[policy thread]error happens when exec policy");
+ }
+
+ // 3 sleep
+ try {
+ Thread.sleep(Config.workload_sched_policy_interval_ms);
+ } catch (InterruptedException e) {
+ LOG.error("error happends when policy exec thread sleep");
+ }
+ }
+ }
+ };
+
+ public void start() {
+ policyExecThread.setName("workload-auto-scheduler-thread");
+ policyExecThread.start();
+ }
+
+ public void createWorkloadSchedPolicy(CreateWorkloadSchedPolicyStmt createStmt) throws UserException {
+ String policyName = createStmt.getPolicyName();
+
+ // 1 create condition
+ List<WorkloadConditionMeta> originConditions = createStmt.getConditions();
+ List<WorkloadCondition> policyConditionList = new ArrayList<>();
+ for (WorkloadConditionMeta cm : originConditions) {
+ WorkloadCondition cond = WorkloadCondition.createWorkloadCondition(cm);
+ policyConditionList.add(cond);
+ }
+
+ // 2 create action
+ List<WorkloadActionMeta> originActions = createStmt.getActions();
+ List<WorkloadAction> policyActionList = new ArrayList<>();
+ for (WorkloadActionMeta workloadActionMeta : originActions) {
+ WorkloadActionType actionName = workloadActionMeta.action;
+ String actionArgs = workloadActionMeta.actionArgs;
+
+ // we need convert wgName to wgId, because wgName may change
+ if (WorkloadActionType.MOVE_QUERY_TO_GROUP.equals(actionName)) {
+ Long wgId = Env.getCurrentEnv().getWorkloadGroupMgr().getWorkloadGroupIdByName(actionArgs);
+ if (wgId == null) {
+ throw new UserException(
+ "can not find workload group " + actionArgs + " when set workload sched policy");
+ }
+ workloadActionMeta.actionArgs = wgId.toString();
Review Comment:
这个tvf的时候处理下就行了，目前的show主要是用于调试的
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on PR #28443:
URL: https://github.com/apache/doris/pull/28443#issuecomment-1862272056
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429751855
##########
fe/fe-core/src/main/java/org/apache/doris/common/publish/WorkloadMoveActionPublisherThread.java:
##########
@@ -0,0 +1,122 @@
+// 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.common.publish;
+
+import org.apache.doris.common.ClientPool;
+import org.apache.doris.common.ThreadPoolManager;
+import org.apache.doris.common.util.Daemon;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.BackendService;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TPublishTopicRequest;
+import org.apache.doris.thrift.TTopicInfoType;
+import org.apache.doris.thrift.TWorkloadMoveQueryToGroupAction;
+import org.apache.doris.thrift.TopicInfo;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+public class WorkloadMoveActionPublisherThread extends Daemon {
Review Comment:
那就所有action一个线程呗
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429570098
##########
fe/fe-core/src/main/java/org/apache/doris/common/publish/WorkloadMoveActionPublisherThread.java:
##########
@@ -0,0 +1,122 @@
+// 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.common.publish;
+
+import org.apache.doris.common.ClientPool;
+import org.apache.doris.common.ThreadPoolManager;
+import org.apache.doris.common.util.Daemon;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.BackendService;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TPublishTopicRequest;
+import org.apache.doris.thrift.TTopicInfoType;
+import org.apache.doris.thrift.TWorkloadMoveQueryToGroupAction;
+import org.apache.doris.thrift.TopicInfo;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+public class WorkloadMoveActionPublisherThread extends Daemon {
Review Comment:
这个不能每个action 搞一个publisher thread。
这个直接走过去的workload group的publish 过程就可以了，不要单独搞thread了。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1427663661
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/CancelQueryAction.java:
##########
@@ -0,0 +1,36 @@
+// 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.workloadschedpolicy;
+
+public class CancelQueryAction implements PolicyAction {
+ @Override
+ public void exec(PolicyQueryInfo queryInfo) {
+ if (queryInfo.context != null) {
+ queryInfo.context.cancelQuery();
+ }
+ }
+
+ public static CancelQueryAction createCancelQueryAction() {
Review Comment:
那个感觉加的类太多了，这个时候还得是一个if else写起来比较快。而且有了工厂模式也还是得有个根据conditionname去判断取哪个factory的过程
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429613380
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/ShowWorkloadSchedPolicyStmt.java:
##########
@@ -0,0 +1,59 @@
+// 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.analysis;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.common.UserException;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.ShowResultSetMetaData;
+import org.apache.doris.resource.workloadschedpolicy.WorkloadSchedPolicyMgr;
+
+public class ShowWorkloadSchedPolicyStmt extends ShowStmt {
Review Comment:
这个目前临时调试用的比较快。tvf表的展示可能得再想想，因为是一个policy对应多个condition和action，而condition和action的个数可能不同，要是弄一个表里可能得弄个字符串拼接了。不过弄成两个tvf表的话，查着又比较复杂
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429858804
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java:
##########
@@ -0,0 +1,509 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.analysis.AlterWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.CreateWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.DropWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.proc.BaseProcResult;
+import org.apache.doris.common.proc.ProcResult;
+import org.apache.doris.common.util.DebugUtil;
+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.service.ExecuteEnv;
+
+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.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
+
+ private static final Logger LOG = LogManager.getLogger(WorkloadSchedPolicyMgr.class);
+
+ @SerializedName(value = "idToPolicy")
+ private Map<Long, WorkloadSchedPolicy> idToPolicy = Maps.newConcurrentMap();
+ private Map<String, WorkloadSchedPolicy> nameToPolicy = Maps.newHashMap();
+
+ private PolicyProcNode policyProcNode = new PolicyProcNode();
+
+ public static final ImmutableList<String> WORKLOAD_SCHED_POLICY_NODE_TITLE_NAMES
+ = new ImmutableList.Builder<String>()
+ .add("Id").add("Name").add("ItemName").add("ItemValue")
+ .build();
+
+ private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public static Comparator<WorkloadSchedPolicy> policyComparator = new Comparator<WorkloadSchedPolicy>() {
+ @Override
+ public int compare(WorkloadSchedPolicy p1, WorkloadSchedPolicy p2) {
+ return p2.getPriority() - p1.getPriority();
+ }
+ };
+
+ private Thread policyExecThread = new Thread() {
+ @Override
+ public void run() {
+ while (true) {
+ try {
+ // 1 get query info map
+ Map<Integer, ConnectContext> connectMap = ExecuteEnv.getInstance().getScheduler()
+ .getConnectionMap();
+ List<WorkloadQueryInfo> queryInfoList = new ArrayList<>();
+
+ // a snapshot for connect context
+ Set<Integer> keySet = new HashSet<>();
+ keySet.addAll(connectMap.keySet());
+
+ for (Integer connectId : keySet) {
+ ConnectContext cctx = connectMap.get(connectId);
+ if (cctx == null) {
+ continue;
+ }
+
+ String username = cctx.getQualifiedUser();
+ long queryTime = System.currentTimeMillis() - cctx.getStartTime();
+
+ WorkloadQueryInfo policyQueryInfo = new WorkloadQueryInfo();
+ policyQueryInfo.queryId = DebugUtil.printId(cctx.queryId());
+ policyQueryInfo.tUniqueId = cctx.queryId();
+ policyQueryInfo.context = cctx;
+ policyQueryInfo.metricMap = new HashMap<>();
+ policyQueryInfo.metricMap.put(WorkloadMetricType.username, username);
+ policyQueryInfo.metricMap.put(WorkloadMetricType.query_time, String.valueOf(queryTime));
+
+ queryInfoList.add(policyQueryInfo);
+ }
+
+ // 2 exec policy
+ if (queryInfoList.size() > 0) {
+ execPolicy(queryInfoList);
+ }
+ } catch (Throwable t) {
+ LOG.error("[policy thread]error happens when exec policy");
+ }
+
+ // 3 sleep
+ try {
+ Thread.sleep(Config.workload_sched_policy_interval_ms);
+ } catch (InterruptedException e) {
+ LOG.error("error happends when policy exec thread sleep");
+ }
+ }
+ }
+ };
+
+ public void start() {
+ policyExecThread.start();
+ }
+
+ public void createWorkloadSchedPolicy(CreateWorkloadSchedPolicyStmt createStmt) throws UserException {
+ String policyName = createStmt.getPolicyName();
+
+ // 1 create condition
+ List<WorkloadConditionMeta> originConditions = createStmt.getConditions();
+ List<WorkloadCondition> policyConditionList = new ArrayList<>();
+ for (WorkloadConditionMeta cm : originConditions) {
+ WorkloadCondition cond = WorkloadCondition.createWorkloadCondition(cm.metricName, cm.op,
+ cm.value);
+ policyConditionList.add(cond);
+ }
+
+ // 2 create action
+ Map<String, String> originActions = createStmt.getActions();
+ List<WorkloadAction> policyActionList = new ArrayList<>();
+ for (Map.Entry<String, String> actionEntry : originActions.entrySet()) {
+ String actionName = actionEntry.getKey();
+ String actionArgs = actionEntry.getValue();
+
+ // we need convert wgName to wgId, because wgName may change
+ if (WorkloadActionType.move_query_to_group.toString().equals(actionName)) {
+ Long wgId = Env.getCurrentEnv().getWorkloadGroupMgr().getWorkloadGroupIdByName(actionArgs);
+ if (wgId == null) {
+ throw new UserException(
+ "can not find workload group " + actionArgs + " when set workload sched policy");
+ }
+ actionEntry.setValue(wgId.toString());
+ }
+
+ WorkloadAction ret = WorkloadAction.createWorkloadAction(actionEntry.getKey(), actionEntry.getValue());
+ policyActionList.add(ret);
+ }
+ checkPolicyActionConflicts(policyActionList);
+
+ // 3 create policy
+ Map<String, String> propMap = createStmt.getProperties();
+ if (propMap == null) {
+ propMap = new HashMap<>();
+ }
+ if (propMap.size() != 0) {
+ checkProperties(propMap);
+ }
+ writeLock();
+ try {
+ if (nameToPolicy.containsKey(createStmt.getPolicyName())) {
+ if (createStmt.isIfNotExists()) {
+ return;
+ } else {
+ throw new UserException("workload schedule policy " + policyName + " already exists ");
+ }
+ }
+ long id = Env.getCurrentEnv().getNextId();
+ WorkloadSchedPolicy policy = new WorkloadSchedPolicy(id, policyName,
+ policyConditionList, policyActionList, propMap);
+ policy.setConditionMeta(originConditions);
+ policy.setActionMeta(originActions);
+ // Env.getCurrentEnv().getEditLog().logCreateWorkloadSchedPolicy(policy);
+ idToPolicy.put(id, policy);
+ nameToPolicy.put(policyName, policy);
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ private void checkPolicyActionConflicts(List<WorkloadAction> actionList) throws UserException {
+ Set<WorkloadActionType> actionTypeSet = new HashSet<>();
+ for (WorkloadAction action : actionList) {
+ if (!actionTypeSet.add(action.getWorkloadActionType())) {
+ throw new UserException("duplicate action in one policy");
+ }
+ }
+ if (actionTypeSet.contains(WorkloadActionType.cancel_query) && actionTypeSet.contains(
+ WorkloadActionType.move_query_to_group)) {
+ throw new UserException(String.format("%s and %s can not exist in one policy at same time",
+ WorkloadActionType.cancel_query, WorkloadActionType.move_query_to_group));
+ }
+ }
+
+ public void execPolicy(List<WorkloadQueryInfo> queryInfoList) {
+ // 1 get a snapshot of policy
+ Set<Long> policyIdSet = new HashSet<>();
+ readLock();
+ try {
+ policyIdSet.addAll(idToPolicy.keySet());
+ } finally {
+ readUnlock();
+ }
+
+ for (WorkloadQueryInfo queryInfo : queryInfoList) {
+ try {
+ // 1 check policy is match
+ Map<WorkloadActionType, Queue<WorkloadSchedPolicy>> matchedPolicyMap = Maps.newHashMap();
+ for (Long policyId : policyIdSet) {
+ WorkloadSchedPolicy policy = idToPolicy.get(policyId);
+ if (policy == null) {
+ continue;
+ }
+ if (policy.isEnabled() && policy.isMatch(queryInfo)) {
+ WorkloadActionType actionType = policy.getFirstActionType();
+ // add to priority queue
+ Queue<WorkloadSchedPolicy> queue = matchedPolicyMap.get(actionType);
+ if (queue == null) {
+ queue = new PriorityQueue<>(policyComparator);
+ matchedPolicyMap.put(actionType, queue);
+ }
+ queue.offer(policy);
Review Comment:
不过我感觉可能得加一个action粒度的去重，因为引入session variable之后，set session的action其实可以重复了
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28443:
URL: https://github.com/apache/doris/pull/28443#issuecomment-1862280989
PR approved by at least one committer and no changes requested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1427568004
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/PolicyMetricType.java:
##########
@@ -0,0 +1,22 @@
+// 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.workloadschedpolicy;
+
+public enum PolicyMetricType {
Review Comment:
这个metric type，不能是一个enum，它得是一个class，我们要为一个MetricType 定义很多属性，比如：
1. 他的类型
2. 他起作用的阶段，比如是query 刚发来就做，【username 匹配这种就是】，query logical plan 解析完， physical plan 生成了
3. 未来可能还有别的
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1427632985
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/CancelQueryAction.java:
##########
@@ -0,0 +1,36 @@
+// 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.workloadschedpolicy;
+
+public class CancelQueryAction implements PolicyAction {
Review Comment:
👌🏻
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1427663661
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/CancelQueryAction.java:
##########
@@ -0,0 +1,36 @@
+// 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.workloadschedpolicy;
+
+public class CancelQueryAction implements PolicyAction {
+ @Override
+ public void exec(PolicyQueryInfo queryInfo) {
+ if (queryInfo.context != null) {
+ queryInfo.context.cancelQuery();
+ }
+ }
+
+ public static CancelQueryAction createCancelQueryAction() {
Review Comment:
那个感觉加的类太多了，这个时候还得是一个if else比较快。而且有了工厂模式也还是得有个根据conditionname去判断取哪个factory的过程
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #28443:
URL: https://github.com/apache/doris/pull/28443#issuecomment-1862074533
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1430888615
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicy.java:
##########
@@ -0,0 +1,205 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonPostProcessable;
+import org.apache.doris.persist.gson.GsonUtils;
+
+import com.esotericsoftware.minlog.Log;
+import com.google.common.collect.ImmutableSet;
+import com.google.gson.annotations.SerializedName;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class WorkloadSchedPolicy implements Writable, GsonPostProcessable {
+
+ public static final String ENABLED = "enabled";
+ public static final String PRIORITY = "priority";
+ public static final ImmutableSet<String> POLICY_PROPERTIES = new ImmutableSet.Builder<String>()
+ .add(ENABLED).add(PRIORITY).build();
+
+ @SerializedName(value = "id")
+ long id;
+ @SerializedName(value = "name")
+ String name;
+ @SerializedName(value = "version")
+ int version;
+
+ @SerializedName(value = "enabled")
+ private volatile boolean enabled;
+ @SerializedName(value = "priority")
+ private volatile int priority;
+
+ @SerializedName(value = "conditionMetaList")
+ List<WorkloadConditionMeta> conditionMetaList;
+ // we regard action as a command, map's key is command, map's value is args, so it's a command list
+ @SerializedName(value = "actionMetaList")
+ List<WorkloadActionMeta> actionMetaList;
+
+ private List<WorkloadCondition> workloadConditionList;
+ private List<WorkloadAction> workloadActionList;
+
+ public WorkloadSchedPolicy(long id, String name, List<WorkloadCondition> workloadConditionList,
+ List<WorkloadAction> workloadActionList, Map<String, String> properties) throws UserException {
+ this.id = id;
+ this.name = name;
+ this.workloadConditionList = workloadConditionList;
+ this.workloadActionList = workloadActionList;
+ // set enable and priority
+ parseAndSetProperties(properties);
+ this.version = 0;
+ }
+
+ // return true, this means all conditions in policy can match queryInfo's data
+ // return false,
+ // 1 metric not match
+ // 2 condition value not match query info's value
+ boolean isMatch(WorkloadQueryInfo queryInfo) {
+ boolean ret = true;
+ for (WorkloadCondition condition : workloadConditionList) {
+ WorkloadMetricType metricType = condition.getMetricType();
+ String value = queryInfo.metricMap.get(metricType);
+ if (value == null) {
+ return false; // query info's metric must match all condition's metric
+ }
+ ret = ret & condition.eval(value);
Review Comment:
if (!condition.eval(value)) return false;
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429633455
##########
fe/fe-core/src/main/java/org/apache/doris/common/publish/WorkloadMoveActionPublisherThread.java:
##########
@@ -0,0 +1,122 @@
+// 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.common.publish;
+
+import org.apache.doris.common.ClientPool;
+import org.apache.doris.common.ThreadPoolManager;
+import org.apache.doris.common.util.Daemon;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.BackendService;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TPublishTopicRequest;
+import org.apache.doris.thrift.TTopicInfoType;
+import org.apache.doris.thrift.TWorkloadMoveQueryToGroupAction;
+import org.apache.doris.thrift.TopicInfo;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+public class WorkloadMoveActionPublisherThread extends Daemon {
Review Comment:
那我觉得 也不能move action 启动一个thread，每个action 都搞一个trhead 太废了，然后别人想写workload group的代码也很费劲。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "yiguolei (via GitHub)" <gi...@apache.org>.
yiguolei commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429637175
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadConditionQueryTime.java:
##########
@@ -0,0 +1,52 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.common.UserException;
+
+public class WorkloadConditionQueryTime implements WorkloadCondition {
+
+ private long value;
+ private WorkloadConditionOperator op;
+
+ public WorkloadConditionQueryTime(WorkloadConditionOperator op, long value) {
+ this.op = op;
+ this.value = value;
+ }
+
+ @Override
+ public boolean eval(String strValue) {
+ long inputLongValue = Long.parseLong(strValue);
+ return WorkloadConditionCompareUtils.compareInteger(op, inputLongValue, value);
+ }
+
+ public static WorkloadConditionQueryTime createWorkloadCondition(WorkloadConditionOperator op, String value)
+ throws UserException {
+ long longValue = Long.parseLong(value);
+ if (longValue < 0) {
+ throw new UserException("invalid query time value, " + longValue + ", it requires >= 0");
+ }
+ return new WorkloadConditionQueryTime(op, longValue);
+ }
+
+ @Override
+ public WorkloadMetricType getMetricType() {
+ return WorkloadMetricType.query_time;
Review Comment:
java 里的枚举一般都是全部大写字母
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feature][execturo]support query schedule policy [doris]
Posted by "wangbo (via GitHub)" <gi...@apache.org>.
wangbo commented on code in PR #28443:
URL: https://github.com/apache/doris/pull/28443#discussion_r1429753869
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java:
##########
@@ -0,0 +1,509 @@
+// 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.workloadschedpolicy;
+
+import org.apache.doris.analysis.AlterWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.CreateWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.DropWorkloadSchedPolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.proc.BaseProcResult;
+import org.apache.doris.common.proc.ProcResult;
+import org.apache.doris.common.util.DebugUtil;
+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.service.ExecuteEnv;
+
+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.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
+
+ private static final Logger LOG = LogManager.getLogger(WorkloadSchedPolicyMgr.class);
+
+ @SerializedName(value = "idToPolicy")
+ private Map<Long, WorkloadSchedPolicy> idToPolicy = Maps.newConcurrentMap();
+ private Map<String, WorkloadSchedPolicy> nameToPolicy = Maps.newHashMap();
+
+ private PolicyProcNode policyProcNode = new PolicyProcNode();
+
+ public static final ImmutableList<String> WORKLOAD_SCHED_POLICY_NODE_TITLE_NAMES
+ = new ImmutableList.Builder<String>()
+ .add("Id").add("Name").add("ItemName").add("ItemValue")
+ .build();
+
+ private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+
+ public static Comparator<WorkloadSchedPolicy> policyComparator = new Comparator<WorkloadSchedPolicy>() {
+ @Override
+ public int compare(WorkloadSchedPolicy p1, WorkloadSchedPolicy p2) {
+ return p2.getPriority() - p1.getPriority();
+ }
+ };
+
+ private Thread policyExecThread = new Thread() {
+ @Override
+ public void run() {
+ while (true) {
+ try {
+ // 1 get query info map
+ Map<Integer, ConnectContext> connectMap = ExecuteEnv.getInstance().getScheduler()
+ .getConnectionMap();
+ List<WorkloadQueryInfo> queryInfoList = new ArrayList<>();
+
+ // a snapshot for connect context
+ Set<Integer> keySet = new HashSet<>();
+ keySet.addAll(connectMap.keySet());
+
+ for (Integer connectId : keySet) {
+ ConnectContext cctx = connectMap.get(connectId);
+ if (cctx == null) {
+ continue;
+ }
+
+ String username = cctx.getQualifiedUser();
+ long queryTime = System.currentTimeMillis() - cctx.getStartTime();
+
+ WorkloadQueryInfo policyQueryInfo = new WorkloadQueryInfo();
+ policyQueryInfo.queryId = DebugUtil.printId(cctx.queryId());
+ policyQueryInfo.tUniqueId = cctx.queryId();
+ policyQueryInfo.context = cctx;
+ policyQueryInfo.metricMap = new HashMap<>();
+ policyQueryInfo.metricMap.put(WorkloadMetricType.username, username);
+ policyQueryInfo.metricMap.put(WorkloadMetricType.query_time, String.valueOf(queryTime));
+
+ queryInfoList.add(policyQueryInfo);
+ }
+
+ // 2 exec policy
+ if (queryInfoList.size() > 0) {
+ execPolicy(queryInfoList);
+ }
+ } catch (Throwable t) {
+ LOG.error("[policy thread]error happens when exec policy");
+ }
+
+ // 3 sleep
+ try {
+ Thread.sleep(Config.workload_sched_policy_interval_ms);
+ } catch (InterruptedException e) {
+ LOG.error("error happends when policy exec thread sleep");
+ }
+ }
+ }
+ };
+
+ public void start() {
+ policyExecThread.start();
+ }
+
+ public void createWorkloadSchedPolicy(CreateWorkloadSchedPolicyStmt createStmt) throws UserException {
+ String policyName = createStmt.getPolicyName();
+
+ // 1 create condition
+ List<WorkloadConditionMeta> originConditions = createStmt.getConditions();
+ List<WorkloadCondition> policyConditionList = new ArrayList<>();
+ for (WorkloadConditionMeta cm : originConditions) {
+ WorkloadCondition cond = WorkloadCondition.createWorkloadCondition(cm.metricName, cm.op,
+ cm.value);
+ policyConditionList.add(cond);
+ }
+
+ // 2 create action
+ Map<String, String> originActions = createStmt.getActions();
+ List<WorkloadAction> policyActionList = new ArrayList<>();
+ for (Map.Entry<String, String> actionEntry : originActions.entrySet()) {
+ String actionName = actionEntry.getKey();
+ String actionArgs = actionEntry.getValue();
+
+ // we need convert wgName to wgId, because wgName may change
+ if (WorkloadActionType.move_query_to_group.toString().equals(actionName)) {
+ Long wgId = Env.getCurrentEnv().getWorkloadGroupMgr().getWorkloadGroupIdByName(actionArgs);
+ if (wgId == null) {
+ throw new UserException(
+ "can not find workload group " + actionArgs + " when set workload sched policy");
+ }
+ actionEntry.setValue(wgId.toString());
+ }
+
+ WorkloadAction ret = WorkloadAction.createWorkloadAction(actionEntry.getKey(), actionEntry.getValue());
+ policyActionList.add(ret);
+ }
+ checkPolicyActionConflicts(policyActionList);
+
+ // 3 create policy
+ Map<String, String> propMap = createStmt.getProperties();
+ if (propMap == null) {
+ propMap = new HashMap<>();
+ }
+ if (propMap.size() != 0) {
+ checkProperties(propMap);
+ }
+ writeLock();
+ try {
+ if (nameToPolicy.containsKey(createStmt.getPolicyName())) {
+ if (createStmt.isIfNotExists()) {
+ return;
+ } else {
+ throw new UserException("workload schedule policy " + policyName + " already exists ");
+ }
+ }
+ long id = Env.getCurrentEnv().getNextId();
+ WorkloadSchedPolicy policy = new WorkloadSchedPolicy(id, policyName,
+ policyConditionList, policyActionList, propMap);
+ policy.setConditionMeta(originConditions);
+ policy.setActionMeta(originActions);
+ // Env.getCurrentEnv().getEditLog().logCreateWorkloadSchedPolicy(policy);
+ idToPolicy.put(id, policy);
+ nameToPolicy.put(policyName, policy);
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ private void checkPolicyActionConflicts(List<WorkloadAction> actionList) throws UserException {
+ Set<WorkloadActionType> actionTypeSet = new HashSet<>();
+ for (WorkloadAction action : actionList) {
+ if (!actionTypeSet.add(action.getWorkloadActionType())) {
+ throw new UserException("duplicate action in one policy");
+ }
+ }
+ if (actionTypeSet.contains(WorkloadActionType.cancel_query) && actionTypeSet.contains(
+ WorkloadActionType.move_query_to_group)) {
+ throw new UserException(String.format("%s and %s can not exist in one policy at same time",
+ WorkloadActionType.cancel_query, WorkloadActionType.move_query_to_group));
+ }
+ }
+
+ public void execPolicy(List<WorkloadQueryInfo> queryInfoList) {
+ // 1 get a snapshot of policy
+ Set<Long> policyIdSet = new HashSet<>();
+ readLock();
+ try {
+ policyIdSet.addAll(idToPolicy.keySet());
+ } finally {
+ readUnlock();
+ }
+
+ for (WorkloadQueryInfo queryInfo : queryInfoList) {
+ try {
+ // 1 check policy is match
+ Map<WorkloadActionType, Queue<WorkloadSchedPolicy>> matchedPolicyMap = Maps.newHashMap();
+ for (Long policyId : policyIdSet) {
+ WorkloadSchedPolicy policy = idToPolicy.get(policyId);
+ if (policy == null) {
+ continue;
+ }
+ if (policy.isEnabled() && policy.isMatch(queryInfo)) {
+ WorkloadActionType actionType = policy.getFirstActionType();
+ // add to priority queue
+ Queue<WorkloadSchedPolicy> queue = matchedPolicyMap.get(actionType);
+ if (queue == null) {
+ queue = new PriorityQueue<>(policyComparator);
+ matchedPolicyMap.put(actionType, queue);
+ }
+ queue.offer(policy);
Review Comment:
这会个是根据priority对policy排序的话，刚测过了，设置并发的两个规则，优先级更高的会生效
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org