This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 3ba435c26e0 branch-4.1: [fix](be) Clean up spill directories during
query teardown (#66458)
3ba435c26e0 is described below
commit 3ba435c26e061fd22fd80fb5f3d84db12b2b28f0
Author: Jerry Hu <[email protected]>
AuthorDate: Thu Aug 6 17:23:07 2026 +0800
branch-4.1: [fix](be) Clean up spill directories during query teardown
(#66458)
### What problem does this PR solve?
Issue Number: None
Related PR: #66328
Problem Summary:
Backport the spill query-directory cleanup from #66328 to `branch-4.1`.
This change:
- records each spill data root before opening the first spill part;
- deletes touched per-query spill directories during `QueryContext`
teardown;
- retains failed deletions and retries them from spill GC and shutdown
paths;
- preserves the no-spill fast path without scanning every configured
spill root.
The recursive CTE and `FragmentMgr` lifecycle changes from the original
PR are intentionally excluded because the `branch-4.1` implementation
diverges. Spill cleanup for query contexts retained by that lifecycle
remains out of scope for this backport.
### Release note
None
### Check List (For Author)
- Test
- [x] Regression test
- ASAN FE/BE build with `./build.sh --be --fe` succeeded.
- `doris-local-regression run -d spill_p0 -s
test_spill_directory_cleanup` (1/1 passed, then rerun 1/1 passed).
- The case confirms real spill writes and observes
`spill_disk_has_spill_data` transition from `0` to `1` and back to `0`
after query cleanup.
- [x] Unit Test
- ASAN `doris_be_test` target compiled and linked successfully.
- `./run-be-ut.sh --run
--filter='SpillFileTest.GCCleansUpFiles:SpillFileTest.QueryContextDeletesEmptySpillDirectory:SpillFileTest.QueryContextCleansUpNestedSpillDirectory:SpillFileTest.QueryContextDeletesResidualSpillDirectory:SpillFileTest.QueryContextCleansUpAllTouchedSpillDirectories:SpillFileTest.QueryContextContinuesCleanupAfterRootFailure:SpillFileTest.QueryContextRetriesSpillDirectoryDeletionUntilSuccess:SpillFileTest.RetryPreservesDirectoryQueuedAfterPendingDrain:SpillFileTest.QueryContextSkipsClean
[...]
-j 64` (10/10 passed, then rerun 10/10 passed)
- `build-support/check-format.sh` (passed)
- `git diff --check` (passed)
- [ ] 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
- Behavior changed:
- [ ] No.
- [x] Yes. Touched spill query directories are removed during query
teardown, and transient deletion failures are retained for retry.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/exec/spill/spill_file.cpp | 4 +-
be/src/exec/spill/spill_file_manager.cpp | 85 ++++-
be/src/exec/spill/spill_file_manager.h | 25 +-
be/src/exec/spill/spill_file_writer.cpp | 3 +
be/src/runtime/query_context.cpp | 11 +
be/src/runtime/query_context.h | 9 +
be/test/vec/spill/spill_file_test.cpp | 406 ++++++++++++++++++++-
.../spill_p0/test_spill_directory_cleanup.groovy | 113 ++++++
8 files changed, 638 insertions(+), 18 deletions(-)
diff --git a/be/src/exec/spill/spill_file.cpp b/be/src/exec/spill/spill_file.cpp
index 78347ea78eb..7f403de57fc 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 0946e3209b7..048e1d2e551 100644
--- a/be/src/exec/spill/spill_file_manager.cpp
+++ b/be/src/exec/spill/spill_file_manager.cpp
@@ -24,6 +24,7 @@
#include <filesystem>
#include <memory>
#include <string>
+#include <utility>
#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<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;
}
diff --git a/be/src/exec/spill/spill_file_manager.h
b/be/src/exec/spill/spill_file_manager.h
index 582df3f1f63..7455789c216 100644
--- a/be/src/exec/spill/spill_file_manager.h
+++ b/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 @@ public:
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 @@ public:
/// 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 @@ public:
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;
@@ -154,6 +162,9 @@ private:
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};
diff --git a/be/src/exec/spill/spill_file_writer.cpp
b/be/src/exec/spill/spill_file_writer.cpp
index ced813e51e3..62320ef9c05 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 249d0ed861f..aaab0ad698c 100644
--- a/be/src/runtime/query_context.cpp
+++ b/be/src/runtime/query_context.cpp
@@ -223,6 +223,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,
@@ -265,6 +270,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 13eb6ffe649..e0368def54e 100644
--- a/be/src/runtime/query_context.h
+++ b/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"
@@ -54,6 +55,7 @@ class PipelineTask;
class QueryTaskController;
class Dependency;
class RecCTEScanLocalState;
+class SpillDataDir;
struct ReportStatusRequest {
const Status status;
@@ -202,6 +204,10 @@ public:
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();
@@ -327,6 +333,9 @@ private:
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();
diff --git a/be/test/vec/spill/spill_file_test.cpp
b/be/test/vec/spill/spill_file_test.cpp
index 67740acdcd2..11b16df1712 100644
--- a/be/test/vec/spill/spill_file_test.cpp
+++ b/be/test/vec/spill/spill_file_test.cpp
@@ -21,8 +21,11 @@
#include <algorithm>
#include <filesystem>
+#include <functional>
#include <memory>
#include <numeric>
+#include <set>
+#include <string>
#include <vector>
#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 @@ protected:
_profile->add_child(_common_profile.get(), true);
_spill_dir = "./ut_dir/spill_file_test";
- auto spill_data_dir = std::make_unique<SpillDataDir>(_spill_dir, 1024L
* 1024 * 128);
+ _second_spill_dir = "./ut_dir/spill_file_test_second";
+ auto spill_data_dir =
+ std::make_unique<SpillDataDir>(_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<SpillDataDir>(
+ _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<std::string, std::unique_ptr<SpillDataDir>>
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 @@ protected:
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<int32_t>& data) {
@@ -112,12 +130,58 @@ protected:
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<MockRuntimeState>(
+ query_id, 0, query_ctx->query_options(), query_globals,
ExecEnv::GetInstance(),
+ query_ctx);
+
+ auto spill_file = std::make_shared<SpillFile>(
+ 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<std::string> _gc_subdirectories(SpillDataDir* data_dir) {
+ std::set<std::string> 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<MockRuntimeState> _runtime_state;
std::unique_ptr<RuntimeProfile> _profile;
std::unique_ptr<RuntimeProfile> _custom_profile;
std::unique_ptr<RuntimeProfile> _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<DebugPoint>();
+ 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<DebugPoint>();
+ 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<DebugPoint>();
+ after_drain_debug_point->execute_limit = 1;
+ after_drain_debug_point->callback = std::function<void()>([&]() {
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 00000000000..ba9b99774fb
--- /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")
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
