Gabriel39 commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3845012323
##########
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:
Resolved by separating process incarnation from per-process generation. Each
report carries the same backend start time published by heartbeat; FE clears or
replaces the complete per-BE shell whenever heartbeat exposes a different
incarnation, using equality rather than wall-clock ordering. A lower-start-time
restart regression test covers this case.
##########
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:
Resolved by removing synchronous prepare-failure statistics callbacks from
both prepare paths. The execPlanFragment response no longer waits on nested
Thrift delivery; the registered ResourceContext remains in the periodic retry
path, while FE waits for expected BE final snapshots with a bounded fallback.
--
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]
