github-actions[bot] commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3844250410
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -194,16 +196,50 @@ bool PipelineFragmentContext::notify_close() {
need_remove = true;
}
all_closed = true;
+ if (_send_report_on_close && !_final_report_submitted) {
+ _final_report_submitted = true;
+ send_final_report = true;
+ }
}
// Allow the fragment to be removed now or after its remaining tasks
close.
_need_notify_close = false;
}
+ if (send_final_report) {
+ // Submit while the fragment map still owns this context because
send_report() obtains a
+ // shared_ptr to keep the asynchronous callback alive.
+ auto st = send_report(true);
+ if (!st.ok()) {
+ LOG_WARNING("Failed to send recursive CTE final report for query
{}, fragment {}: {}",
+ print_id(_query_id), _fragment_id, st.to_string());
+ }
+ }
if (need_remove) {
_exec_env->fragment_mgr()->remove_pipeline_context({_query_id,
_fragment_id});
}
return all_closed;
}
+void PipelineFragmentContext::report_prepare_failure(const Status&
prepare_status) {
+ if (_runtime_state == nullptr || _runtime_state->query_type() !=
TQueryType::LOAD) {
+ return;
+ }
+ // A prepare failure has no later task-close callback, so establish the
same
+ // statistics-before-audit edge as a normal final fragment report before
returning it to FE.
+ ReportStatusRequest req {.status = prepare_status,
+ .runtime_states = {},
+ .done = true,
+ .coord_addr = _query_ctx->coord_addr,
+ .query_id = _query_id,
+ .fragment_id = _fragment_id,
+ .fragment_instance_id = TUniqueId(),
+ .backend_num = -1,
+ .runtime_state = _runtime_state.get(),
+ .load_error_url = "",
+ .first_error_msg =
std::string(prepare_status.msg()),
+ .cancel_fn = [](const Status&) {}};
+ _coordinator_callback(req);
Review Comment:
[P1] Preserve the prepare-failure publication edge when this callback stalls
or fails. This synchronous ReportExecStatus can wait for the 60-second BE
Thrift timeout, while FE waits only 30 seconds for the independent
execPlanFragment BRPC; FE can therefore time out that RPC, finish the failed
DML, and enqueue its audit while statistics are still in flight.
Client/transport failures are also discarded because this returns void with a
no-op `cancel_fn`. Return/handle a delivery result or retain a retry/defer
token that prevents audit readiness until FE accepts the snapshot, and cover
both prepare paths.
##########
be/src/runtime/runtime_query_statistics_mgr.cpp:
##########
@@ -83,35 +79,37 @@ static Status _do_report_exec_stats_rpc(const
TNetworkAddress& coor_addr,
}
} catch (apache::thrift::TApplicationException& e) {
if (e.getType() == e.UNKNOWN_METHOD) {
- LOG_WARNING(
- "Failed to report query profile to {} due to {}, usually
because the frontend "
- "is not upgraded, check the version",
- PrintThriftNetworkAddress(coor_addr), e.what());
+ LOG_WARNING("Execution statistics report to {} failed: {}; FE may
not be upgraded",
+ PrintThriftNetworkAddress(fe_addr), e.what());
} else {
- LOG_WARNING(
- "Failed to report query profile to {}, reason: {}, you can
see fe log for "
- "details.",
- PrintThriftNetworkAddress(coor_addr), e.what());
+ LOG_WARNING("Execution statistics report to {} failed: {}",
+ PrintThriftNetworkAddress(fe_addr), e.what());
}
return Status::RpcError("Send stats failed");
} catch (apache::thrift::TException& e) {
- LOG_WARNING("Failed to report query profile to {}, reason: {} ",
- PrintThriftNetworkAddress(coor_addr), e.what());
+ LOG_WARNING("Failed to report execution statistics to {}, reason: {} ",
+ PrintThriftNetworkAddress(fe_addr), e.what());
std::this_thread::sleep_for(
std::chrono::milliseconds(config::thrift_client_retry_interval_ms * 2));
// just reopen to disable this connection
static_cast<void>(rpc_client.reopen(config::thrift_rpc_timeout_ms));
- return Status::RpcError("Transport exception when report query
profile");
+ return Status::RpcError("Transport exception when reporting execution
statistics");
} catch (std::exception& e) {
LOG_WARNING(
- "Failed to report query profile to {}, reason: {}, you can see
fe log for details.",
- PrintThriftNetworkAddress(coor_addr), e.what());
- return Status::RpcError("Send report query profile failed");
+ "Failed to report execution statistics to {}, reason: {}, you
can see fe log for "
+ "details.",
+ PrintThriftNetworkAddress(fe_addr), e.what());
+ return Status::RpcError("Send execution statistics failed");
}
return Status::OK();
}
+// Reserve low bits for per-process generations while wall-clock high bits
keep restarted BEs
+// newer than statistics that FE may still cache for the same backend ID.
+RuntimeQueryStatisticsMgr::RuntimeQueryStatisticsMgr()
+ : _next_query_statistics_generation(UnixMillis() << 20) {}
Review Comment:
[P2] Do not use a reversible clock as the restart epoch. `UnixMillis()` uses
CLOCK_REALTIME and may move backward, so a restarted BE can seed a lower
generation than the value FE still retains for the same backend/query key;
`shouldReplaceQueryStatistics()` then rejects every report from the new process
regardless of sequence. Use a restart/incarnation value that cannot decrease
for a stable backend ID (or clear state on an observed incarnation change), and
test a retained key with a lower restart seed.
##########
fe/fe-core/src/main/java/org/apache/doris/resource/workloadschedpolicy/WorkloadRuntimeStatusMgr.java:
##########
@@ -154,24 +128,31 @@ public void submitFinishQueryToAudit(AuditEvent event) {
}
}
- private List<AuditEvent> getQueryNeedAudit() {
+ @VisibleForTesting
+ List<AuditEvent> getQueryNeedAudit() {
List<AuditEvent> ret = new ArrayList<>();
long currentTime = System.currentTimeMillis();
- queryAuditEventLogWriteLock();
- try {
- int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
- Iterator<AuditEvent> iter = queryAuditEventList.iterator();
- while (iter.hasNext()) {
- AuditEvent ae = iter.next();
- if (currentTime - ae.pushToAuditLogQueueTime >
queryAuditLogTimeout) {
- ret.add(ae);
- iter.remove();
- } else {
+ int queryAuditLogTimeout = Config.query_audit_log_timeout_ms;
+ while (true) {
+ AuditEvent auditEvent;
+ queryAuditEventLogWriteLock();
+ try {
+ if (queryAuditEventList.isEmpty()) {
break;
}
+ auditEvent = queryAuditEventList.get(0);
+ if (currentTime - auditEvent.pushToAuditLogQueueTime <=
queryAuditLogTimeout) {
+ break;
+ }
+ queryAuditEventList.remove(0);
+ } finally {
+ queryAuditEventLogWriteUnlock();
}
- } finally {
- queryAuditEventLogWriteUnlock();
+ // Fragment completion updates happen before the coordinator can
enqueue this event.
+ // Reading the live map here preserves that ordering without
holding the queue lock.
+ applyQueryStatisticsToAuditEvent(auditEvent,
Review Comment:
[P2] Batch live audit hydration. This calls `buildQueryStatistics()` for
every due event, and that helper scans every retained BE, changing the previous
one-merge-plus-O(1)-lookups cycle into O(events * backends) before the daemon
can dispatch the batch; the configured queue can hold 250,000 events. Drain the
due IDs and aggregate them in one BE-map traversal (or maintain a per-query
aggregate), while preserving the completion-before-enqueue edge.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -194,16 +196,50 @@ bool PipelineFragmentContext::notify_close() {
need_remove = true;
}
all_closed = true;
+ if (_send_report_on_close && !_final_report_submitted) {
+ _final_report_submitted = true;
+ send_final_report = true;
+ }
}
// Allow the fragment to be removed now or after its remaining tasks
close.
_need_notify_close = false;
}
+ if (send_final_report) {
+ // Submit while the fragment map still owns this context because
send_report() obtains a
+ // shared_ptr to keep the asynchronous callback alive.
+ auto st = send_report(true);
Review Comment:
[P1] Propagate a rejected recursive final callback. FINAL_CLOSE now consumes
the one-shot before `send_report()`, logs a rejected bounded-pool submission,
can remove the PFC, and still returns OK through `rerun_fragment`. The
recursive source therefore cannot retry while FE's fragment-done latch never
receives this terminal report and waits until timeout. Do not
remove/acknowledge FINAL_CLOSE until submission is accepted (or retain
retryable state), and cover deterministic pool rejection.
--
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]
