[enhancement](spilldisk)Cancel query fast when reserver memory failed and could not find revocable tasks (#59330)

### What problem does this PR solve?

There  are some problems in the previous code:
1. revocable memory size is ineffective for non-spill operators and
remains 0 in all cases. In this scenario, if we disable spill and enable
reserve, it appears that memlimit will stop working entirely.
2. too complicated in process reserve failed.

In this PR, I simplified the logic:
1. If the query is not enable spill, and if reserve memory failed, just
disable reserve memory and it will enable memory check immediately and
let the query run.
2. If reserve memory failed and not find any revocable tasks, then just
cancel the query.
3. If reserve memory failed, should add to paused query immediately
because it maybe the last block for join build. If not paused, it may
try to allocate a lot of memory.
4. If there are some query in cancel stage, then not check paused query
for a bit while because the cancelling query may release some memory.

Related PR: #xxx

Problem Summary:

### Release note

None

### Check List (For Author)

- Test <!-- At least one of them must be included. -->
    - [ ] Regression test
    - [ ] Unit Test
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason?  -->

- Behavior changed:
    - [ ] No.
    - [ ] Yes. <!-- Explain the behavior change -->

- Does this need documentation?
    - [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->

Author: yiguolei

be/src/pipeline/pipeline_task.cpp

@@ -640,6 +640,14 @@ Status PipelineTask::do_revoke_memory(const std::shared_ptr<SpillContext>& spill
 
 bool PipelineTask::_try_to_reserve_memory(const size_t reserve_size, OperatorBase* op) {
     auto st = thread_context()->thread_mem_tracker_mgr->try_reserve(reserve_size);
+    // If reserve memory failed and the query is not enable spill, just disable reserve memory(this will enable
+    // memory hard limit check, and will cancel the query if allocate memory failed) and let it run.
+    if (!st.ok() && !_state->enable_spill()) {
+        LOG(INFO) << print_id(_query_id) << " reserve memory failed due to " << st
+                  << ", and it is not enable spill, disable reserve memory and let it run";
+        _state->get_query_ctx()->resource_ctx()->task_controller()->disable_reserve_memory();
+        return true;
+    }
     COUNTER_UPDATE(_memory_reserve_times, 1);
     auto sink_revocable_mem_size = _sink->revocable_mem_size(_state);
     if (st.ok() && _state->enable_force_spill() && _sink->is_spillable() &&
@@ -656,13 +664,30 @@ bool PipelineTask::_try_to_reserve_memory(const size_t reserve_size, OperatorBas
                 op->node_id(), _state->task_id(),
                 PrettyPrinter::print_bytes(op->revocable_mem_size(_state)),
                 PrettyPrinter::print_bytes(sink_revocable_mem_size), st.to_string());
-        // PROCESS_MEMORY_EXCEEDED error msg alread contains process_mem_log_str
+        // PROCESS_MEMORY_EXCEEDED error msg already contains process_mem_log_str
         if (!st.is<ErrorCode::PROCESS_MEMORY_EXCEEDED>()) {
             debug_msg +=
                     fmt::format(", debug info: {}", GlobalMemoryArbitrator::process_mem_log_str());
         }
-        LOG_EVERY_N(INFO, 100) << debug_msg;
         // If sink has enough revocable memory, trigger revoke memory
+        LOG(INFO) << fmt::format(
+                "Query: {} sink: {}, node id: {}, task id: "
+                "{}, revocable mem size: {}",
+                print_id(_query_id), _sink->get_name(), _sink->node_id(), _state->task_id(),
+                PrettyPrinter::print_bytes(sink_revocable_mem_size));
+        ExecEnv::GetInstance()->workload_group_mgr()->add_paused_query(
+                _state->get_query_ctx()->resource_ctx()->shared_from_this(), reserve_size, st);
+        _spilling = true;
+        return false;
+        // !!! Attention:
+        // In the past, if reserve failed, not add this query to paused list, because it is very small, will not
+        // consume a lot of memory. But need set low memory mode to indicate that the system should
+        // not use too much memory.
+        // But if we only set _state->get_query_ctx()->set_low_memory_mode() here, and return true, the query will
+        // continue to run and not blocked, and this reserve maybe the last block of join sink opertorator, and it will
+        // build hash table directly and will consume a lot of memory. So that should return false directly.
+        // TODO: we should using a global system buffer management logic to deal with low memory mode.
+        /**
         if (sink_revocable_mem_size >= vectorized::SpillStream::MIN_SPILL_WRITE_BATCH_MEM) {
             LOG(INFO) << fmt::format(
                     "Query: {} sink: {}, node id: {}, task id: "
@@ -674,11 +699,8 @@ bool PipelineTask::_try_to_reserve_memory(const size_t reserve_size, OperatorBas
             _spilling = true;
             return false;
         } else {
-            // If reserve failed, not add this query to paused list, because it is very small, will not
-            // consume a lot of memory. But need set low memory mode to indicate that the system should
-            // not use too much memory.
             _state->get_query_ctx()->set_low_memory_mode();
-        }
+        } */
     }
     return true;
 }

be/src/runtime/memory/mem_tracker_limiter.h

@@ -126,7 +126,6 @@ class MemTrackerLimiter final {
     int64_t group_num() const { return _group_num; }
     int64_t limit() const { return _limit; }
     void set_limit(int64_t new_mem_limit) { _limit = new_mem_limit; }
-    bool enable_check_limit() const { return _enable_check_limit; }
     void set_enable_check_limit(bool enable_check_limit) {
         _enable_check_limit = enable_check_limit;
     }
@@ -298,7 +297,7 @@ inline void MemTrackerLimiter::cache_consume(int64_t bytes) {
 }
 
 inline Status MemTrackerLimiter::check_limit(int64_t bytes) {
-    if (bytes <= 0 || !enable_check_limit() || _limit <= 0) {
+    if (bytes <= 0 || !_enable_check_limit || _limit <= 0) {
         return Status::OK();
     }