[fix](be) Hand off query spill cleanup to background GC

### What problem does this PR solve?

Issue Number: None

Related PR: None

Problem Summary: Query teardown previously deleted the complete spill directory tree synchronously. A large residual spill tree could block a scheduler worker, and a filesystem failure was only logged without retry. In addition, recursive CTE rerunnable parameters retained QueryContext after a successful final close. Empty query directories are now removed with a bounded rmdir; non-empty trees are atomically handed off to the spill GC directory, with failed handoffs retried. Rerunnable parameters are released after FINAL_CLOSE and during query-manager shutdown outside the protecting mutex, allowing QueryContext and its spill directories to be reclaimed.

### Release note

None

### Check List (For Author)

- Test: Unit Test
    - `./run-be-ut.sh --run --filter='SpillFileTest.QueryContextDeletesEmptySpillDirectoryWithoutHandoff:SpillFileTest.QueryContextCleansUpNestedSpillDirectory:SpillFileTest.QueryContextCleansUpResidualSpillDirectory:SpillFileTest.QueryContextContinuesCleanupAfterRootFailure:SpillFileTest.FinalCloseReleasesRerunnableQueryContextAndHandsOffSpillDirectory:FragmentMgrRerunnableParamsTest.StopReleasesLastQueryContextRefOutsideLock'`
- Behavior changed: Yes. Query spill cleanup is handed off to background GC and retried after transient failures; recursive CTE final-close state no longer retains QueryContext.
- Does this need documentation: No

Author: mrhhsg

be/src/exec/spill/spill_file.cpp

@@ -57,8 +57,8 @@ void SpillFile::gc() {
                                                    _spill_dir, status.to_string());
         }
     }
-    // decrease spill data usage anyway, since in ~QueryContext() spill data of the query will be
-    // clean up as a last resort
+    // Decrease spill data usage even if per-file cleanup failed. QueryContext teardown hands the
+    // whole query spill directory to the spill GC path as the final cleanup step.
     _data_dir->update_spill_data_usage(-_total_written_bytes);
     _total_written_bytes = 0;
 }

be/src/exec/spill/spill_file_manager.cpp

@@ -24,27 +24,46 @@
 #include <filesystem>
 #include <memory>
 #include <string>
+#include <utility>
 
 #include "common/logging.h"
 #include "common/metrics/doris_metrics.h"
 #include "exec/spill/spill_file.h"
 #include "io/fs/file_system.h"
 #include "io/fs/local_file_system.h"
 #include "storage/olap_define.h"
+#include "util/debug_points.h"
 #include "util/parse_util.h"
 #include "util/pretty_printer.h"
 #include "util/time.h"
+#include "util/uid_util.h"
 
 namespace doris {
 
 SpillFileManager::~SpillFileManager() {
+    // QueryContext destruction can still hand off directories after stop(), for example while
+    // VDataStreamMgr is being destroyed. Retry the bounded metadata operation once more before
+    // dropping the in-memory retry state. Any source path that still cannot be moved remains under
+    // the active spill root and will be picked up by init() after restart.
+    _retry_pending_query_spill_directories();
     DorisMetrics::instance()->metric_registry()->deregister_entity(_entity);
 }
 
 SpillFileManager::SpillFileManager(
         std::unordered_map<std::string, std::unique_ptr<SpillDataDir>>&& spill_store_map)
         : _spill_store_map(std::move(spill_store_map)), _stop_background_threads_latch(1) {}
 
+void SpillFileManager::stop() {
+    _stop_background_threads_latch.count_down();
+    if (_spill_gc_thread) {
+        _spill_gc_thread->join();
+    }
+    // The GC thread may observe the stop latch before processing a recently queued handoff. Keep
+    // shutdown bounded by retrying only rmdir/rename here; recursive deletion remains asynchronous
+    // or is recovered by init() on the next start.
+    _retry_pending_query_spill_directories();
+}
+
 Status SpillFileManager::init() {
     LOG(INFO) << "init spill stream manager";
     RETURN_IF_ERROR(_init_spill_store_map());
@@ -162,6 +181,71 @@ void SpillFileManager::delete_spill_file(SpillFileSPtr spill_file) {
     spill_file->gc();
 }
 
+void SpillFileManager::handoff_query_spill_directory(const std::string& query_id,
+                                                     SpillDataDir* data_dir) {
+    PendingQuerySpillDirectory pending_directory {
+            .query_dir = data_dir->get_spill_data_path(query_id),
+            .gc_dir = data_dir->get_spill_data_gc_path(
+                    fmt::format("{}-{}", query_id, UniqueId::gen_uid().to_string())),
+    };
+
+    auto status = Status::OK();
+    DBUG_EXECUTE_IF("fault_inject::spill_file_manager::handoff_query_spill_directory", {
+        status = Status::Error<INTERNAL_ERROR>("injected query spill directory handoff failure");
+    });
+    if (status.ok()) {
+        status = _try_handoff_query_spill_directory(pending_directory);
+    }
+    if (!status.ok()) {
+        {
+            std::lock_guard lock(_pending_query_spill_directories_mutex);
+            _pending_query_spill_directories.emplace_back(std::move(pending_directory));
+        }
+        LOG(WARNING) << fmt::format(
+                "failed to hand off spill query directory, dir {}, error: {}; "
+                "the spill GC will retry",
+                data_dir->get_spill_data_path(query_id), status.to_string());
+    }
+}
+
+Status SpillFileManager::_try_handoff_query_spill_directory(
+        const PendingQuerySpillDirectory& pending_directory) {
+    const auto& fs = io::global_local_filesystem();
+    auto status = fs->delete_empty_directory(pending_directory.query_dir);
+    if (status.ok()) {
+        return Status::OK();
+    }
+    if (!status.is<ErrorCode::DIRECTORY_NOT_EMPTY>()) {
+        return status;
+    }
+    return fs->rename(pending_directory.query_dir, pending_directory.gc_dir);
+}
+
+void SpillFileManager::_retry_pending_query_spill_directories() {
+    std::vector<PendingQuerySpillDirectory> pending_directories;
+    {
+        std::lock_guard lock(_pending_query_spill_directories_mutex);
+        pending_directories.swap(_pending_query_spill_directories);
+    }
+
+    std::vector<PendingQuerySpillDirectory> failed_directories;
+    for (auto& pending_directory : pending_directories) {
+        auto status = _try_handoff_query_spill_directory(pending_directory);
+        if (!status.ok()) {
+            LOG_EVERY_T(WARNING, 10) << fmt::format(
+                    "failed to retry spill query directory handoff, dir {}, error: {}",
+                    pending_directory.query_dir, status.to_string());
+            failed_directories.emplace_back(std::move(pending_directory));
+        }
+    }
+    if (!failed_directories.empty()) {
+        std::lock_guard lock(_pending_query_spill_directories_mutex);
+        for (auto& pending_directory : failed_directories) {
+            _pending_query_spill_directories.emplace_back(std::move(pending_directory));
+        }
+    }
+}
+
 void SpillFileManager::gc(int32_t max_work_time_ms) {
     bool exists = true;
     bool has_work = false;
@@ -181,6 +265,7 @@ void SpillFileManager::gc(int32_t max_work_time_ms) {
             LOG(INFO) << msg;
         }
     }};
+    _retry_pending_query_spill_directories();
     for (const auto& [path, store_dir] : _spill_store_map) {
         std::string gc_root_dir = store_dir->get_spill_data_gc_path();
 

be/src/exec/spill/spill_file_manager.h

@@ -20,10 +20,12 @@
 #include <atomic>
 #include <memory>
 #include <mutex>
+#include <string>
 #include <unordered_map>
 #include <vector>
 
 #include "common/metrics/metrics.h"
+#include "common/status.h"
 #include "exec/spill/spill_file.h"
 #include "storage/options.h"
 #include "util/threadpool.h"
@@ -118,12 +120,7 @@ class SpillFileManager {
 
     Status init();
 
-    void stop() {
-        _stop_background_threads_latch.count_down();
-        if (_spill_gc_thread) {
-            _spill_gc_thread->join();
-        }
-    }
+    void stop();
 
     // Create SpillFile and register it
     // @param relative_path  Operator-formatted path under the spill root,
@@ -133,26 +130,41 @@ class SpillFileManager {
     /// Get a unique ID for constructing spill file paths.
     uint64_t next_id() { return id_++; }
 
-    // Mark SpillFile for deletion; asynchronously delete spill files in the GC thread
+    // Delete SpillFile data synchronously.
     void delete_spill_file(SpillFileSPtr spill_file);
 
+    // Remove an empty per-query directory directly. A non-empty tree is atomically handed off to
+    // the spill GC directory so query teardown never recursively walks the tree. Failed handoffs
+    // are retained by the manager and retried by the spill GC thread.
+    void handoff_query_spill_directory(const std::string& query_id, SpillDataDir* data_dir);
+
     void gc(int32_t max_work_time_ms);
 
     void update_spill_write_bytes(int64_t bytes) { _spill_write_bytes_counter->increment(bytes); }
 
     void update_spill_read_bytes(int64_t bytes) { _spill_read_bytes_counter->increment(bytes); }
 
 private:
+    struct PendingQuerySpillDirectory {
+        std::string query_dir;
+        std::string gc_dir;
+    };
+
     void _init_metrics();
     Status _init_spill_store_map();
     void _spill_gc_thread_callback();
+    Status _try_handoff_query_spill_directory(const PendingQuerySpillDirectory& pending_directory);
+    void _retry_pending_query_spill_directories();
     std::vector<SpillDataDir*> _get_stores_for_spill(TStorageMedium::type storage_medium);
 
     std::unordered_map<std::string, std::unique_ptr<SpillDataDir>> _spill_store_map;
 
     CountDownLatch _stop_background_threads_latch;
     std::shared_ptr<Thread> _spill_gc_thread;
 
+    std::mutex _pending_query_spill_directories_mutex;
+    std::vector<PendingQuerySpillDirectory> _pending_query_spill_directories;
+
     std::atomic_uint64_t id_ = 0;
 
     std::shared_ptr<MetricEntity> _entity {nullptr};

be/src/exec/spill/spill_file_writer.cpp

@@ -149,6 +149,9 @@ Status SpillFileWriter::write_block(RuntimeState* state, const Block& block) {
 
     // Lazily open the first part
     if (!_file_writer) {
+        if (_current_part_index == 0) {
+            state->get_query_ctx()->record_spill_data_dir(_data_dir);
+        }
         RETURN_IF_ERROR(_open_next_part());
     }
 

be/src/runtime/fragment_mgr.cpp

@@ -353,10 +353,14 @@ void FragmentMgr::stop() {
     // destructred and remove it from _query_ctx_map_delay_delete which is destructring. it's UB.
     _query_ctx_map_delay_delete.clear();
     _pipeline_map.clear();
+    decltype(_rerunnable_params_map) rerunnable_params_map;
     {
         std::lock_guard<std::mutex> lk(_rerunnable_params_lock);
-        _rerunnable_params_map.clear();
+        rerunnable_params_map.swap(_rerunnable_params_map);
     }
+    // RerunableFragmentInfo holds QueryContext. Destroy it after releasing the lock because
+    // QueryContext::~QueryContext() can re-enter remove_query_context().
+    rerunnable_params_map.clear();
 }
 
 static void empty_function(RuntimeState*, Status*) {}
@@ -431,16 +435,21 @@ void FragmentMgr::remove_pipeline_context(std::pair<TUniqueId, int> key) {
 void FragmentMgr::remove_query_context(const TUniqueId& key) {
     // Clean up any saved rerunnable params for this query to avoid memory leaks.
     // This covers both cancel and normal destruction paths.
+    decltype(_rerunnable_params_map) rerunnable_params_map;
     {
         std::lock_guard<std::mutex> lk(_rerunnable_params_lock);
         for (auto it = _rerunnable_params_map.begin(); it != _rerunnable_params_map.end();) {
             if (it->first.first == key) {
-                it = _rerunnable_params_map.erase(it);
+                auto current = it++;
+                rerunnable_params_map.insert(_rerunnable_params_map.extract(current));
             } else {
                 ++it;
             }
         }
     }
+    // Releasing query_ctx can re-enter this method, so destroy the detached entries outside
+    // _rerunnable_params_lock.
+    rerunnable_params_map.clear();
     _query_ctx_map_delay_delete.erase(key);
 #ifndef BE_TEST
     _query_ctx_map.erase(key);
@@ -1252,7 +1261,7 @@ Status FragmentMgr::transmit_rec_cte_block(
 // wait_for_destroy: collect deregister RF IDs, store brpc closure, trigger old PFC close
 // rebuild: increment stage, deregister old RFs, create+prepare new PFC from saved params
 // submit: submit the new PFC's pipeline tasks for execution
-// final_close: async wait for close, send final report, clean up (last round only)
+// final_close: async wait for close, send final report, clean up saved params (last round only)
 //
 // The brpc ClosureGuard is stored in the PFC so the RPC response is deferred until
 // the PFC is fully destroyed. This gives the caller (RecCTESourceOperatorX) a
@@ -1293,6 +1302,16 @@ Status FragmentMgr::rerun_fragment(const std::shared_ptr<brpc::ClosureGuard>& gu
         SCOPED_ATTACH_TASK(query_ctx);
         RETURN_IF_ERROR(
                 fragment_ctx->listen_wait_close(guard, stage == PRerunFragmentParams::FINAL_CLOSE));
+
+        if (stage == PRerunFragmentParams::FINAL_CLOSE) {
+            // extract() transfers ownership without destroying RerunableFragmentInfo under the
+            // lock. In particular, releasing its QueryContext may re-enter remove_query_context().
+            decltype(_rerunnable_params_map)::node_type final_close_info;
+            {
+                std::lock_guard<std::mutex> lk(_rerunnable_params_lock);
+                final_close_info = _rerunnable_params_map.extract({query_id, fragment_id});
+            }
+        }
         fragment_ctx->notify_close();
         return Status::OK();
     } else if (stage == PRerunFragmentParams::REBUILD) {

be/src/runtime/fragment_mgr.h

@@ -228,8 +228,10 @@ class FragmentMgr : public RestMonitorIface {
 
     // Saved params and callback for rerunnable (recursive CTE) fragments.
     // Only populated when need_notify_close == true during exec_plan_fragment.
-    // Lifecycle: created in exec_plan_fragment(), used in rerun_fragment(rebuild)
-    // to recreate PFC with fresh state, cleaned up in remove_query_context().
+    // Lifecycle: created in exec_plan_fragment(), retained across wait/rebuild/submit rounds,
+    // and removed after a successful final_close. remove_query_context() and stop() provide
+    // fallback cleanup. Entries must be destroyed without holding _rerunnable_params_lock because
+    // releasing query_ctx can re-enter FragmentMgr::remove_query_context().
     struct RerunableFragmentInfo {
         // Runtime filter IDs registered by the old PFC, collected during wait_for_destroy.
         // These are deregistered from the RuntimeFilterMgr before the new PFC is created.

be/src/runtime/query_context.cpp

@@ -219,6 +219,11 @@ void QueryContext::init_query_task_controller() {
 #endif
 }
 
+void QueryContext::record_spill_data_dir(SpillDataDir* data_dir) {
+    std::lock_guard lock(_spill_data_dirs_mutex);
+    _spill_data_dirs.emplace(data_dir);
+}
+
 QueryContext::~QueryContext() {
     SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(query_mem_tracker());
     // query mem tracker consumption is equal to 0, it means that after QueryContext is created,
@@ -247,6 +252,12 @@ QueryContext::~QueryContext() {
     obj_pool.clear();
     _merge_controller_handler.reset();
 
+    if (auto* spill_file_mgr = _exec_env->spill_file_mgr()) {
+        for (auto* data_dir : _spill_data_dirs) {
+            spill_file_mgr->handoff_query_spill_directory(print_id(_query_id), data_dir);
+        }
+    }
+
     DorisMetrics::instance()->query_ctx_cnt->increment(-1);
     // fragment_mgr is nullptr in unittest
     if (ExecEnv::GetInstance()->fragment_mgr()) {

be/src/runtime/query_context.h

@@ -28,6 +28,7 @@
 #include <mutex>
 #include <string>
 #include <unordered_map>
+#include <unordered_set>
 
 #include "common/config.h"
 #include "common/factory_creator.h"
@@ -55,6 +56,7 @@ class PipelineTask;
 class QueryTaskController;
 class Dependency;
 class RecCTEScanLocalState;
+class SpillDataDir;
 
 struct ReportStatusRequest {
     const Status status;
@@ -203,6 +205,10 @@ class QueryContext : public std::enable_shared_from_this<QueryContext> {
 
     TUniqueId query_id() const { return _query_id; }
 
+    // Record a spill data directory before opening the first spill part so teardown only visits
+    // touched roots.
+    void record_spill_data_dir(SpillDataDir* data_dir);
+
     // Expose task-level query progress counters for runtime statistics reporting.
     void add_total_task_num(int delta);
     void inc_finished_task_num();
@@ -329,6 +335,9 @@ class QueryContext : public std::enable_shared_from_this<QueryContext> {
     MonotonicStopWatch _query_watcher;
     bool _is_nereids = false;
 
+    std::mutex _spill_data_dirs_mutex;
+    std::unordered_set<SpillDataDir*> _spill_data_dirs;
+
     std::shared_ptr<ResourceContext> _resource_ctx;
 
     void _init_resource_context();

be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp

@@ -183,4 +183,41 @@ TEST(FragmentMgrDelayDeleteMapTest, ClearShouldNotAbortWhenReleasingLastQueryCon
     exec_env->_fragment_mgr = previous_fragment_mgr;
 }
 
+TEST(FragmentMgrRerunnableParamsTest, StopReleasesLastQueryContextRefOutsideLock) {
+    auto* exec_env = ExecEnv::GetInstance();
+    auto* previous_fragment_mgr = exec_env->_fragment_mgr;
+    auto* fragment_mgr = new FragmentMgr(exec_env);
+    exec_env->_fragment_mgr = fragment_mgr;
+
+    TUniqueId query_id;
+    query_id.__set_hi(303);
+    query_id.__set_lo(404);
+
+    TQueryOptions query_options;
+    query_options.__set_query_type(TQueryType::SELECT);
+    query_options.__set_execution_timeout(60);
+    query_options.__set_mem_limit(64L * 1024 * 1024);
+
+    TNetworkAddress fe_addr;
+    fe_addr.hostname = "127.0.0.1";
+    fe_addr.port = 9030;
+
+    auto query_ctx =
+            QueryContext::create(query_id, exec_env, query_options, fe_addr,
+                                 /*is_nereids*/ true, fe_addr, QuerySource::INTERNAL_FRONTEND);
+    std::weak_ptr<QueryContext> weak_query_ctx = query_ctx;
+    {
+        std::lock_guard<std::mutex> lock(fragment_mgr->_rerunnable_params_lock);
+        fragment_mgr->_rerunnable_params_map[{query_id, 1}].query_ctx = query_ctx;
+    }
+    query_ctx.reset();
+
+    EXPECT_FALSE(weak_query_ctx.expired());
+    fragment_mgr->stop();
+    EXPECT_TRUE(weak_query_ctx.expired());
+
+    exec_env->_fragment_mgr = previous_fragment_mgr;
+    delete fragment_mgr;
+}
+
 } // namespace doris