branch-4.1: [fix](be) Clean up spill directories during query teardown

be/src/exec/spill/spill_file.cpp

@@ -58,8 +58,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 deletes the
+    // whole query spill directory and retains failures for later retries.
     _data_dir->update_spill_data_usage(-_total_written_bytes);
     _total_written_bytes = 0;
 }

be/src/exec/spill/spill_file_manager.cpp

@@ -24,13 +24,15 @@
 #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"
@@ -39,13 +41,29 @@ namespace doris {
 #include "common/compile_check_begin.h"
 
 SpillFileManager::~SpillFileManager() {
+    // QueryContext destruction can still queue failed deletions after stop(), for example while
+    // VDataStreamMgr is being destroyed. Retry them once more before dropping the in-memory state.
+    // Any directory that still cannot be deleted remains under the active spill root and will be
+    // moved to the GC root 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 failed deletion.
+    // Retry the pending directories after the thread exits; later failures get one final retry in
+    // the destructor.
+    _retry_pending_query_spill_directories();
+}
+
 Status SpillFileManager::init() {
     LOG(INFO) << "init spill stream manager";
     RETURN_IF_ERROR(_init_spill_store_map());
@@ -98,7 +116,7 @@ void SpillFileManager::_init_metrics() {
             _spill_read_bytes_metric.get()));
 }
 
-// clean up stale spilled files
+// Retry failed query-directory deletions and clean up stale spill files.
 void SpillFileManager::_spill_gc_thread_callback() {
     while (!_stop_background_threads_latch.wait_for(
             std::chrono::milliseconds(config::spill_gc_interval_ms))) {
@@ -163,6 +181,66 @@ void SpillFileManager::delete_spill_file(SpillFileSPtr spill_file) {
     spill_file->gc();
 }
 
+void SpillFileManager::delete_query_spill_directory(const std::string& query_id,
+                                                    SpillDataDir* data_dir) {
+    PendingQuerySpillDirectory pending_directory {
+            .query_dir = data_dir->get_spill_data_path(query_id),
+    };
+
+    auto status = _try_delete_query_spill_directory(pending_directory);
+    if (!status.ok()) {
+        std::lock_guard lock(_pending_query_spill_directories_mutex);
+        ++pending_directory.failed_count;
+        _pending_query_spill_directories.emplace_back(std::move(pending_directory));
+    }
+}
+
+Status SpillFileManager::_try_delete_query_spill_directory(
+        const PendingQuerySpillDirectory& pending_directory) {
+    DBUG_EXECUTE_IF("fault_inject::spill_file_manager::delete_query_spill_directory", {
+        return Status::Error<INTERNAL_ERROR>("injected query spill directory deletion failure");
+    });
+    const auto& fs = io::global_local_filesystem();
+    return fs->delete_directory(pending_directory.query_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);
+    }
+    DBUG_EXECUTE_IF(
+            "fault_inject::spill_file_manager::retry_pending_query_spill_directories_after_drain",
+            { DBUG_RUN_CALLBACK(); });
+
+    // Limit repeated warnings for a persistently unavailable directory while retaining it for
+    // every subsequent retry.
+    constexpr int log_interval = 5;
+    std::vector<PendingQuerySpillDirectory> failed_directories;
+    for (auto& pending_directory : pending_directories) {
+        auto status = _try_delete_query_spill_directory(pending_directory);
+        if (status.ok()) {
+            continue;
+        }
+
+        ++pending_directory.failed_count;
+        if (pending_directory.failed_count % log_interval == 0) {
+            LOG(WARNING) << fmt::format(
+                    "failed to retry deleting spill query directory, 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;
@@ -182,6 +260,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();
 
@@ -253,12 +332,12 @@ SpillDataDir::SpillDataDir(std::string path, int64_t capacity_bytes,
 }
 
 bool is_directory_empty(const std::filesystem::path& dir) {
+    // Spill cleanup may delete the directory while the iterator is constructed or advanced. Treat
+    // that race as empty for these presence metrics.
     try {
         return std::filesystem::is_directory(dir) &&
                std::filesystem::directory_iterator(dir) ==
                        std::filesystem::end(std::filesystem::directory_iterator {});
-        // this method is not thread safe, the file referenced by directory_iterator
-        // maybe moved to spill_gc dir during this function call, so need to catch expection
     } catch (const std::filesystem::filesystem_error&) {
         return true;
     }

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"
@@ -119,12 +121,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,
@@ -134,26 +131,40 @@ 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);
 
+    // Recursively delete a per-query spill directory during query teardown. Failed deletions are
+    // retained by the manager and retried by its GC and shutdown paths.
+    void delete_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 {
+        int failed_count {0};
+        std::string query_dir;
+    };
+
     void _init_metrics();
     Status _init_spill_store_map();
     void _spill_gc_thread_callback();
+    Status _try_delete_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

@@ -152,6 +152,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/query_context.cpp

@@ -205,6 +205,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->delete_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"
@@ -50,6 +51,7 @@ class PipelineTask;
 class QueryTaskController;
 class Dependency;
 class RecCTEScanLocalState;
+class SpillDataDir;
 
 struct ReportStatusRequest {
     const Status status;
@@ -198,6 +200,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();
@@ -319,6 +325,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();