mrhhsg commented on code in PR #68032:
URL: https://github.com/apache/doris/pull/68032#discussion_r4279090307
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -178,11 +180,11 @@ private List<AuditEvent> getQueryNeedAudit() {
if (waitTimeMs <= queryAuditLogTimeout) {
continue;
}
- Set<Long> expectedBackendIds =
externalDmlAuditBackendIds.get(ae);
+ Set<Long> expectedBackendIds = auditBackendIds.get(ae);
Review Comment:
Resolving, unchanged: Flight audit timing is not changed by this PR. On
master a Flight statement's audit event is released after
`query_audit_log_timeout_ms` with whatever BE statistics have arrived, for
every field; this PR only makes events with known participating BEs wait for
their final reports, bounded by `be_report_query_statistics_timeout_ms`.
Deferring Flight audits until DoGet finishes is a separate change to Flight
auditing.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowDataCommand.java:
##########
@@ -585,8 +594,22 @@ public String toString() {
return toSql();
}
+ /**
+ * Bytes of query spill currently held in object storage, summed over the
alive BEs from their
+ * latest disk reports. Only the master FE keeps these values, so the
command is forwarded to it.
+ */
+ private long getRemoteSpillSize() throws AnalysisException {
+ long total = 0;
+ for (Backend backend :
Env.getCurrentSystemInfo().getAllBackendsByAllCluster().values()) {
+ if (backend.isAlive()) {
Review Comment:
Resolving, kept on purpose: a BE the master considers dead may be down, and
its last report would otherwise be counted indefinitely (a gracefully stopped
BE has usually deleted its spill, so a stale non-zero value would overcount).
Heartbeat failing while disk reports keep succeeding is transient; the BE is
counted again as soon as its heartbeat recovers.
##########
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:
Fixed in 2c17ec7ee0b: `_retry_pending_spill_directories` now puts the
untried and the failed entries back as one batch under the mutex (untried
first, as before; they were already merged with each other when queued), and
only the directories queued while the pass ran go through
`_add_pending_directory`. A pass therefore costs one rescan per newly queued
directory instead of one per kept entry; `stop()` and the destructor use the
same function. Covered by
`SpillFileS3Test.PendingRemoteDeleteBacklogIsRequeuedInBulk`: 16 queued
spill-file prefixes, a slow failing object store, and a duplicate plus a new
directory queued during the pass (via the
`retry_pending_spill_directories_after_drain` debug point); it checks that the
backlog is kept in order, the duplicate is merged and the new one is added,
then that everything is deleted.
##########
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:
Not changed: this is not specific to the remote pair. The final
`PQueryStatistics` of a result set is filled by `GetResultBatchCtx::on_close()`
(`be/src/exec/sink/writer/vmysql_result_writer.cpp`) with `returned_rows` only,
so on master the scan bytes, CPU time, peak memory and local spill bytes of the
audit event all come from the same `WorkloadRuntimeStatusMgr` BE reports
(`runAfterCatalogReady` overwrites them from `queryStatisticsMap`) and are lost
in exactly the same way when `clearReportTimeoutBeStatistics()` evicts an idle
BE's entry before the audit event is queued. This PR adds two fields to that
path and makes the audit wait for the participants' final reports while they
are there. Retaining final statistics through result delivery is a change to
the statistics manager for every field and belongs in its own PR.
--
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]
