This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 3dff55a962e [fix](be) Clean up spill directories during query teardown
(#66328)
3dff55a962e is described below
commit 3dff55a962e4609e6f0ad8f731183b8f9d88ea91
Author: Jerry Hu <[email protected]>
AuthorDate: Wed Aug 5 09:03:58 2026 +0800
[fix](be) Clean up spill directories during query teardown (#66328)
## What problem does this PR solve?
Per-operator spill cleanup can leave empty per-query directories or
residual files. `QueryContext` had no reliable cleanup path for all
spill roots touched by a query. In addition, rerunnable recursive CTE
entries retained a shared `QueryContext` reference after successful
`FINAL_CLOSE`, so normal completion could not release the `QueryContext`
and trigger spill cleanup.
This change:
- records each spill data root before the query opens its first spill
part;
- recursively deletes the touched per-query spill directories
synchronously during `QueryContext` teardown;
- retains failed deletions and retries them during later spill GC and
shutdown paths without a fixed retry limit;
- appends retry failures under the queue lock so directories enqueued
while a retry batch is processed are not lost;
- detaches recursive CTE rerunnable entries on `FINAL_CLOSE`, query
removal, and `FragmentMgr` shutdown, and destroys them after releasing
`_rerunnable_params_lock`.
### Release note
None
### Check List (For Author)
- Test: Unit Test
- `./run-be-ut.sh --run
--filter='SpillFileTest.GCCleansUpFiles:SpillFileTest.QueryContextDeletesEmptySpillDirectory:SpillFileTest.QueryContextCleansUpNestedSpillDirectory:SpillFileTest.QueryContextDeletesResidualSpillDirectory:SpillFileTest.QueryContextCleansUpAllTouchedSpillDirectories:SpillFileTest.QueryContextContinuesCleanupAfterRootFailure:SpillFileTest.QueryContextRetriesSpillDirectoryDeletionUntilSuccess:SpillFileTest.RetryPreservesDirectoryQueuedAfterPendingDrain:SpillFileTest.FinalCloseReleasesReru
[...]
(11/11 passed)
- `./run-be-ut.sh --run
--filter='FragmentMgrRerunnableParamsTest.StopReleasesLastQueryContextRefOutsideLock'`
(1/1 passed)
- `build-support/check-format.sh` (passed)
- `git diff --check` (passed)
- Static analysis:
- `build-support/run-clang-tidy.sh --build-dir be/ut_build_ASAN` was
attempted, but the local analysis was blocked by the toolchain failing
to find `stddef.h` and an existing unmatched `NOLINTEND` in
`be/src/core/types.h`.
- Behavior changed: Yes. Touched per-query spill directories are removed
synchronously during `QueryContext` teardown; transient deletion
failures are retried, and successful recursive CTE final close releases
rerunnable state.
- Does this need documentation: No
---
be/src/exec/operator/rec_cte_source_operator.h | 8 +-
be/src/exec/pipeline/pipeline_fragment_context.cpp | 8 +-
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/fragment_mgr.cpp | 34 +-
be/src/runtime/fragment_mgr.h | 6 +-
be/src/runtime/query_context.cpp | 11 +
be/src/runtime/query_context.h | 9 +
.../fragment_mgr_cross_cluster_cancel_test.cpp | 37 ++
be/test/vec/spill/spill_file_test.cpp | 497 ++++++++++++++++++++-
12 files changed, 691 insertions(+), 36 deletions(-)
diff --git a/be/src/exec/operator/rec_cte_source_operator.h
b/be/src/exec/operator/rec_cte_source_operator.h
index 284f754f97d..7f12f254922 100644
--- a/be/src/exec/operator/rec_cte_source_operator.h
+++ b/be/src/exec/operator/rec_cte_source_operator.h
@@ -158,16 +158,14 @@ private:
// │ final_close (=4) │
// │ • listen_wait_close(guard, │
// │ need_send_report = true) │
-// │ • notify_close() on PFC │
// │ • send final status report │
-// │ • clean up completely │
+// │ • remove saved rerunnable params │
+// │ • notify_close() on PFC │
// └──────────────┬─────────────────────┘
// │
// ▼
// ┌────────────────────────────────────┐
// │ Fragment Destroyed │
-// │ rerunnable_params removed on │
-// │ query end │
// └────────────────────────────────────┘
//
class RecCTESourceOperatorX : public OperatorX<RecCTESourceLocalState> {
@@ -326,4 +324,4 @@ private:
bool _is_used_by_other_rec_cte = false;
};
-} // namespace doris
\ No newline at end of file
+} // namespace doris
diff --git a/be/src/exec/pipeline/pipeline_fragment_context.cpp
b/be/src/exec/pipeline/pipeline_fragment_context.cpp
index def2cf473ae..91991ce3aa2 100644
--- a/be/src/exec/pipeline/pipeline_fragment_context.cpp
+++ b/be/src/exec/pipeline/pipeline_fragment_context.cpp
@@ -173,8 +173,8 @@ bool PipelineFragmentContext::is_timeout(timespec now)
const {
}
// notify_close() transitions the PFC from "waiting for external close
notification" to
-// "self-managed close". For recursive CTE fragments, the old PFC is kept
alive until
-// the rerun_fragment(wait_for_destroy) RPC calls this to trigger shutdown.
+// "self-managed close". A recursive CTE PFC normally remains registered until
rerun_fragment()
+// calls this for WAIT_FOR_DESTROY or FINAL_CLOSE; cancellation can also call
it.
// Returns true if all tasks have already closed (i.e., the PFC can be safely
destroyed).
bool PipelineFragmentContext::notify_close() {
bool all_closed = false;
@@ -183,7 +183,7 @@ bool PipelineFragmentContext::notify_close() {
std::lock_guard<std::mutex> l(_task_mutex);
if (_closed_tasks >= _total_tasks) {
if (_need_notify_close) {
- // Fragment was cancelled and waiting for notify to close.
+ // The fragment finished while waiting for the external close
notification.
// Record that we need to remove from fragment mgr, but do it
// after releasing _task_mutex to avoid ABBA deadlock with
// dump_pipeline_tasks() (which acquires _pipeline_map lock
@@ -192,7 +192,7 @@ bool PipelineFragmentContext::notify_close() {
}
all_closed = true;
}
- // make fragment release by self after cancel
+ // Allow the fragment to be removed now or after its remaining tasks
close.
_need_notify_close = false;
}
if (need_remove) {
diff --git a/be/src/exec/spill/spill_file.cpp b/be/src/exec/spill/spill_file.cpp
index d49d5aead1d..be9a32b61f3 100644
--- a/be/src/exec/spill/spill_file.cpp
+++ b/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 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 c74eeee9d02..eb56fb14a13 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"
@@ -38,6 +40,11 @@
namespace doris {
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);
}
@@ -45,6 +52,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());
@@ -97,7 +115,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))) {
@@ -162,6 +180,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;
@@ -181,6 +259,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();
@@ -252,12 +331,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 b90e1760236..1e3042d6ad3 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"
@@ -118,12 +120,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,
@@ -133,9 +130,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); }
@@ -143,9 +144,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;
@@ -153,6 +161,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 d68d52c9640..a2dde22ddf1 100644
--- a/be/src/exec/spill/spill_file_writer.cpp
+++ b/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());
}
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index 44c407875c5..ca2a700900b 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/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*) {}
@@ -429,18 +433,23 @@ 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.
+ // Successful FINAL_CLOSE removes rerunnable entries one fragment at a
time. Detach any entries
+ // that remain when the query is removed, such as on cancellation.
+ 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;
}
}
}
+ // Destroy detached entries outside _rerunnable_params_lock. Releasing
their last QueryContext
+ // reference can run QueryContext::~QueryContext(), which calls this
method.
+ rerunnable_params_map.clear();
_query_ctx_map_delay_delete.erase(key);
#ifndef BE_TEST
_query_ctx_map.erase(key);
@@ -1253,10 +1262,10 @@ 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: send final report, detach saved params, trigger the last PFC
close
//
-// 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
+// During wait_for_destroy, 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
// synchronization point to know when the old PFC has finished all its tasks.
Status FragmentMgr::rerun_fragment(const std::shared_ptr<brpc::ClosureGuard>&
guard,
const TUniqueId& query_id, int fragment_id,
@@ -1294,6 +1303,17 @@ 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) {
+ // Detach the entry under the lock and destroy it afterward. This
keeps the invariant
+ // that RerunableFragmentInfo, which owns QueryContext, is never
destroyed while holding
+ // _rerunnable_params_lock.
+ 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) {
diff --git a/be/src/runtime/fragment_mgr.h b/be/src/runtime/fragment_mgr.h
index ab78c18555a..9065ae9e5a8 100644
--- a/be/src/runtime/fragment_mgr.h
+++ b/be/src/runtime/fragment_mgr.h
@@ -228,8 +228,10 @@ private:
// 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 are detached under _rerunnable_params_lock
and destroyed afterward;
+ // releasing their last QueryContext reference can call
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.
diff --git a/be/src/runtime/query_context.cpp b/be/src/runtime/query_context.cpp
index bc0e7ed1502..3f7ef26e40c 100644
--- a/be/src/runtime/query_context.cpp
+++ b/be/src/runtime/query_context.cpp
@@ -225,6 +225,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,
@@ -253,6 +258,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 2dedffb7576..5d327c2f348 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"
@@ -55,6 +56,7 @@ class PipelineTask;
class QueryTaskController;
class Dependency;
class RecCTEScanLocalState;
+class SpillDataDir;
struct ReportStatusRequest {
const Status status;
@@ -203,6 +205,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();
@@ -329,6 +335,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/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
b/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
index 70dd2e874ba..85de9266100 100644
--- a/be/test/runtime/fragment_mgr_cross_cluster_cancel_test.cpp
+++ b/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
diff --git a/be/test/vec/spill/spill_file_test.cpp
b/be/test/vec/spill/spill_file_test.cpp
index 09173a7a343..9df4bf626f3 100644
--- a/be/test/vec/spill/spill_file_test.cpp
+++ b/be/test/vec/spill/spill_file_test.cpp
@@ -21,22 +21,32 @@
#include <algorithm>
#include <filesystem>
+#include <functional>
#include <memory>
#include <numeric>
+#include <set>
+#include <string>
#include <vector>
#include "common/config.h"
#include "core/block/block.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
+#include "exec/pipeline/pipeline_fragment_context.h"
#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/fragment_mgr.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 {
@@ -78,13 +88,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();
@@ -93,11 +111,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) {
@@ -111,12 +131,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;
};
// ═══════════════════════════════════════════════════════════════════════
@@ -889,7 +955,420 @@ 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,
FinalCloseReleasesRerunnableQueryContextAndDeletesSpillDirectory) {
+ ExecEnv::GetInstance()->spill_file_mgr()->stop();
+
+ 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.hi = 13;
+ query_id.lo = 14;
+ constexpr int first_fragment_id = 1;
+ constexpr int second_fragment_id = 2;
+ Defer restore_fragment_mgr([&] {
+ fragment_mgr->remove_query_context(query_id);
+ fragment_mgr->remove_pipeline_context({query_id, first_fragment_id});
+ fragment_mgr->remove_pipeline_context({query_id, second_fragment_id});
+ fragment_mgr->stop();
+ delete fragment_mgr;
+ exec_env->_fragment_mgr = previous_fragment_mgr;
+ });
+
+ auto query_ctx = MockQueryContext::create(query_id);
+ std::weak_ptr<QueryContext> weak_query_ctx = query_ctx;
+ const auto query_id_str = print_id(query_id);
+ const 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,
"recursive_cte");
+ _create_residual_file(query_dir + "/residual/temporary-data");
+
+ auto create_fragment_context = [&](int fragment_id) {
+ TPipelineFragmentParams params;
+ params.__set_query_id(query_id);
+ params.__set_fragment_id(fragment_id);
+ params.__set_need_notify_close(true);
+ auto context = std::make_shared<PipelineFragmentContext>(
+ query_id, params, query_ctx, exec_env, [](RuntimeState*,
Status*) {});
+ fragment_mgr->_pipeline_map.insert({query_id, fragment_id}, context);
+ g_fragment_executing_count << 1;
+ query_ctx->set_pipeline_context(fragment_id, context);
+ {
+ std::lock_guard lock(fragment_mgr->_rerunnable_params_lock);
+ auto& info = fragment_mgr->_rerunnable_params_map[{query_id,
fragment_id}];
+ info.params = params;
+ info.query_ctx = query_ctx;
+ }
+ return context;
+ };
+
+ auto first_fragment_ctx = create_fragment_context(first_fragment_id);
+ auto second_fragment_ctx = create_fragment_context(second_fragment_id);
+ query_ctx.reset();
+ first_fragment_ctx.reset();
+ second_fragment_ctx.reset();
+
+ ASSERT_FALSE(weak_query_ctx.expired());
+ auto st = fragment_mgr->rerun_fragment({}, query_id, first_fragment_id,
+ PRerunFragmentParams::FINAL_CLOSE);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ EXPECT_EQ(fragment_mgr->_pipeline_map.num_items(), 1);
+ {
+ std::lock_guard lock(fragment_mgr->_rerunnable_params_lock);
+ EXPECT_FALSE(fragment_mgr->_rerunnable_params_map.contains({query_id,
first_fragment_id}));
+ EXPECT_TRUE(fragment_mgr->_rerunnable_params_map.contains({query_id,
second_fragment_id}));
+ }
+ EXPECT_FALSE(weak_query_ctx.expired());
+
+ bool exists = false;
+ st = io::global_local_filesystem()->exists(query_dir, &exists);
+ ASSERT_TRUE(st.ok());
+ EXPECT_TRUE(exists);
+
+ st = fragment_mgr->rerun_fragment({}, query_id, second_fragment_id,
+ PRerunFragmentParams::FINAL_CLOSE);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ EXPECT_EQ(fragment_mgr->_pipeline_map.num_items(), 0);
+ {
+ std::lock_guard lock(fragment_mgr->_rerunnable_params_lock);
+ EXPECT_TRUE(fragment_mgr->_rerunnable_params_map.empty());
+ }
+ EXPECT_TRUE(weak_query_ctx.expired());
+
+ st = io::global_local_filesystem()->exists(query_dir, &exists);
+ ASSERT_TRUE(st.ok());
+ EXPECT_FALSE(exists);
+
+ EXPECT_EQ(_gc_subdirectories(_data_dir_ptr), gc_subdirectories_before);
+}
+
+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);
@@ -906,11 +1385,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);
}
// ═══════════════════════════════════════════════════════════════════════
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
