f

be/src/runtime/query_context.h

@@ -325,6 +325,13 @@ class QueryContext : public std::enable_shared_from_this<QueryContext> {
         std::lock_guard l(_paused_mutex);
         if (_paused_error_status.is<ErrorCode::QUERY_MEMORY_EXCEED>()) {
             return;
+        } else if (_paused_error_status.is<ErrorCode::WORKLOAD_GROUP_MEMORY_EXCEED>()) {
+            if (st.is<ErrorCode::QUERY_MEMORY_EXCEED>()) {
+                _paused_error_status = st;
+                return;
+            } else {
+                return;
+            }
         } else {
             _paused_error_status = st;
         }

be/src/runtime/workload_group/workload_group.h

@@ -235,6 +235,10 @@ class WorkloadGroup : public std::enable_shared_from_this<WorkloadGroup> {
 
     int64_t load_buffer_limit() { return _load_buffer_limit; }
 
+    void update_memory_sufficent(bool is_sufficient) { _is_sufficient = is_sufficient; }
+
+    bool memory_sufficent() const { return _is_sufficient; }
+
 private:
     mutable std::shared_mutex _mutex; // lock _name, _version, _cpu_share, _memory_limit
     const uint64_t _id;
@@ -245,6 +249,7 @@ class WorkloadGroup : public std::enable_shared_from_this<WorkloadGroup> {
     // If the wg's memory reached high water mark, then the load buffer
     // will be restricted to this limit.
     int64_t _load_buffer_limit;
+    std::Atomic<bool> _is_sufficient = true;
 
     // memory used by load memtable
     int64_t _active_mem_usage = 0;

be/src/runtime/workload_group/workload_group_manager.cpp

@@ -273,7 +273,8 @@ void WorkloadGroupMgr::refresh_wg_weighted_memory_limit() {
             } else {
                 // If low water mark is not reached, then use process memory limit as query memory limit.
                 // It means it will not take effect.
-                if (!is_low_wartermark) {
+                // If there are some query in paused list, then limit should take effect.
+                if (!is_low_wartermark && wg.second->memory_sufficent()) {
                     query_weighted_mem_limit = wg_high_water_mark_limit;
                 } else {
                     query_weighted_mem_limit =
@@ -327,7 +328,6 @@ void WorkloadGroupMgr::add_paused_query(const std::shared_ptr<QueryContext>& que
     std::lock_guard<std::mutex> lock(_paused_queries_lock);
     DCHECK(query_ctx != nullptr);
     auto wg = query_ctx->workload_group();
-    query_ctx->update_paused_state(true);
     auto&& [it, inserted] = _paused_queries_list[wg].emplace(query_ctx);
     if (inserted) {
         LOG(INFO) << "here insert one new paused query: " << it->query_id()
@@ -386,12 +386,13 @@ void WorkloadGroupMgr::handle_paused_queries() {
                     query_ctx->cancel(doris::Status::Error<QUERY_MEMORY_EXCEED>(
                             "query reserve memory failed, but could not find  memory that could "
                             "release or spill to disk"));
-                } else
+                } else {
                     LOG(INFO) << "query: " << print_id(max_revocable_query->query_id()) << ", has "
                               << revocable_tasks.size()
                               << " tasks to revoke memory, max revocable size: "
                               << max_revocable_size;
-                SCOPED_ATTACH_TASK(max_revocable_query.get());
+                }
+                SCOPED_ATTACH_TASK(query_ctx.get());
                 // TODO, should spill the task that has max memory, not all
                 for (auto* task : revocable_tasks) {
                     auto st = task->revoke_memory();
@@ -400,8 +401,19 @@ void WorkloadGroupMgr::handle_paused_queries() {
                         break;
                     }
                 }
+                query_it = queries_list.erase(query_it);
+                continue;
+            } else {
+                ++query_it;
             }
-            query_it = queries_list.erase(query_it);
+        }
+
+        // If there are still some queries that not exceed query limit, but exceed wg's memlimit
+        // then, we should notify the workload group that memory is not enough
+        if (!queries_list.empty() && wg->memory_sufficent()) {
+            wg->update_memory_sufficent(false);
+            LOG(INFO) << "wg: " << wg->debug_string()
+                      << " has some query paused, update memory to insufficent";
             continue;
         }