github-actions[bot] commented on code in PR #68032:
URL: https://github.com/apache/doris/pull/68032#discussion_r4259921833
##########
be/src/exec/spill/spill_file_manager.cpp:
##########
@@ -345,68 +489,153 @@ void SpillFileManager::delete_spill_file(SpillFileSPtr
spill_file) {
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),
+ if (data_dir->is_remote()) {
+ // Remote spill is deleted per spill file, when the SpillFile is
destroyed (retried by the
+ // GC thread when that failed); nothing is deleted by directory on
object storage.
+ return;
+ }
+ PendingSpillDirectory pending_directory {
+ .failed_count = 0,
+ .dir = data_dir->get_spill_data_path(query_id),
+ .data_dir = data_dir,
+ .fs = nullptr,
+ .charged_bytes = 0,
+ .persisted_bytes = 0,
};
- auto status = _try_delete_query_spill_directory(pending_directory);
+ auto status = _try_delete_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));
+ _add_pending_directory(std::move(pending_directory));
+ }
+}
+
+void SpillFileManager::retry_spill_directory_deletion(SpillDataDir* data_dir,
io::FileSystemSPtr fs,
+ std::string dir, int64_t
charged_bytes,
+ int64_t persisted_bytes)
{
+ _add_pending_directory({.failed_count = 1,
+ .dir = std::move(dir),
+ .data_dir = data_dir,
+ .fs = std::move(fs),
+ .charged_bytes = charged_bytes,
+ .persisted_bytes = persisted_bytes});
+}
+
+void SpillFileManager::_add_pending_directory(PendingSpillDirectory
pending_directory) {
+ auto is_under = [](const std::string& dir, const std::string& ancestor) {
+ return dir.size() > ancestor.size() && dir.starts_with(ancestor) &&
+ dir[ancestor.size()] == '/';
+ };
+ std::lock_guard lock(_pending_spill_directories_mutex);
+ for (auto& pending : _pending_spill_directories) {
+ if (pending.data_dir == pending_directory.data_dir && pending.fs ==
pending_directory.fs &&
+ (pending.dir == pending_directory.dir ||
+ is_under(pending_directory.dir, pending.dir))) {
+ // Deleting the pending ancestor deletes these objects too.
+ pending.charged_bytes += pending_directory.charged_bytes;
+ pending.persisted_bytes += pending_directory.persisted_bytes;
+ return;
+ }
}
+ std::erase_if(_pending_spill_directories, [&](const PendingSpillDirectory&
pending) {
+ if (pending.data_dir == pending_directory.data_dir && pending.fs ==
pending_directory.fs &&
+ is_under(pending.dir, pending_directory.dir)) {
+ pending_directory.charged_bytes += pending.charged_bytes;
+ pending_directory.persisted_bytes += pending.persisted_bytes;
+ return true;
+ }
+ return false;
+ });
+ _pending_spill_directories.emplace_back(std::move(pending_directory));
}
-Status SpillFileManager::_try_delete_query_spill_directory(
- const PendingQuerySpillDirectory& pending_directory) {
+Status SpillFileManager::_try_delete_spill_directory(
+ const PendingSpillDirectory& pending_directory) {
{
- std::lock_guard lock(_pending_query_spill_directories_mutex);
- if
(_external_spill_directory_leases.contains(pending_directory.query_dir)) {
+ std::lock_guard lock(_pending_spill_directories_mutex);
+ if (_external_spill_directory_leases.contains(pending_directory.dir)) {
return Status::InternalError("external spill directory is still in
use: {}",
- pending_directory.query_dir);
+ pending_directory.dir);
}
}
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);
+ // Every pending directory was written through the store, which was ready
by then and
+ // never goes back.
+ auto fs = pending_directory.fs != nullptr ? pending_directory.fs
+ :
pending_directory.data_dir->fs();
+ DORIS_CHECK(fs != nullptr) << "spill store " <<
pending_directory.data_dir->path()
+ << " is not ready";
+ return fs->delete_directory(pending_directory.dir);
+}
+
+void SpillFileManager::_retry_pending_local_spill_directories() {
+ MonotonicStopWatch watch;
+ watch.start();
+ _retry_pending_spill_directories(watch,
std::numeric_limits<int64_t>::max(),
+ /*retry_remote=*/false);
}
-void SpillFileManager::_retry_pending_query_spill_directories() {
- std::vector<PendingQuerySpillDirectory> pending_directories;
+void SpillFileManager::_retry_pending_spill_directories(const
MonotonicStopWatch& watch,
+ int64_t
max_work_time_ns,
+ bool retry_remote) {
+ std::vector<PendingSpillDirectory> pending_directories;
{
- std::lock_guard lock(_pending_query_spill_directories_mutex);
- pending_directories.swap(_pending_query_spill_directories);
+ std::lock_guard lock(_pending_spill_directories_mutex);
+ pending_directories.swap(_pending_spill_directories);
}
- DBUG_EXECUTE_IF(
-
"fault_inject::spill_file_manager::retry_pending_query_spill_directories_after_drain",
- { DBUG_RUN_CALLBACK(); });
+
DBUG_EXECUTE_IF("fault_inject::spill_file_manager::retry_pending_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;
+ std::vector<PendingSpillDirectory> untried_directories;
+ std::vector<PendingSpillDirectory> failed_directories;
+ size_t tried = 0;
for (auto& pending_directory : pending_directories) {
- auto status = _try_delete_query_spill_directory(pending_directory);
+ // A remote deletion can wait for a request timeout: after the time
slice is used up the
+ // remaining directories are queued again without being tried, ahead
of the ones that
+ // just failed, so that a slow failing directory cannot starve the
others.
+ if ((!retry_remote && pending_directory.data_dir->is_remote()) ||
+ (tried > 0 && watch.elapsed_time() >= max_work_time_ns)) {
+ untried_directories.emplace_back(std::move(pending_directory));
+ continue;
+ }
+ ++tried;
+ auto status = _try_delete_spill_directory(pending_directory);
if (status.ok()) {
+
pending_directory.data_dir->release(pending_directory.charged_bytes);
+
pending_directory.data_dir->release_persisted_bytes(pending_directory.persisted_bytes);
continue;
}
+ if (pending_directory.data_dir->is_remote()) {
+ auto reconcile_status = reconcile_remote_spill_bytes(
+ pending_directory.data_dir, pending_directory.fs,
pending_directory.dir,
+ &pending_directory.persisted_bytes);
+ if (!reconcile_status.ok()) {
+ LOG_EVERY_T(WARNING, 60)
+ << "failed to reconcile remote spill bytes after
deletion: "
+ << reconcile_status;
+ }
+ }
+
++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 to retry deleting spill directory, dir {}, error:
{}",
+ pending_directory.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));
- }
+ for (auto& pending_directory : untried_directories) {
+ _add_pending_directory(std::move(pending_directory));
Review Comment:
[P2] Preserve the backlog without quadratic reinsertion. When remote DELETEs
fail, partitioned spills leave many unrelated per-file prefixes here (remote
query cleanup does not coalesce them). After the first slow retry exhausts the
GC time slice, this loop still re-adds every untouched entry through
_add_pending_directory, which scans the growing vector twice. N surviving
prefixes therefore require N*(N-1) entry visits outside the time budget; 10,000
prefixes cause roughly 100 million visits under the pending-directory mutex on
every GC pass. stop() and the destructor repeat this even while skipping all
remote requests. Preserve/rotate the already-deduplicated backlog in bulk, and
merge concurrent additions without pairwise rescanning; include a large-backlog
case alongside the current three-entry test. This is separate from the existing
network-timeout retry finding.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/AuditLogHelper.java:
##########
@@ -274,6 +274,10 @@ private static void logAuditLogImpl(ConnectContext ctx,
String origStmt, Stateme
statistics.getSpillWriteBytesToLocalStorage())
.setSpillReadBytesFromLocalStorage(statistics == null ? 0 :
statistics.getSpillReadBytesFromLocalStorage())
+ // Remote spill totals arrive in final BE reports;
WorkloadRuntimeStatusMgr
+ // waits for the participants, with a bounded fallback if a BE
never reports.
+ .setSpillWriteBytesToRemoteStorage(0)
Review Comment:
[P2] Retain final remote totals until audit consumes them. A MySQL query can
finish its BE execution while the FE is still sending the last buffered rows to
a slow client. After its final statistics RPC succeeds,
RuntimeQueryStatisticsMgr removes the query and an otherwise idle BE stops
reporting; clearReportTimeoutBeStatistics then drops its entire FE entry after
the default 60 seconds, even though the query is still registered. A 90-second
client stall is allowed by the default 600-second net_write_timeout. When
result delivery finally completes, these zero-seeded fields are audited after
the participant wait expires: the final report already arrived and was evicted,
and PQueryStatistics carries no remote spill fallback. Retain final statistics
through result delivery and pending audit consumption, or transport the remote
pair in the final result statistics. Add an
early-final-report/slow-result-delivery test; the current tests only delay
final reports until after audit is queue
d.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
