github-actions[bot] commented on code in PR #67340:
URL: https://github.com/apache/doris/pull/67340#discussion_r3942493565
##########
be/src/exec/operator/exchange_sink_buffer.cpp:
##########
@@ -346,6 +346,7 @@ Status ExchangeSinkBuffer::_send_rpc(RpcInstance&
instance_data) {
}
// The eos here only indicates that the current exchange sink has
reached eos.
// However, the queue still contains data from other exchange
sinks, so RPCs need to continue being sent.
+ // `_send_rpc` must be the LAST operation in this function,
because it may reuse the callback!
Review Comment:
[P1] Keep the active handler alive across reentrant reuse
This call is not actually the last operation in the handler: on a
synchronous failure from the nested send, lines 351-353 still use this lambda's
captures. For another queued block on the same channel, `_send_rpc()` reuses
the same `ExchangeSendCallback` and `addSuccessHandler()` assigns over
`_suc_fn` while that very target is executing. If HTTP attachment/DNS/client
setup then returns an error before launching the RPC, execution comes back here
and calls `_failed` through a lambda whose stored target has already been
destroyed. Please keep a local copy of the selected handler before invoking it
(or defer handler replacement until it returns); the broadcast branch has the
same pattern.
##########
be/src/exec/runtime_filter/runtime_filter_producer.h:
##########
@@ -17,15 +17,78 @@
#pragma once
+#include <glog/logging.h>
+
+#include <memory>
#include <mutex>
#include "exec/pipeline/dependency.h"
#include "exec/runtime_filter/runtime_filter.h"
#include "runtime/query_context.h"
-#include "runtime/runtime_profile.h"
+#include "util/brpc_closure.h"
namespace doris {
#include "common/compile_check_begin.h"
+
+// Callback for sync-size RPCs. Handles errors (disable wrapper + sub
dependency) in call().
+class SyncSizeCallback : public DummyBrpcCallback<PSendFilterSizeResponse> {
+ ENABLE_FACTORY_CREATOR(SyncSizeCallback);
+
+public:
+ SyncSizeCallback(std::shared_ptr<Dependency> dependency,
+ std::shared_ptr<RuntimeFilterWrapper> wrapper,
+ std::weak_ptr<QueryContext> context)
+ : _dependency(std::move(dependency)), _wrapper(wrapper),
_context(std::move(context)) {}
+
+ void call() override {
+ // On error: disable the wrapper and sub the dependency here, because
set_synced_size()
+ // will never be called (the merge node won't respond with a sync).
+ // On success: do NOT sub here. The merge node will respond with
sync_filter_size,
+ // which calls set_synced_size() -> _dependency->sub().
+ if (this->cntl_->Failed()) {
+ LOG(WARNING) << fmt::format("RPC meet failed: {}",
this->cntl_->ErrorText());
+ if (auto w = _wrapper.lock()) {
+ w->set_state(RuntimeFilterWrapper::State::DISABLED,
this->cntl_->ErrorText());
+ }
+ if (auto ctx = _context.lock()) {
+ if (!ctx->ignore_runtime_filter_error()) {
+ ctx->cancel(
+ Status::NetworkError("RPC meet failed: {}",
this->cntl_->ErrorText()));
+ }
+ }
+ auto p =
std::dynamic_pointer_cast<CountedFinishDependency>(_dependency);
+ DORIS_CHECK(p);
+ p->sub();
+ return;
+ }
+
+ Status status = Status::create(this->response_->status());
+ if (!status.ok()) {
+ LOG(WARNING) << "RPC meet error status: " << status;
+ if (auto w = _wrapper.lock()) {
+ w->set_state(RuntimeFilterWrapper::State::DISABLED,
status.to_string());
+ }
+ if (auto ctx = _context.lock()) {
+ if (!ctx->ignore_runtime_filter_error()) {
+ ctx->cancel(status);
Review Comment:
[P1] Preserve EOF as a non-cancelling sync result
This branch now calls `QueryContext::cancel()` for `END_OF_FILE` whenever
`ignore_runtime_filter_error=false`. That status is a normal late-response
sentinel here: `FragmentMgr::send_filter_size()` returns EOF when the
merge-side query context has already finished. The old `SyncSizeClosure` still
disabled the wrapper and released the dependency, but its base handler
explicitly returned before cancelling on EOF. With this change, an otherwise
healthy local fragment can have all of its pipelines cancelled just because the
remote merge context disappeared first. Please retain the
wrapper-disable/`sub()` cleanup while exempting EOF from logging/cancellation,
and cover `SyncSizeCallback` with the non-ignoring option (the added EOF test
exercises a different callback).
##########
be/src/exec/runtime_filter/runtime_filter_mgr.cpp:
##########
@@ -669,11 +691,14 @@ Status
RuntimeFilterMergeControllerEntity::_send_rf_to_target(
}
auto st = Status::OK();
- for (auto& target : targets) {
+ cnt_val.publish_callbacks.resize(targets.size());
Review Comment:
[P1] Synchronize callback storage with recursive reset
`merge()` releases `cnt_val.mtx` before calling `_send_rf_to_target()`, so
this resize/write is not synchronized with `GlobalMergeContext::reset()`, which
takes that mutex and clears `publish_callbacks`. Waiting for the old PFC to be
destroyed does not drain its fire-and-forget `merge_filter` RPC; an old merge
handler can pass the stage check, drop the mutex, and still be publishing when
the next recursive round resets the context. That permits concurrent
`clear()`/`resize()` on the same vector (or lets an old stage repopulate the
new stage's owner set), which is undefined behavior. Please serialize the whole
publish setup with reset and revalidate the stage, or make callback ownership
per-RPC/self-owned so reset never mutates shared in-flight storage.
##########
be/src/exec/runtime_filter/runtime_filter.cpp:
##########
@@ -36,13 +36,12 @@ Status RuntimeFilter::_push_to_remote(RuntimeState* state,
const TNetworkAddress
auto merge_filter_request = std::make_shared<PMergeFilterRequest>();
merge_filter_request->set_stage(_stage);
- auto merge_filter_callback =
DummyBrpcCallback<PMergeFilterResponse>::create_shared();
+ _merge_filter_callback =
HandleErrorBrpcCallback<PMergeFilterResponse>::create_shared(
Review Comment:
[P2] Release the callback/controller after RPC completion
Storing this callback on the filter also keeps its `brpc::Controller` alive
after `Run()`; the serialized filter is appended to that controller's
`request_attachment()` below. In the pinned brpc 1.4.0 implementation the
request attachment is cleared by controller reset/destruction, not by normal
RPC completion, and there is no completion-path reset here. The same pattern
appears in `GlobalMergeContext::publish_callbacks`, so successful
merge/direct-publish RPCs retain an extra serialized bloom-filter buffer until
a CTE reset or query teardown (potentially tens of MiB per filter during a long
probe). Please use completion-scoped ownership like
`RuntimeFilterRelayRpcClosure`, or otherwise release/reset each owner safely at
the end of its callback.
--
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]
