github-actions[bot] commented on code in PR #67053:
URL: https://github.com/apache/doris/pull/67053#discussion_r3840943195
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2532,6 +2542,11 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
if (_exec_env->cluster_info()->backend_id != 0) {
Review Comment:
These new per-query callbacks can reach FE concurrently, invalidating
`WorkloadRuntimeStatusMgr`'s explicit assumption that one BE sends updates
sequentially. When that BE has no entry yet, Q1 and Q2 can both `get(null)`,
create different `BeReportInfo`s, overwrite the top-level map, and update one
now-detached per-query map; one final snapshot disappears even though both
callbacks return success and unregister. Make FE initialization atomic
(`computeIfAbsent`/`putIfAbsent`, updating the retained object) and add a
barrier-based two-query test.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2532,6 +2542,11 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
if (_exec_env->cluster_info()->backend_id != 0) {
params.__set_backend_id(_exec_env->cluster_info()->backend_id);
Review Comment:
This piggybacks the final counters on the RPC to `req.coord_addr`, but query
statistics deliberately target `current_connect_fe`
(`TaskController::fe_addr()`), which differs for a forwarded request. If client
FE A proxies an external-table INSERT to coordinator FE B, S1 is stored only in
B's in-memory workload map; this callback succeeds and unregisters the BE
state, while A queues the audit and can see only an older periodic S0 or
nothing. Deliver/acknowledge the final snapshot at `current_connect_fe` (or
explicitly forward it to the audit-owning FE) before unregistering, and cover a
proxy request with different FE addresses.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2532,6 +2542,11 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
if (_exec_env->cluster_info()->backend_id != 0) {
params.__set_backend_id(_exec_env->cluster_info()->backend_id);
}
+ if (req.report_query_statistics) {
Review Comment:
The statistics envelope is optional for audit correctness, but adding it
before `validate_report_exec_status_size()` can turn a final LOAD report that
previously fit into a local size failure. In that boundary case the essential
status/commit metadata is never sent, `cancel_fn` runs, and external-file
ownership is rejected solely because of this extra payload. Validate/preserve
the essential final report first and include statistics only when they fit (or
send them separately), with a test where the base report fits but
base+statistics exceeds the limit.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2604,18 +2619,23 @@ void
PipelineFragmentContext::_coordinator_callback(const ReportStatusRequest& r
LOG_INFO("Going to cancel query {} since report exec status got rpc
failed: {}",
print_id(req.query_id), rpc_status.to_string());
Review Comment:
Unregistering here assumes the acknowledged final snapshot cannot be
superseded, but the periodic path may already hold an older S0 outside
`_resource_contexts_map_lock`. S1 can arrive first and make this erase succeed;
the delayed/retried S0 then arrives after coordinator removal, and FE's
`updateBeQueryStats()` blindly `put()`s it over S1 without a sequence or
terminal-`query_finished` guard. The audit can therefore log the pre-final
counters with no BE entry left to repair them. Add generation-aware monotonic
ordering (or serialize the two publication paths) and test a delayed periodic
S0 arriving after final S1.
##########
be/src/runtime/fragment_mgr.cpp:
##########
@@ -523,6 +523,7 @@ Status FragmentMgr::_get_or_create_query_ctx(const
TPipelineFragmentParams& para
query_ctx = QueryContext::create(
query_id, _exec_env, params.query_options,
params.coord,
params.is_nereids, params.current_connect_fe,
actual_query_source);
+
query_ctx->set_fragment_num_on_host(params.fragment_num_on_host);
Review Comment:
`fragment_num_on_host` is counted in a different unit from this new gate.
Both FE dispatchers count every local fragment instance
(`FInstanceExecParam`/`AssignedJob`), but then group all same-worker instances
of one fragment into one `TPipelineFragmentParams` and one
`PipelineFragmentContext`; BE calls `finish_fragment()` once when that whole
PFC closes. For one fragment with two `local_params`, FE sends 2 here and BE
increments only to 1, so `report_query_statistics` is never set and the audit
falls back to the delayed/destructor path. Count distinct worker/fragment PFCs
(or advance by the grouped instance count), reject/fix zero explicitly, and
test real dispatch with multiple local instances.
--
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]
