[fix](be) Refine revocable memory accounting for spill

Issue Number: None

Related PR: None

Problem Summary: Exclude small non-spillable revocable buffers from pipeline task revocable memory accounting and handle queries without revocable tasks when triggering memory revocation.

None

- Test: No need to test (commit existing staged changes only)

- Behavior changed: Yes (revocable memory estimation and empty revocation handling are adjusted)

- Does this need documentation: No

Author: mrhhsg

be/src/exec/pipeline/pipeline_task.cpp

@@ -417,14 +417,7 @@ bool PipelineTask::_should_trigger_revoking(const size_t reserve_size) const {
     }
 
     if (is_high_memory_pressure) {
-        const auto revocable_size = [&]() {
-            size_t total = _sink->revocable_mem_size(_state);
-            for (const auto& op : _operators) {
-                total += op->revocable_mem_size(_state);
-            }
-            return total;
-        }();
-
+        const auto revocable_size = _get_revocable_size();
         const auto total_estimated_revocable = revocable_size * parallelism;
         return total_estimated_revocable >= int64_t(double(query_limit) * 0.2);
     }
@@ -1006,20 +999,31 @@ std::string PipelineTask::debug_string() {
     return fmt::to_string(debug_string_buffer);
 }
 
-size_t PipelineTask::get_revocable_size() const {
-    if (!_opened || is_finalized() || _running || (_eos && !_spilling)) {
-        return 0;
-    }
-
+size_t PipelineTask::_get_revocable_size() const {
     // Sum revocable memory from every operator in the pipeline + the sink.
     // Each operator reports only its own revocable memory (no child recursion).
-    size_t total = _sink->revocable_mem_size(_state);
+    size_t total = 0;
+    size_t sink_revocable_size = _sink->revocable_mem_size(_state);
+    if (sink_revocable_size >= SpillFile::MIN_SPILL_WRITE_BATCH_MEM) {
+        total += sink_revocable_size;
+    }
     for (const auto& op : _operators) {
-        total += op->revocable_mem_size(_state);
+        size_t ops_revocable_size = op->revocable_mem_size(_state);
+        if (ops_revocable_size >= SpillFile::MIN_SPILL_WRITE_BATCH_MEM) {
+            total += ops_revocable_size;
+        }
     }
     return total;
 }
 
+size_t PipelineTask::get_revocable_size() const {
+    if (!_opened || is_finalized() || _running || (_eos && !_spilling)) {
+        return 0;
+    }
+
+    return _get_revocable_size();
+}
+
 Status PipelineTask::revoke_memory(const std::shared_ptr<SpillContext>& spill_context) {
     DCHECK(spill_context);
     if (is_finalized()) {

be/src/exec/pipeline/pipeline_task.h

@@ -204,6 +204,7 @@ class PipelineTask : public std::enable_shared_from_this<PipelineTask> {
     // otherwise return true.
     bool _try_to_reserve_memory(const size_t reserve_size, OperatorBase* op);
     bool _should_trigger_revoking(const size_t reserve_size) const;
+    size_t _get_revocable_size() const;
 
     const TUniqueId _query_id;
     const uint32_t _index;

be/src/runtime/workload_management/query_task_controller.cpp

@@ -17,6 +17,8 @@
 
 #include "runtime/workload_management/query_task_controller.h"
 
+#include <algorithm>
+
 #include "exec/pipeline/pipeline_fragment_context.h"
 #include "runtime/query_context.h"
 #include "runtime/workload_management/task_controller.h"
@@ -149,7 +151,19 @@ Status QueryTaskController::revoke_memory() {
         fragments.emplace_back(std::move(fragment_ctx));
     }
 
-    std::sort(tasks.begin(), tasks.end(), [](auto&& l, auto&& r) { return l.first > r.first; });
+    if (tasks.empty()) {
+        LOG(INFO) << fmt::format(
+                "Query {} try to revoke memory, but there is no revocable task, maybe because the "
+                "query was spilled already. Query memory usage: {}, wg info: {}. {}",
+                print_id(query_ctx->query_id()),
+                PrettyPrinter::print_bytes(query_ctx->query_mem_tracker()->consumption()),
+                query_ctx->workload_group()->memory_debug_string(),
+                doris::ProcessProfile::instance()->memory_profile()->process_memory_detail_str());
+        query_ctx->set_memory_sufficient(true);
+        return Status::OK();
+    }
+
+    std::ranges::sort(tasks, [](auto&& l, auto&& r) { return l.first > r.first; });
 
     // Do not use memlimit, use current memory usage.
     // For example, if current limit is 1.6G, but current used is 1G, if reserve failed