Current state: ["Accepted"]
Discussion thread: https://lists.apache.org/thread/nft6d2jptoyqnnz5lv5846fc4b6jxhh8
JIRA or Github Issue:
Released: <Doris Version>
Google Doc: <If the design in question is unclear or needs to be discussed and reviewed, a Google Doc can be used first to facilitate comments from others.>
Support for resource queue is motivated by the following points:
Large queries or jobs preempt cluster resources, resulting in small queries that cannot be completed quickly;
Unable to limit the submission of large queries. A lot of parallel large queries lead to problems such as BE OOM or full preemption of cluster resources.
The CPU resources and memory resources of a database instance are limited. These resources affect the query performance of the database. When the database load reaches a certain level, each query will compete for CPU resources and memory resources, resulting in low overall query performance.
Resource Queue: User can specify the number of concurrent queries that the database can run and the number of queries queued according to your own business. This can ensure that there are expected system resources when executing the query, so as to obtain the expected query performance.
Creating resource queues for queries is common in various database products. Aliyun AnalyticDB specifies the number of concurrent queries that the database can run, the memory size that each query can use, and the CPU resources that can be used by creating a resource queue[1]; In the process of using the cloud data warehouse PostgreSQL, a single complex query may consume too many resources and affect other users' queries or calculations. When it is necessary to limit the consumption of system resources for a single user or query statement, Tencent Cloud uses resource queues to limit[2].
The resource queue stores two types of information:
Queue configuration: describes the resource limits available for this queue, such as: Concurrency, CPU, memory, scan rows, etc.
Matching policy: After a query (such as select/insert) is submitted, a matching queue will be matched according to the job information. Matching rules can be username, IP, database name and table name.
// create resource queue
CREATE RESOURCE_QUEUE [IF NOT EXISTS] queue_name
WITH RESOURCE (
"max_concurrency" = "1", // Limit the number of queries running simultaneously in the queue, default 1.
"max_queue_size" = "10" // Limit the number of queries queued in the queue, default 10.
)
WITH MATCHING POLICY (
"user" = "rd_group*", // Match the prefix of user name, default root.
"ip" = "192.10.1.*" // Match the prefix of IP, default *.
);
// drop resource queue
DROP RESOURCE_QUEUE [IF EXISTS] queue_name;
// show resource queues
SHOW ALL RESOURCE_QUEUE;
// show specified queue: queueId, pendingNum, runningNum, queueConfig, matchingPolicy
SHOW RESOURCE_QUEUE queue_name;
// rename resource queue
ALTER RESOURCE_QUEUE queue_name RENAME new_queue_name;
// alter resource queue
ALTER RESOURCE_QUEUE queue_name SET RESOURCE (...) SET MATCHING POLICY (...); |
1. We can only rename or alter resource queue when it's empty.
2. max_concurrency >= 1 && max_queue_size >= 0, error settings will be ignored.
3. When multiple queues are matched, the best queue(pending queries are least) is returned.
// Parent class of all asynchronous job
class AsyncJob {
public void run();
public void cancel();
}
class SqlJob extends AsyncJob {
private long id;
private String label;
private Type type
private StatementBase sql;
}
class ResourceQueue {
private long id;
private ResourceQueueConfig config;
private MatchingPolicy policy;
private Queue<AsyncJob> pendingQueue;
private Queue<AsyncJob> runningQueue;
private Queue<AsyncJob> finishQueue;
public void addJob(AsyncJob job) {
pendingQueue.add(job);
editlog.write(job);
}
public void cancelJob(long jobId) {
AsyncJob job = findJobInPendingAndRunningQueu();
job.cancel();
}
public void run() {
while(true) {
AsyncJob job = pendingQueue.poll();
runningQueue.add(job);
job.run();
finishQueue.add(job);
}
}
// Scheduled task for cleaning up finished jobs
public TimerTask() {
@Override
public void run() {
removeExpireJobFromFinishQueue()
}
}
}
// Manager of resource queues
class ResourceQueueMgr {
private Map<Long, ResourceQueue> id2Queue;
public ResourceQueue matchQueue(AsyncJob job) {
// Return the matching queue according to job information
}
public List<List<String>> showQueueInfo() {
}
}
// Sumit asynchronous job
class AsyncJobManager() {
private ResourceQueueMgr queueMgr;
// jobid -> queue
private Map<Long, Long> jobId2Queue;
// label -> queue
private Map<String, Long> jobLabel2Queue;
public void submitAsyncJob(AsyncJob job) {
ResourceQueue queue = queueMgr.matchQueue(job);
queue.addJob(job);
}
public List<List<String>> showJob(long jobid) {
long queueId = jobId2Queue.get(jobId);
ResourceQueue queue = queueMgr.getQueue(queueId);
AsyncJob job = queue.getJob(jobId);
return job.getShowInfo();
}
public void cancelJob(long jobId) {
long queueId = jobId2Queue.get(jobId);
ResourceQueue queue = queueMgr.getQueue(queueId);
AsyncJob job = queue.getJob(jobId);
job.cancel();
}
}
|