diff --git a/be/src/exec/spill/spill_file.cpp b/be/src/exec/spill/spill_file.cpp index 78347ea78ebf9e..7f403de57fcb22 100644 --- a/be/src/exec/spill/spill_file.cpp +++ b/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; } diff --git a/be/src/exec/spill/spill_file_manager.cpp b/be/src/exec/spill/spill_file_manager.cpp index 0946e3209b7bec..048e1d2e5513fe 100644 --- a/be/src/exec/spill/spill_file_manager.cpp +++ b/be/src/exec/spill/spill_file_manager.cpp @@ -24,6 +24,7 @@ #include #include #include +#include #include "common/logging.h" #include "common/metrics/doris_metrics.h" @@ -31,6 +32,7 @@ #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,6 +41,11 @@ 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); } @@ -46,6 +53,17 @@ SpillFileManager::SpillFileManager( std::unordered_map>&& 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("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 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 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; } diff --git a/be/src/exec/spill/spill_file_manager.h b/be/src/exec/spill/spill_file_manager.h index 582df3f1f63c9b..7455789c2162d7 100644 --- a/be/src/exec/spill/spill_file_manager.h +++ b/be/src/exec/spill/spill_file_manager.h @@ -20,10 +20,12 @@ #include #include #include +#include #include #include #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,9 +131,13 @@ 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); } @@ -144,9 +145,16 @@ class SpillFileManager { 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 _get_stores_for_spill(TStorageMedium::type storage_medium); std::unordered_map> _spill_store_map; @@ -154,6 +162,9 @@ class SpillFileManager { CountDownLatch _stop_background_threads_latch; std::shared_ptr _spill_gc_thread; + std::mutex _pending_query_spill_directories_mutex; + std::vector _pending_query_spill_directories; + std::atomic_uint64_t id_ = 0; std::shared_ptr _entity {nullptr}; diff --git a/be/src/exec/spill/spill_file_writer.cpp b/be/src/exec/spill/spill_file_writer.cpp index ced813e51e3080..62320ef9c05641 100644 --- a/be/src/exec/spill/spill_file_writer.cpp +++ b/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()); } diff --git a/be/src/runtime/query_context.cpp b/be/src/runtime/query_context.cpp index 29748d7639d06a..aeaf2eccf707ed 100644 --- a/be/src/runtime/query_context.cpp +++ b/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()) { diff --git a/be/src/runtime/query_context.h b/be/src/runtime/query_context.h index 3102cc8529613a..a4b55d001a5f1b 100644 --- a/be/src/runtime/query_context.h +++ b/be/src/runtime/query_context.h @@ -28,6 +28,7 @@ #include #include #include +#include #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 { 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 { MonotonicStopWatch _query_watcher; bool _is_nereids = false; + std::mutex _spill_data_dirs_mutex; + std::unordered_set _spill_data_dirs; + std::shared_ptr _resource_ctx; void _init_resource_context(); diff --git a/be/test/vec/spill/spill_file_test.cpp b/be/test/vec/spill/spill_file_test.cpp index 67740acdcd2e97..11b16df17128a1 100644 --- a/be/test/vec/spill/spill_file_test.cpp +++ b/be/test/vec/spill/spill_file_test.cpp @@ -21,8 +21,11 @@ #include #include +#include #include #include +#include +#include #include #include "common/config.h" @@ -32,11 +35,16 @@ #include "exec/spill/spill_file_manager.h" #include "exec/spill/spill_file_reader.h" #include "exec/spill/spill_file_writer.h" +#include "io/fs/file_writer.h" #include "io/fs/local_file_system.h" #include "runtime/exec_env.h" #include "runtime/runtime_profile.h" #include "testutil/column_helper.h" +#include "testutil/mock/mock_query_context.h" #include "testutil/mock/mock_runtime_state.h" +#include "util/debug_points.h" +#include "util/defer_op.h" +#include "util/uid_util.h" namespace doris::vectorized { @@ -79,13 +87,21 @@ class SpillFileTest : public testing::Test { _profile->add_child(_common_profile.get(), true); _spill_dir = "./ut_dir/spill_file_test"; - auto spill_data_dir = std::make_unique(_spill_dir, 1024L * 1024 * 128); + _second_spill_dir = "./ut_dir/spill_file_test_second"; + auto spill_data_dir = + std::make_unique(_spill_dir, 1024L * 1024 * 128, TStorageMedium::SSD); auto st = io::global_local_filesystem()->create_directory(spill_data_dir->path(), false); ASSERT_TRUE(st.ok()) << "create directory failed: " << st.to_string(); + auto second_spill_data_dir = std::make_unique( + _second_spill_dir, 1024L * 1024 * 128, TStorageMedium::HDD); + st = io::global_local_filesystem()->create_directory(second_spill_data_dir->path(), false); + ASSERT_TRUE(st.ok()) << "create directory failed: " << st.to_string(); std::unordered_map> data_map; _data_dir_ptr = spill_data_dir.get(); + _second_data_dir_ptr = second_spill_data_dir.get(); data_map.emplace("test", std::move(spill_data_dir)); + data_map.emplace("test_second", std::move(second_spill_data_dir)); auto* spill_file_manager = new SpillFileManager(std::move(data_map)); ExecEnv::GetInstance()->_spill_file_mgr = spill_file_manager; st = spill_file_manager->init(); @@ -94,11 +110,13 @@ class SpillFileTest : public testing::Test { void TearDown() override { ExecEnv::GetInstance()->spill_file_mgr()->stop(); + _runtime_state.reset(); SAFE_DELETE(ExecEnv::GetInstance()->_spill_file_mgr); // Clean up test directory auto st = io::global_local_filesystem()->delete_directory(_spill_dir); (void)st; - _runtime_state.reset(); + st = io::global_local_filesystem()->delete_directory(_second_spill_dir); + (void)st; } Block _create_int_block(const std::vector& data) { @@ -112,12 +130,58 @@ class SpillFileTest : public testing::Test { return block; } + void _write_and_release_spill_file(const TUniqueId& query_id, QueryContext* query_ctx, + SpillDataDir* data_dir, const std::string& relative_path) { + TQueryGlobals query_globals; + auto runtime_state = std::make_unique( + query_id, 0, query_ctx->query_options(), query_globals, ExecEnv::GetInstance(), + query_ctx); + + auto spill_file = std::make_shared( + data_dir, fmt::format("{}/{}", print_id(query_id), relative_path)); + + SpillFileWriterSPtr writer; + auto st = spill_file->create_writer(runtime_state.get(), _profile.get(), writer); + ASSERT_TRUE(st.ok()); + auto block = _create_int_block({1, 2, 3}); + st = writer->write_block(runtime_state.get(), block); + ASSERT_TRUE(st.ok()); + st = writer->close(); + ASSERT_TRUE(st.ok()); + writer.reset(); + spill_file.reset(); + } + + void _create_residual_file(const std::string& file_path) { + auto st = io::global_local_filesystem()->create_directory( + std::filesystem::path(file_path).parent_path(), false); + ASSERT_TRUE(st.ok()) << st.to_string(); + io::FileWriterPtr writer; + st = io::global_local_filesystem()->create_file(file_path, &writer); + ASSERT_TRUE(st.ok()) << st.to_string(); + st = writer->close(); + ASSERT_TRUE(st.ok()) << st.to_string(); + } + + std::set _gc_subdirectories(SpillDataDir* data_dir) { + std::set subdirectories; + for (const auto& entry : + std::filesystem::directory_iterator(data_dir->get_spill_data_gc_path())) { + if (entry.is_directory()) { + subdirectories.emplace(entry.path().filename().string()); + } + } + return subdirectories; + } + std::unique_ptr _runtime_state; std::unique_ptr _profile; std::unique_ptr _custom_profile; std::unique_ptr _common_profile; std::string _spill_dir; + std::string _second_spill_dir; SpillDataDir* _data_dir_ptr = nullptr; + SpillDataDir* _second_data_dir_ptr = nullptr; }; // ═══════════════════════════════════════════════════════════════════════ @@ -890,7 +954,331 @@ TEST_F(SpillFileTest, GCCleansUpFiles) { ASSERT_FALSE(exists); } -TEST_F(SpillFileTest, DeleteSpillFileThroughManager) { +TEST_F(SpillFileTest, QueryContextDeletesEmptySpillDirectory) { + ExecEnv::GetInstance()->spill_file_mgr()->stop(); + + TUniqueId query_id; + query_id.hi = 1; + query_id.lo = 2; + auto query_id_str = print_id(query_id); + auto query_ctx = MockQueryContext::create(query_id); + + auto query_dir = _data_dir_ptr->get_spill_data_path(query_id_str); + const auto gc_subdirectories_before = _gc_subdirectories(_data_dir_ptr); + _write_and_release_spill_file(query_id, query_ctx.get(), _data_dir_ptr, "query_context_gc"); + + bool exists = false; + auto st = io::global_local_filesystem()->exists(query_dir, &exists); + ASSERT_TRUE(st.ok()); + ASSERT_TRUE(exists); + + query_ctx.reset(); + + st = io::global_local_filesystem()->exists(query_dir, &exists); + ASSERT_TRUE(st.ok()); + ASSERT_FALSE(exists); + st = io::global_local_filesystem()->exists(_data_dir_ptr->get_spill_data_path(), &exists); + ASSERT_TRUE(st.ok()); + ASSERT_TRUE(exists); + EXPECT_EQ(_gc_subdirectories(_data_dir_ptr), gc_subdirectories_before); +} + +TEST_F(SpillFileTest, QueryContextCleansUpNestedSpillDirectory) { + TUniqueId query_id; + query_id.hi = 3; + query_id.lo = 4; + auto query_id_str = print_id(query_id); + auto query_ctx = MockQueryContext::create(query_id); + + auto query_dir = _data_dir_ptr->get_spill_data_path(query_id_str); + auto nested_dir = query_dir + "/nested"; + _write_and_release_spill_file(query_id, query_ctx.get(), _data_dir_ptr, + "nested/query_context_gc"); + + bool exists = false; + auto st = io::global_local_filesystem()->exists(query_dir, &exists); + ASSERT_TRUE(st.ok()); + ASSERT_TRUE(exists); + st = io::global_local_filesystem()->exists(nested_dir, &exists); + ASSERT_TRUE(st.ok()); + ASSERT_TRUE(exists); + + query_ctx.reset(); + + st = io::global_local_filesystem()->exists(query_dir, &exists); + ASSERT_TRUE(st.ok()); + ASSERT_FALSE(exists); +} + +TEST_F(SpillFileTest, QueryContextDeletesResidualSpillDirectory) { + ExecEnv::GetInstance()->spill_file_mgr()->stop(); + + TUniqueId query_id; + query_id.hi = 5; + query_id.lo = 6; + auto query_id_str = print_id(query_id); + auto query_ctx = MockQueryContext::create(query_id); + + auto query_dir = _data_dir_ptr->get_spill_data_path(query_id_str); + const auto gc_subdirectories_before = _gc_subdirectories(_data_dir_ptr); + _write_and_release_spill_file(query_id, query_ctx.get(), _data_dir_ptr, "query_context_gc"); + + auto residual_file = query_dir + "/residual/temporary-data"; + _create_residual_file(residual_file); + + bool exists = false; + auto st = io::global_local_filesystem()->exists(residual_file, &exists); + ASSERT_TRUE(st.ok()); + ASSERT_TRUE(exists); + + query_ctx.reset(); + + st = io::global_local_filesystem()->exists(query_dir, &exists); + ASSERT_TRUE(st.ok()); + ASSERT_FALSE(exists); + + EXPECT_EQ(_gc_subdirectories(_data_dir_ptr), gc_subdirectories_before); +} + +TEST_F(SpillFileTest, QueryContextCleansUpAllTouchedSpillDirectories) { + TUniqueId query_id; + query_id.hi = 9; + query_id.lo = 10; + auto query_id_str = print_id(query_id); + auto query_ctx = MockQueryContext::create(query_id); + + auto first_query_dir = _data_dir_ptr->get_spill_data_path(query_id_str); + auto second_query_dir = _second_data_dir_ptr->get_spill_data_path(query_id_str); + const auto first_gc_subdirectories_before = _gc_subdirectories(_data_dir_ptr); + const auto second_gc_subdirectories_before = _gc_subdirectories(_second_data_dir_ptr); + _write_and_release_spill_file(query_id, query_ctx.get(), _data_dir_ptr, "first"); + _write_and_release_spill_file(query_id, query_ctx.get(), _second_data_dir_ptr, "second"); + + bool first_exists = false; + bool second_exists = false; + auto st = io::global_local_filesystem()->exists(first_query_dir, &first_exists); + ASSERT_TRUE(st.ok()); + st = io::global_local_filesystem()->exists(second_query_dir, &second_exists); + ASSERT_TRUE(st.ok()); + ASSERT_TRUE(first_exists); + ASSERT_TRUE(second_exists); + + query_ctx.reset(); + + st = io::global_local_filesystem()->exists(first_query_dir, &first_exists); + ASSERT_TRUE(st.ok()); + st = io::global_local_filesystem()->exists(second_query_dir, &second_exists); + ASSERT_TRUE(st.ok()); + ASSERT_FALSE(first_exists); + ASSERT_FALSE(second_exists); + EXPECT_EQ(_gc_subdirectories(_data_dir_ptr), first_gc_subdirectories_before); + EXPECT_EQ(_gc_subdirectories(_second_data_dir_ptr), second_gc_subdirectories_before); +} + +TEST_F(SpillFileTest, QueryContextContinuesCleanupAfterRootFailure) { + ExecEnv::GetInstance()->spill_file_mgr()->stop(); + + TUniqueId query_id; + query_id.hi = 11; + query_id.lo = 12; + auto query_id_str = print_id(query_id); + auto query_ctx = MockQueryContext::create(query_id); + + auto first_query_dir = _data_dir_ptr->get_spill_data_path(query_id_str); + auto second_query_dir = _second_data_dir_ptr->get_spill_data_path(query_id_str); + const auto first_gc_subdirectories_before = _gc_subdirectories(_data_dir_ptr); + const auto second_gc_subdirectories_before = _gc_subdirectories(_second_data_dir_ptr); + _write_and_release_spill_file(query_id, query_ctx.get(), _data_dir_ptr, "first"); + _write_and_release_spill_file(query_id, query_ctx.get(), _second_data_dir_ptr, "second"); + _create_residual_file(first_query_dir + "/residual/temporary-data"); + _create_residual_file(second_query_dir + "/residual/temporary-data"); + + const auto live_query_dir = _data_dir_ptr->get_spill_data_path("live-query"); + const auto live_query_file = live_query_dir + "/sentinel"; + _create_residual_file(live_query_file); + + bool first_exists = false; + bool second_exists = false; + auto st = io::global_local_filesystem()->exists(first_query_dir, &first_exists); + ASSERT_TRUE(st.ok()); + st = io::global_local_filesystem()->exists(second_query_dir, &second_exists); + ASSERT_TRUE(st.ok()); + ASSERT_TRUE(first_exists); + ASSERT_TRUE(second_exists); + + const bool previous_enable_debug_points = config::enable_debug_points; + constexpr auto debug_point_name = + "fault_inject::spill_file_manager::delete_query_spill_directory"; + Defer restore_debug_point([&] { + DebugPoints::instance()->remove(debug_point_name); + config::enable_debug_points = previous_enable_debug_points; + }); + auto debug_point = std::make_shared(); + debug_point->execute_limit = 1; + config::enable_debug_points = true; + DebugPoints::instance()->add(debug_point_name, debug_point); + + query_ctx.reset(); + + st = io::global_local_filesystem()->exists(first_query_dir, &first_exists); + ASSERT_TRUE(st.ok()); + st = io::global_local_filesystem()->exists(second_query_dir, &second_exists); + ASSERT_TRUE(st.ok()); + ASSERT_NE(first_exists, second_exists); + ASSERT_EQ(debug_point->execute_num.load(), 2); + + ExecEnv::GetInstance()->spill_file_mgr()->gc(10000); + + st = io::global_local_filesystem()->exists(first_query_dir, &first_exists); + ASSERT_TRUE(st.ok()); + st = io::global_local_filesystem()->exists(second_query_dir, &second_exists); + ASSERT_TRUE(st.ok()); + EXPECT_FALSE(first_exists); + EXPECT_FALSE(second_exists); + + bool live_query_exists = false; + st = io::global_local_filesystem()->exists(live_query_file, &live_query_exists); + ASSERT_TRUE(st.ok()); + EXPECT_TRUE(live_query_exists); + + EXPECT_EQ(_gc_subdirectories(_data_dir_ptr), first_gc_subdirectories_before); + EXPECT_EQ(_gc_subdirectories(_second_data_dir_ptr), second_gc_subdirectories_before); +} + +TEST_F(SpillFileTest, QueryContextRetriesSpillDirectoryDeletionUntilSuccess) { + ExecEnv::GetInstance()->spill_file_mgr()->stop(); + + TUniqueId query_id; + query_id.hi = 15; + query_id.lo = 16; + auto query_id_str = print_id(query_id); + auto query_ctx = MockQueryContext::create(query_id); + + auto query_dir = _data_dir_ptr->get_spill_data_path(query_id_str); + const auto gc_subdirectories_before = _gc_subdirectories(_data_dir_ptr); + _write_and_release_spill_file(query_id, query_ctx.get(), _data_dir_ptr, "retry_cleanup"); + _create_residual_file(query_dir + "/residual/temporary-data"); + + const bool previous_enable_debug_points = config::enable_debug_points; + constexpr auto debug_point_name = + "fault_inject::spill_file_manager::delete_query_spill_directory"; + Defer restore_debug_point([&] { + DebugPoints::instance()->remove(debug_point_name); + config::enable_debug_points = previous_enable_debug_points; + }); + auto debug_point = std::make_shared(); + debug_point->execute_limit = 5; + config::enable_debug_points = true; + DebugPoints::instance()->add(debug_point_name, debug_point); + + query_ctx.reset(); + + bool exists = false; + auto st = io::global_local_filesystem()->exists(query_dir, &exists); + ASSERT_TRUE(st.ok()); + ASSERT_TRUE(exists); + + for (int i = 0; i < 4; ++i) { + ExecEnv::GetInstance()->spill_file_mgr()->gc(10000); + st = io::global_local_filesystem()->exists(query_dir, &exists); + ASSERT_TRUE(st.ok()); + ASSERT_TRUE(exists); + } + + ExecEnv::GetInstance()->spill_file_mgr()->gc(10000); + st = io::global_local_filesystem()->exists(query_dir, &exists); + ASSERT_TRUE(st.ok()); + EXPECT_FALSE(exists); + EXPECT_EQ(debug_point->execute_num.load(), 6); + EXPECT_EQ(_gc_subdirectories(_data_dir_ptr), gc_subdirectories_before); +} + +TEST_F(SpillFileTest, RetryPreservesDirectoryQueuedAfterPendingDrain) { + ExecEnv::GetInstance()->spill_file_mgr()->stop(); + + TUniqueId first_query_id; + first_query_id.hi = 17; + first_query_id.lo = 18; + auto first_query_ctx = MockQueryContext::create(first_query_id); + auto first_query_dir = _data_dir_ptr->get_spill_data_path(print_id(first_query_id)); + _write_and_release_spill_file(first_query_id, first_query_ctx.get(), _data_dir_ptr, + "first_retry_cleanup"); + _create_residual_file(first_query_dir + "/residual/temporary-data"); + + TUniqueId second_query_id; + second_query_id.hi = 19; + second_query_id.lo = 20; + auto second_query_ctx = MockQueryContext::create(second_query_id); + auto second_query_dir = _data_dir_ptr->get_spill_data_path(print_id(second_query_id)); + _write_and_release_spill_file(second_query_id, second_query_ctx.get(), _data_dir_ptr, + "second_retry_cleanup"); + _create_residual_file(second_query_dir + "/residual/temporary-data"); + + const bool previous_enable_debug_points = config::enable_debug_points; + constexpr auto delete_debug_point_name = + "fault_inject::spill_file_manager::delete_query_spill_directory"; + constexpr auto after_drain_debug_point_name = + "fault_inject::spill_file_manager::retry_pending_query_spill_directories_after_drain"; + Defer restore_debug_points([&] { + DebugPoints::instance()->remove(after_drain_debug_point_name); + DebugPoints::instance()->remove(delete_debug_point_name); + config::enable_debug_points = previous_enable_debug_points; + }); + config::enable_debug_points = true; + DebugPoints::instance()->add(delete_debug_point_name); + + first_query_ctx.reset(); + + auto after_drain_debug_point = std::make_shared(); + after_drain_debug_point->execute_limit = 1; + after_drain_debug_point->callback = std::function([&]() { second_query_ctx.reset(); }); + DebugPoints::instance()->add(after_drain_debug_point_name, after_drain_debug_point); + + ExecEnv::GetInstance()->spill_file_mgr()->gc(10000); + + bool first_exists = false; + auto st = io::global_local_filesystem()->exists(first_query_dir, &first_exists); + ASSERT_TRUE(st.ok()); + bool second_exists = false; + st = io::global_local_filesystem()->exists(second_query_dir, &second_exists); + ASSERT_TRUE(st.ok()); + ASSERT_TRUE(first_exists); + ASSERT_TRUE(second_exists); + ASSERT_EQ(after_drain_debug_point->execute_num.load(), 1); + + DebugPoints::instance()->remove(after_drain_debug_point_name); + DebugPoints::instance()->remove(delete_debug_point_name); + ExecEnv::GetInstance()->spill_file_mgr()->gc(10000); + + st = io::global_local_filesystem()->exists(first_query_dir, &first_exists); + ASSERT_TRUE(st.ok()); + st = io::global_local_filesystem()->exists(second_query_dir, &second_exists); + ASSERT_TRUE(st.ok()); + EXPECT_FALSE(first_exists); + EXPECT_FALSE(second_exists); +} + +TEST_F(SpillFileTest, QueryContextSkipsCleanupWithoutSpill) { + TUniqueId query_id; + query_id.hi = 7; + query_id.lo = 8; + auto query_id_str = print_id(query_id); + auto query_ctx = MockQueryContext::create(query_id); + auto query_dir = _data_dir_ptr->get_spill_data_path(query_id_str); + + // No spill root was recorded for this query, so teardown must leave this untracked directory. + auto st = io::global_local_filesystem()->create_directory(query_dir, false); + ASSERT_TRUE(st.ok()); + + query_ctx.reset(); + + bool exists = false; + st = io::global_local_filesystem()->exists(query_dir, &exists); + ASSERT_TRUE(st.ok()); + ASSERT_TRUE(exists); +} + +TEST_F(SpillFileTest, DeleteSpillFileThroughManagerSynchronously) { SpillFileSPtr spill_file; auto st = ExecEnv::GetInstance()->spill_file_mgr()->create_spill_file("test_query/mgr_delete", spill_file); @@ -907,11 +1295,17 @@ TEST_F(SpillFileTest, DeleteSpillFileThroughManager) { st = writer->close(); ASSERT_TRUE(st.ok()); - // Delete through manager (async GC) + auto spill_file_dir = _data_dir_ptr->get_spill_data_path("test_query/mgr_delete"); + bool exists = false; + st = io::global_local_filesystem()->exists(spill_file_dir, &exists); + ASSERT_TRUE(st.ok()); + ASSERT_TRUE(exists); + ExecEnv::GetInstance()->spill_file_mgr()->delete_spill_file(spill_file); - // Run GC to process the deletion - ExecEnv::GetInstance()->spill_file_mgr()->gc(1000); + st = io::global_local_filesystem()->exists(spill_file_dir, &exists); + ASSERT_TRUE(st.ok()); + ASSERT_FALSE(exists); } // ═══════════════════════════════════════════════════════════════════════ diff --git a/regression-test/suites/spill_p0/test_spill_directory_cleanup.groovy b/regression-test/suites/spill_p0/test_spill_directory_cleanup.groovy new file mode 100644 index 00000000000000..ba9b99774fb60f --- /dev/null +++ b/regression-test/suites/spill_p0/test_spill_directory_cleanup.groovy @@ -0,0 +1,113 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import org.apache.doris.regression.util.Http + +suite("test_spill_directory_cleanup", "nonConcurrent") { + def backends = sql_return_maparray("SHOW BACKENDS").findAll { + it.Alive.toString().equalsIgnoreCase("true") + } + assertTrue(!backends.isEmpty(), "No alive backend found") + + def readSpillMetric = { String metricName -> + def valuesByBackend = [:] + backends.each { backend -> + def endpoint = "http://${backend.Host}:${backend.HttpPort}/metrics" + def metrics = Http.GET(endpoint, false, false).toString() + def metricPattern = '(?m)^' + java.util.regex.Pattern.quote(metricName) + + '(?:\\{[^}]*\\})?\\s+(\\d+)$' + def matcher = java.util.regex.Pattern.compile(metricPattern).matcher(metrics) + long value = 0 + int matchedMetrics = 0 + while (matcher.find()) { + value += matcher.group(1).toLong() + matchedMetrics++ + } + assertTrue(matchedMetrics > 0, "Metric ${metricName} not found on ${endpoint}") + valuesByBackend[backend.BackendId.toString()] = value + } + return valuesByBackend + } + + def waitForSpillDirectoryState = { boolean expectedPresent, String stage -> + def lastPresence = [:] + Throwable lastError = null + long deadline = System.currentTimeMillis() + 60_000 + while (System.currentTimeMillis() < deadline) { + try { + lastPresence = readSpillMetric("doris_be_spill_disk_has_spill_data") + boolean anyPresent = lastPresence.values().any { it > 0 } + if (anyPresent == expectedPresent) { + logger.info("Spill directory state at ${stage}: ${lastPresence}") + return lastPresence + } + lastError = null + } catch (Throwable t) { + lastError = t + } + sleep(500) + } + def errorDetail = lastError == null ? "" : ", last error: ${lastError.message}" + assertTrue(false, + "Timed out waiting for spill directory state at ${stage}; " + + "expectedPresent=${expectedPresent}, lastPresence=${lastPresence}" + + errorDetail) + } + + GetDebugPoint().clearDebugPointsForAllBEs() + waitForSpillDirectoryState(false, "before query") + def spillWriteBytesBefore = + readSpillMetric("doris_be_spill_write_bytes").values().sum(0L) + + def deleteFailureDebugPoint = + "fault_inject::spill_file_manager::delete_query_spill_directory" + try { + // Keep the otherwise short-lived empty query directory visible until the metric refreshes. + // Disabling this point lets the spill GC retry the same deletion. + GetDebugPoint().enableDebugPointForAllBEs(deleteFailureDebugPoint, [timeout: "120"]) + + sql "SET enable_spill = true" + sql "SET enable_force_spill = true" + sql "SET spill_min_revocable_mem = 1048576" + sql "SET parallel_pipeline_task_num = 1" + sql "SET batch_size = 1024" + sql "SET enable_reserve_memory = true" + + def result = sql """ + SELECT COUNT(*) + FROM ( + SELECT number + FROM numbers("number" = "200000") + GROUP BY number + HAVING SUM(number) >= 0 + ) t + """ + assertEquals("200000", result[0][0].toString()) + + def spillWriteBytesAfter = + readSpillMetric("doris_be_spill_write_bytes").values().sum(0L) + assertTrue(spillWriteBytesAfter > spillWriteBytesBefore, + "The query did not write spill data: before=${spillWriteBytesBefore}, " + + "after=${spillWriteBytesAfter}") + waitForSpillDirectoryState(true, "after forced spill") + } finally { + GetDebugPoint().disableDebugPointForAllBEs(deleteFailureDebugPoint) + } + + // The next spill GC cycle retries the failed query-directory deletion and refreshes the metric. + waitForSpillDirectoryState(false, "after query cleanup") +}