You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2024/02/27 02:15:01 UTC
(doris) 06/15: [Improment](executor)Add scanbytes/scanrows condition (#31364)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 1127b0065a8a5a864889b592a9697bf1c1789cc0
Author: wangbo <wa...@apache.org>
AuthorDate: Mon Feb 26 19:01:35 2024 +0800
[Improment](executor)Add scanbytes/scanrows condition (#31364)
* Add scanbytes/scanrows condition
* fix reg
---
be/src/agent/workload_sched_policy_listener.cpp | 4 +-
be/src/runtime/fragment_mgr.cpp | 7 --
be/src/runtime/runtime_query_statistics_mgr.cpp | 19 ++++
be/src/runtime/runtime_query_statistics_mgr.h | 8 ++
.../runtime/workload_management/workload_action.h | 4 +-
.../workload_management/workload_condition.h | 4 +-
.../workload_sched_policy_mgr.cpp | 5 +
.../main/java/org/apache/doris/common/Config.java | 3 -
.../main/java/org/apache/doris/catalog/Env.java | 6 -
.../publish/WorkloadActionPublishThread.java | 123 ---------------------
.../workloadschedpolicy/WorkloadAction.java | 2 -
.../WorkloadActionMoveQueryToGroup.java | 67 -----------
.../workloadschedpolicy/WorkloadCondition.java | 4 +
...tion.java => WorkloadConditionBeScanBytes.java} | 32 ++++--
...ition.java => WorkloadConditionBeScanRows.java} | 32 ++++--
.../workloadschedpolicy/WorkloadConditionMeta.java | 4 +
.../workloadschedpolicy/WorkloadMetricType.java | 2 +-
.../workloadschedpolicy/WorkloadSchedPolicy.java | 4 +-
.../WorkloadSchedPolicyMgr.java | 22 +---
gensrc/thrift/BackendService.thrift | 12 +-
.../test_workload_sched_policy.out | 1 -
.../test_workload_sched_policy.groovy | 21 +---
22 files changed, 101 insertions(+), 285 deletions(-)
diff --git a/be/src/agent/workload_sched_policy_listener.cpp b/be/src/agent/workload_sched_policy_listener.cpp
index 461fd2cbb0f..689a600367a 100644
--- a/be/src/agent/workload_sched_policy_listener.cpp
+++ b/be/src/agent/workload_sched_policy_listener.cpp
@@ -69,9 +69,7 @@ void WorkloadschedPolicyListener::handle_topic_info(const std::vector<TopicInfo>
policy_map.emplace(tpolicy.id, std::move(policy_ptr));
}
size_t new_policy_size = policy_map.size();
- if (new_policy_size > 0) {
- _exec_env->workload_sched_policy_mgr()->update_workload_sched_policy(std::move(policy_map));
- }
+ _exec_env->workload_sched_policy_mgr()->update_workload_sched_policy(std::move(policy_map));
LOG(INFO) << "[workload_schedule]finish update workload schedule policy, size="
<< new_policy_size;
}
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index f6985538854..e52b71d277b 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -1506,17 +1506,10 @@ void FragmentMgr::_setup_shared_hashtable_for_broadcast_join(const TPipelineFrag
void FragmentMgr::get_runtime_query_info(std::vector<WorkloadQueryInfo>* query_info_list) {
{
std::lock_guard<std::mutex> lock(_lock);
- // todo: use monotonic time
- VecDateTimeValue now = VecDateTimeValue::local_time();
for (const auto& q : _query_ctx_map) {
WorkloadQueryInfo workload_query_info;
workload_query_info.query_id = print_id(q.first);
workload_query_info.tquery_id = q.first;
-
- uint64_t query_time_millisecond = q.second->query_time(now) * 1000;
- workload_query_info.metric_map.emplace(WorkloadMetricType::QUERY_TIME,
- std::to_string(query_time_millisecond));
- // todo, add scan rows, scan bytes
query_info_list->push_back(workload_query_info);
}
}
diff --git a/be/src/runtime/runtime_query_statistics_mgr.cpp b/be/src/runtime/runtime_query_statistics_mgr.cpp
index a658e527f61..c6f70643f13 100644
--- a/be/src/runtime/runtime_query_statistics_mgr.cpp
+++ b/be/src/runtime/runtime_query_statistics_mgr.cpp
@@ -172,6 +172,25 @@ std::shared_ptr<QueryStatistics> RuntimeQueryStatiticsMgr::get_runtime_query_sta
return qs_ptr;
}
+void RuntimeQueryStatiticsMgr::get_metric_map(
+ std::string query_id, std::map<WorkloadMetricType, std::string>& metric_map) {
+ QueryStatistics ret_qs;
+ int64_t query_time_ms = 0;
+ {
+ std::shared_lock<std::shared_mutex> read_lock(_qs_ctx_map_lock);
+ if (_query_statistics_ctx_map.find(query_id) != _query_statistics_ctx_map.end()) {
+ for (auto const& qs : _query_statistics_ctx_map[query_id]->_qs_list) {
+ ret_qs.merge(*qs);
+ }
+ query_time_ms =
+ MonotonicMillis() - _query_statistics_ctx_map.at(query_id)->_query_start_time;
+ }
+ }
+ metric_map.emplace(WorkloadMetricType::QUERY_TIME, std::to_string(query_time_ms));
+ metric_map.emplace(WorkloadMetricType::SCAN_ROWS, std::to_string(ret_qs.get_scan_rows()));
+ metric_map.emplace(WorkloadMetricType::SCAN_BYTES, std::to_string(ret_qs.get_scan_bytes()));
+}
+
void RuntimeQueryStatiticsMgr::set_workload_group_id(std::string query_id, int64_t wg_id) {
// wg id just need eventual consistency, read lock is ok
std::shared_lock<std::shared_mutex> read_lock(_qs_ctx_map_lock);
diff --git a/be/src/runtime/runtime_query_statistics_mgr.h b/be/src/runtime/runtime_query_statistics_mgr.h
index 98d4f554728..69b283b6d14 100644
--- a/be/src/runtime/runtime_query_statistics_mgr.h
+++ b/be/src/runtime/runtime_query_statistics_mgr.h
@@ -21,6 +21,8 @@
#include <string>
#include "runtime/query_statistics.h"
+#include "runtime/workload_management/workload_condition.h"
+#include "util/time.h"
namespace doris {
@@ -29,6 +31,7 @@ public:
QueryStatisticsCtx(TNetworkAddress fe_addr) : _fe_addr(fe_addr) {
this->_is_query_finished = false;
this->_wg_id = -1;
+ this->_query_start_time = MonotonicMillis();
}
~QueryStatisticsCtx() = default;
@@ -40,6 +43,7 @@ public:
TNetworkAddress _fe_addr;
int64_t _query_finish_time;
int64_t _wg_id;
+ int64_t _query_start_time;
};
class RuntimeQueryStatiticsMgr {
@@ -58,6 +62,10 @@ public:
void set_workload_group_id(std::string query_id, int64_t wg_id);
+ // used for workload scheduler policy
+ void get_metric_map(std::string query_id,
+ std::map<WorkloadMetricType, std::string>& metric_map);
+
private:
std::shared_mutex _qs_ctx_map_lock;
std::map<std::string, std::unique_ptr<QueryStatisticsCtx>> _query_statistics_ctx_map;
diff --git a/be/src/runtime/workload_management/workload_action.h b/be/src/runtime/workload_management/workload_action.h
index 29c01320b78..785acc73c3a 100644
--- a/be/src/runtime/workload_management/workload_action.h
+++ b/be/src/runtime/workload_management/workload_action.h
@@ -55,9 +55,7 @@ private:
class WorkloadActionFactory {
public:
static std::unique_ptr<WorkloadAction> create_workload_action(TWorkloadAction* action) {
- if (TWorkloadActionType::type::MOVE_QUERY_TO_GROUP == action->action) {
- return std::make_unique<WorkloadActionMoveQuery>(action->action_args);
- } else if (TWorkloadActionType::type::CANCEL_QUERY == action->action) {
+ if (TWorkloadActionType::type::CANCEL_QUERY == action->action) {
return std::make_unique<WorkloadActionCancelQuery>();
}
LOG(ERROR) << "not find a action " << action->action;
diff --git a/be/src/runtime/workload_management/workload_condition.h b/be/src/runtime/workload_management/workload_condition.h
index 3486742c968..96387a2af41 100644
--- a/be/src/runtime/workload_management/workload_condition.h
+++ b/be/src/runtime/workload_management/workload_condition.h
@@ -84,9 +84,9 @@ public:
TWorkloadMetricType::type metric_name = t_cond->metric_name;
if (TWorkloadMetricType::type::QUERY_TIME == metric_name) {
return std::make_unique<WorkloadConditionQueryTime>(op, str_val);
- } else if (TWorkloadMetricType::type::SCAN_ROWS == metric_name) {
+ } else if (TWorkloadMetricType::type::BE_SCAN_ROWS == metric_name) {
return std::make_unique<WorkloadConditionScanRows>(op, str_val);
- } else if (TWorkloadMetricType::type::SCAN_BYTES == metric_name) {
+ } else if (TWorkloadMetricType::type::BE_SCAN_BYTES == metric_name) {
return std::make_unique<WorkloadConditionScanBytes>(op, str_val);
}
LOG(ERROR) << "not find a metric name " << metric_name;
diff --git a/be/src/runtime/workload_management/workload_sched_policy_mgr.cpp b/be/src/runtime/workload_management/workload_sched_policy_mgr.cpp
index 731dd0c8661..c41a9e723e3 100644
--- a/be/src/runtime/workload_management/workload_sched_policy_mgr.cpp
+++ b/be/src/runtime/workload_management/workload_sched_policy_mgr.cpp
@@ -18,6 +18,7 @@
#include "runtime/workload_management/workload_sched_policy_mgr.h"
#include "runtime/fragment_mgr.h"
+#include "runtime/runtime_query_statistics_mgr.h"
namespace doris {
@@ -76,6 +77,7 @@ void WorkloadSchedPolicyMgr::_schedule_workload() {
while (!_stop_latch.wait_for(std::chrono::milliseconds(500))) {
// 1 get query info
std::vector<WorkloadQueryInfo> list;
+ //todo(wb) maybe we can get runtime queryinfo from RuntimeQueryStatiticsMgr directly
_exec_env->fragment_mgr()->get_runtime_query_info(&list);
// todo: add timer
if (list.size() == 0) {
@@ -84,6 +86,9 @@ void WorkloadSchedPolicyMgr::_schedule_workload() {
for (int i = 0; i < list.size(); i++) {
WorkloadQueryInfo* query_info_ptr = &(list[i]);
+ _exec_env->runtime_query_statistics_mgr()->get_metric_map(query_info_ptr->query_id,
+ query_info_ptr->metric_map);
+
// 2 get matched policy
std::map<WorkloadActionType, std::shared_ptr<WorkloadSchedPolicy>> matched_policy_map;
{
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 541592f2095..0d89becbb28 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -2391,9 +2391,6 @@ public class Config extends ConfigBase {
@ConfField(mutable = true)
public static int workload_sched_policy_interval_ms = 10000; // 10s
- @ConfField(mutable = true)
- public static int workload_action_interval_ms = 10000; // 10s
-
@ConfField(mutable = true, masterOnly = true)
public static int workload_max_policy_num = 25;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
index 8c65a71dafa..76ff70589b9 100755
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
@@ -111,7 +111,6 @@ import org.apache.doris.common.io.CountingDataOutputStream;
import org.apache.doris.common.io.Text;
import org.apache.doris.common.publish.TopicPublisher;
import org.apache.doris.common.publish.TopicPublisherThread;
-import org.apache.doris.common.publish.WorkloadActionPublishThread;
import org.apache.doris.common.publish.WorkloadGroupPublisher;
import org.apache.doris.common.util.Daemon;
import org.apache.doris.common.util.DynamicPartitionUtil;
@@ -522,8 +521,6 @@ public class Env {
private TopicPublisherThread topicPublisherThread;
- private WorkloadActionPublishThread workloadActionPublisherThread;
-
private MTMVService mtmvService;
private InsertOverwriteManager insertOverwriteManager;
@@ -761,8 +758,6 @@ public class Env {
this.queryCancelWorker = new QueryCancelWorker(systemInfo);
this.topicPublisherThread = new TopicPublisherThread(
"TopicPublisher", Config.publish_topic_info_interval_ms, systemInfo);
- this.workloadActionPublisherThread = new WorkloadActionPublishThread("WorkloadActionPublisher",
- Config.workload_action_interval_ms, systemInfo);
this.mtmvService = new MTMVService();
this.insertOverwriteManager = new InsertOverwriteManager();
}
@@ -1042,7 +1037,6 @@ public class Env {
workloadGroupMgr.startUpdateThread();
workloadSchedPolicyMgr.start();
- workloadActionPublisherThread.start();
workloadRuntimeStatusMgr.start();
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/publish/WorkloadActionPublishThread.java b/fe/fe-core/src/main/java/org/apache/doris/common/publish/WorkloadActionPublishThread.java
deleted file mode 100644
index cacd7a9da9f..00000000000
--- a/fe/fe-core/src/main/java/org/apache/doris/common/publish/WorkloadActionPublishThread.java
+++ /dev/null
@@ -1,123 +0,0 @@
-// 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));
- }
- }
-
- public class WorkloadMoveActionTask implements Runnable {
-
- private TPublishTopicRequest request;
-
- private Backend be;
-
- private ResponseHandler handler;
-
- public WorkloadMoveActionTask(TPublishTopicRequest request, Backend be,
- ResponseHandler handler) {
- this.request = request;
- this.be = be;
- this.handler = handler;
- }
-
- @Override
- public void run() {
- long beginTime = System.currentTimeMillis();
- try {
- TNetworkAddress addr = new TNetworkAddress(be.getHost(), be.getBePort());
- BackendService.Client client = ClientPool.backendPool.borrowObject(addr);
- client.publishTopicInfo(request);
- LOG.info("publish move action topic to be {} success, time cost={} ms",
- be.getHost(), (System.currentTimeMillis() - beginTime));
- } catch (Exception e) {
- LOG.warn("publish move action topic to be {} error happens: , time cost={} ms",
- be.getHost(), (System.currentTimeMillis() - beginTime), e);
- } finally {
- handler.onResponse(be);
- }
- }
- }
-}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadAction.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadAction.java
index d9298781476..661ea6a45fa 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadAction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadAction.java
@@ -30,8 +30,6 @@ public interface WorkloadAction {
throws UserException {
if (WorkloadActionType.CANCEL_QUERY.equals(workloadActionMeta.action)) {
return WorkloadActionCancelQuery.createWorkloadAction();
- } else if (WorkloadActionType.MOVE_QUERY_TO_GROUP.equals(workloadActionMeta.action)) {
- return WorkloadActionMoveQueryToGroup.createWorkloadAction(workloadActionMeta.actionArgs);
} else if (WorkloadActionType.SET_SESSION_VARIABLE.equals(workloadActionMeta.action)) {
return WorkloadActionSetSessionVar.createWorkloadAction(workloadActionMeta.actionArgs);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadActionMoveQueryToGroup.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadActionMoveQueryToGroup.java
deleted file mode 100644
index 59e09345fa9..00000000000
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadActionMoveQueryToGroup.java
+++ /dev/null
@@ -1,67 +0,0 @@
-// 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.publish.WorkloadActionPublishThread;
-import org.apache.doris.qe.QeProcessorImpl;
-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;
-
-public class WorkloadActionMoveQueryToGroup implements WorkloadAction {
-
- private static final Logger LOG = LogManager.getLogger(WorkloadActionMoveQueryToGroup.class);
-
- private long dstWgId;
-
- public WorkloadActionMoveQueryToGroup(long dstWgId) {
- this.dstWgId = dstWgId;
- }
-
- @Override
- public void exec(WorkloadQueryInfo queryInfo) {
- if (queryInfo.context != null && !queryInfo.context.isKilled()
- && queryInfo.tUniqueId != null
- && QeProcessorImpl.INSTANCE.getCoordinator(queryInfo.tUniqueId) != null) {
- LOG.info("try move query {} to group {}", queryInfo.queryId, dstWgId);
-
- TWorkloadMoveQueryToGroupAction moveQueryToGroupAction = new TWorkloadMoveQueryToGroupAction();
- moveQueryToGroupAction.setQueryId(queryInfo.tUniqueId);
- moveQueryToGroupAction.setWorkloadGroupId(dstWgId);
-
- TopicInfo topicInfo = new TopicInfo();
- topicInfo.setMoveAction(moveQueryToGroupAction);
-
- WorkloadActionPublishThread.putWorkloadAction(TTopicInfoType.MOVE_QUERY_TO_GROUP, topicInfo);
- }
- }
-
- @Override
- public WorkloadActionType getWorkloadActionType() {
- return WorkloadActionType.MOVE_QUERY_TO_GROUP;
- }
-
- public static WorkloadActionMoveQueryToGroup createWorkloadAction(String groupId) {
- long wgId = Long.parseLong(groupId);
- return new WorkloadActionMoveQueryToGroup(wgId);
- }
-
-}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadCondition.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadCondition.java
index 1f75d81794f..5d89d2afae9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadCondition.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadCondition.java
@@ -33,6 +33,10 @@ public interface WorkloadCondition {
return WorkloadConditionUsername.createWorkloadCondition(cm.op, cm.value);
} else if (WorkloadMetricType.QUERY_TIME.equals(cm.metricName)) {
return WorkloadConditionQueryTime.createWorkloadCondition(cm.op, cm.value);
+ } else if (WorkloadMetricType.BE_SCAN_ROWS.equals(cm.metricName)) {
+ return WorkloadConditionBeScanRows.createWorkloadCondition(cm.op, cm.value);
+ } else if (WorkloadMetricType.BE_SCAN_BYTES.equals(cm.metricName)) {
+ return WorkloadConditionBeScanBytes.createWorkloadCondition(cm.op, cm.value);
}
throw new UserException("invalid metric name:" + cm.metricName);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadCondition.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadConditionBeScanBytes.java
similarity index 51%
copy from fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadCondition.java
copy to fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadConditionBeScanBytes.java
index 1f75d81794f..7431f2e0c4f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadCondition.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadConditionBeScanBytes.java
@@ -19,22 +19,34 @@ package org.apache.doris.resource.workloadschedpolicy;
import org.apache.doris.common.UserException;
+public class WorkloadConditionBeScanBytes implements WorkloadCondition {
-public interface WorkloadCondition {
+ private long value;
- boolean eval(String strValue);
+ private WorkloadConditionOperator op;
- WorkloadMetricType getMetricType();
+ public WorkloadConditionBeScanBytes(WorkloadConditionOperator op, long value) {
+ this.op = op;
+ this.value = value;
+ }
+
+ @Override
+ public boolean eval(String strValue) {
+ // currently not support run in fe, so this condition never match
+ return false;
+ }
- // NOTE(wb) currently createPolicyCondition is also used when replay meta, it better not contains heavy check
- static WorkloadCondition createWorkloadCondition(WorkloadConditionMeta cm)
+ public static WorkloadConditionBeScanBytes createWorkloadCondition(WorkloadConditionOperator op, String value)
throws UserException {
- if (WorkloadMetricType.USERNAME.equals(cm.metricName)) {
- return WorkloadConditionUsername.createWorkloadCondition(cm.op, cm.value);
- } else if (WorkloadMetricType.QUERY_TIME.equals(cm.metricName)) {
- return WorkloadConditionQueryTime.createWorkloadCondition(cm.op, cm.value);
+ long longValue = Long.parseLong(value);
+ if (longValue < 0) {
+ throw new UserException("invalid scan bytes value, " + longValue + ", it requires >= 0");
}
- throw new UserException("invalid metric name:" + cm.metricName);
+ return new WorkloadConditionBeScanBytes(op, longValue);
}
+ @Override
+ public WorkloadMetricType getMetricType() {
+ return WorkloadMetricType.BE_SCAN_BYTES;
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadCondition.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadConditionBeScanRows.java
similarity index 51%
copy from fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadCondition.java
copy to fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadConditionBeScanRows.java
index 1f75d81794f..c2fb638e082 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadCondition.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadConditionBeScanRows.java
@@ -19,22 +19,34 @@ package org.apache.doris.resource.workloadschedpolicy;
import org.apache.doris.common.UserException;
+public class WorkloadConditionBeScanRows implements WorkloadCondition {
-public interface WorkloadCondition {
+ private long value;
- boolean eval(String strValue);
+ private WorkloadConditionOperator op;
- WorkloadMetricType getMetricType();
+ public WorkloadConditionBeScanRows(WorkloadConditionOperator op, long value) {
+ this.op = op;
+ this.value = value;
+ }
+
+ @Override
+ public boolean eval(String strValue) {
+ // currently not support run in fe, so this condition never match
+ return false;
+ }
- // NOTE(wb) currently createPolicyCondition is also used when replay meta, it better not contains heavy check
- static WorkloadCondition createWorkloadCondition(WorkloadConditionMeta cm)
+ public static WorkloadConditionBeScanRows createWorkloadCondition(WorkloadConditionOperator op, String value)
throws UserException {
- if (WorkloadMetricType.USERNAME.equals(cm.metricName)) {
- return WorkloadConditionUsername.createWorkloadCondition(cm.op, cm.value);
- } else if (WorkloadMetricType.QUERY_TIME.equals(cm.metricName)) {
- return WorkloadConditionQueryTime.createWorkloadCondition(cm.op, cm.value);
+ long longValue = Long.parseLong(value);
+ if (longValue < 0) {
+ throw new UserException("invalid scan rows value, " + longValue + ", it requires >= 0");
}
- throw new UserException("invalid metric name:" + cm.metricName);
+ return new WorkloadConditionBeScanRows(op, longValue);
}
+ @Override
+ public WorkloadMetricType getMetricType() {
+ return WorkloadMetricType.BE_SCAN_ROWS;
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadConditionMeta.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadConditionMeta.java
index d5d2f922f3f..52f50f924fc 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadConditionMeta.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadConditionMeta.java
@@ -44,6 +44,10 @@ public class WorkloadConditionMeta {
return WorkloadMetricType.USERNAME;
} else if (WorkloadMetricType.QUERY_TIME.toString().equalsIgnoreCase(metricStr)) {
return WorkloadMetricType.QUERY_TIME;
+ } else if (WorkloadMetricType.BE_SCAN_ROWS.toString().equalsIgnoreCase(metricStr)) {
+ return WorkloadMetricType.BE_SCAN_ROWS;
+ } else if (WorkloadMetricType.BE_SCAN_BYTES.toString().equalsIgnoreCase(metricStr)) {
+ return WorkloadMetricType.BE_SCAN_BYTES;
}
throw new UserException("invalid metric name:" + metricStr);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadMetricType.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadMetricType.java
index f81d75c675f..ed17414ec45 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadMetricType.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadMetricType.java
@@ -18,5 +18,5 @@
package org.apache.doris.resource.workloadschedpolicy;
public enum WorkloadMetricType {
- USERNAME, QUERY_TIME, SCAN_ROWS, SCAN_BYTES
+ USERNAME, QUERY_TIME, BE_SCAN_ROWS, BE_SCAN_BYTES
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicy.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicy.java
index 13637b5b8be..8b028e3e75f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicy.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicy.java
@@ -52,7 +52,9 @@ public class WorkloadSchedPolicy implements Writable, GsonPostProcessable {
// used for convert fe type to thrift type
private static ImmutableMap<WorkloadMetricType, TWorkloadMetricType> METRIC_MAP
= new ImmutableMap.Builder<WorkloadMetricType, TWorkloadMetricType>()
- .put(WorkloadMetricType.QUERY_TIME, TWorkloadMetricType.QUERY_TIME).build();
+ .put(WorkloadMetricType.QUERY_TIME, TWorkloadMetricType.QUERY_TIME)
+ .put(WorkloadMetricType.BE_SCAN_ROWS, TWorkloadMetricType.BE_SCAN_ROWS)
+ .put(WorkloadMetricType.BE_SCAN_BYTES, TWorkloadMetricType.BE_SCAN_BYTES).build();
private static ImmutableMap<WorkloadActionType, TWorkloadActionType> ACTION_MAP
= new ImmutableMap.Builder<WorkloadActionType, TWorkloadActionType>()
.put(WorkloadActionType.MOVE_QUERY_TO_GROUP, TWorkloadActionType.MOVE_QUERY_TO_GROUP)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java
index 0074020735e..5c35d1ee095 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadSchedPolicyMgr.java
@@ -86,9 +86,8 @@ public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
.add(WorkloadActionType.CANCEL_QUERY).build();
public static final ImmutableSet<WorkloadMetricType> BE_METRIC_SET
- = new ImmutableSet.Builder<WorkloadMetricType>().add(WorkloadMetricType.SCAN_ROWS)
- .add(WorkloadMetricType.SCAN_BYTES).add(WorkloadMetricType.QUERY_TIME)
- .build();
+ = new ImmutableSet.Builder<WorkloadMetricType>().add(WorkloadMetricType.BE_SCAN_ROWS)
+ .add(WorkloadMetricType.BE_SCAN_BYTES).add(WorkloadMetricType.QUERY_TIME).build();
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
@@ -122,15 +121,12 @@ public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
}
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);
}
@@ -174,19 +170,7 @@ public class WorkloadSchedPolicyMgr implements Writable, GsonPostProcessable {
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();
- }
-
+ // todo(wb) support move action
WorkloadAction ret = WorkloadAction.createWorkloadAction(workloadActionMeta);
policyActionList.add(ret);
}
diff --git a/gensrc/thrift/BackendService.thrift b/gensrc/thrift/BackendService.thrift
index c59abd65d3b..24edaefc103 100644
--- a/gensrc/thrift/BackendService.thrift
+++ b/gensrc/thrift/BackendService.thrift
@@ -179,15 +179,10 @@ struct TWorkloadGroupInfo {
9: optional i32 scan_thread_num
}
-struct TWorkloadMoveQueryToGroupAction {
- 1: optional Types.TUniqueId query_id
- 2: optional i64 workload_group_id
-}
-
enum TWorkloadMetricType {
QUERY_TIME
- SCAN_ROWS
- SCAN_BYTES
+ BE_SCAN_ROWS
+ BE_SCAN_BYTES
}
enum TCompareOperator {
@@ -226,8 +221,7 @@ struct TWorkloadSchedPolicy {
struct TopicInfo {
1: optional TWorkloadGroupInfo workload_group_info
- 2: optional TWorkloadMoveQueryToGroupAction move_action
- 3: optional TWorkloadSchedPolicy workload_sched_policy
+ 2: optional TWorkloadSchedPolicy workload_sched_policy
}
struct TPublishTopicRequest {
diff --git a/regression-test/data/workload_manager_p0/test_workload_sched_policy.out b/regression-test/data/workload_manager_p0/test_workload_sched_policy.out
index 0c86700866d..d32fff321e4 100644
--- a/regression-test/data/workload_manager_p0/test_workload_sched_policy.out
+++ b/regression-test/data/workload_manager_p0/test_workload_sched_policy.out
@@ -2,7 +2,6 @@
-- !select_policy_tvf --
be_policy query_time > 10 cancel_query 10 false 0
fe_policy username = root set_session_variable "workload_group=normal" 10 false 0
-move_action_policy query_time > 10 move_query_to_group "normal" 0 true 0
set_action_policy username = root set_session_variable "workload_group=normal" 0 true 0
test_cancel_policy query_time > 10 cancel_query 0 false 0
diff --git a/regression-test/suites/workload_manager_p0/test_workload_sched_policy.groovy b/regression-test/suites/workload_manager_p0/test_workload_sched_policy.groovy
index cabba8ee008..603bbdf520e 100644
--- a/regression-test/suites/workload_manager_p0/test_workload_sched_policy.groovy
+++ b/regression-test/suites/workload_manager_p0/test_workload_sched_policy.groovy
@@ -20,7 +20,6 @@ suite("test_workload_sched_policy") {
sql "set experimental_enable_nereids_planner = false;"
sql "drop workload schedule policy if exists test_cancel_policy;"
- sql "drop workload schedule policy if exists move_action_policy;"
sql "drop workload schedule policy if exists set_action_policy;"
sql "drop workload schedule policy if exists fe_policy;"
sql "drop workload schedule policy if exists be_policy;"
@@ -30,17 +29,12 @@ suite("test_workload_sched_policy") {
" conditions(query_time > 10) " +
" actions(cancel_query) properties('enabled'='false'); "
- // 2 create cancel policy
- sql "create workload schedule policy move_action_policy " +
- "conditions(query_time > 10) " +
- "actions(move_query_to_group 'normal');"
-
- // 3 create set policy
+ // 2 create set policy
sql "create workload schedule policy set_action_policy " +
"conditions(username='root') " +
"actions(set_session_variable 'workload_group=normal');"
- // 4 create policy run in fe
+ // 3 create policy run in fe
sql "create workload schedule policy fe_policy " +
"conditions(username='root') " +
"actions(set_session_variable 'workload_group=normal') " +
@@ -49,7 +43,7 @@ suite("test_workload_sched_policy") {
"'priority'='10' " +
");"
- // 5 create policy run in be
+ // 4 create policy run in be
sql "create workload schedule policy be_policy " +
"conditions(query_time > 10) " +
"actions(cancel_query) " +
@@ -96,14 +90,6 @@ suite("test_workload_sched_policy") {
exception "priority can only between"
}
- test {
- sql "create workload schedule policy conflict_policy " +
- "conditions (query_time > 0)" +
- "actions(cancel_query, move_query_to_group 'normal');"
-
- exception "can not exist in one policy at same time"
- }
-
test {
sql "create workload schedule policy conflict_policy " +
"conditions (query_time > 0) " +
@@ -122,7 +108,6 @@ suite("test_workload_sched_policy") {
// drop
sql "drop workload schedule policy test_cancel_policy;"
- sql "drop workload schedule policy move_action_policy;"
sql "drop workload schedule policy set_action_policy;"
sql "drop workload schedule policy fe_policy;"
sql "drop workload schedule policy be_policy;"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org