@@ -98,51 +98,55 @@ Status RuntimeFilterProducer::publish(RuntimeState* state, bool build_hash_table
9898 return Status::OK ();
9999}
100100
101- class SyncSizeClosure :
public AutoReleaseClosure <PSendFilterSizeRequest,
102- DummyBrpcCallback<PSendFilterSizeResponse>> {
103- std::shared_ptr<Dependency> _dependency;
104- // Should use weak ptr here, because when query context deconstructs, should also delete runtime filter
105- // context, it not the memory is not released. And rpc is in another thread, it will hold rf context
106- // after query context because the rpc is not returned.
107- std::weak_ptr<RuntimeFilterWrapper> _wrapper;
108- using Base =
109- AutoReleaseClosure<PSendFilterSizeRequest, DummyBrpcCallback<PSendFilterSizeResponse>>;
110- friend class RuntimeFilterProducer ;
111- ENABLE_FACTORY_CREATOR (SyncSizeClosure);
101+ // Callback for sync-size RPCs. Handles errors (disable wrapper + sub dependency) in call().
102+ class SyncSizeCallback :
public DummyBrpcCallback <PSendFilterSizeResponse> {
103+ ENABLE_FACTORY_CREATOR (SyncSizeCallback);
104+
105+ public:
106+ SyncSizeCallback (std::shared_ptr<Dependency> dependency,
107+ std::shared_ptr<RuntimeFilterWrapper> wrapper,
108+ std::weak_ptr<QueryContext> context)
109+ : _dependency(std::move(dependency)), _wrapper(wrapper), _context(std::move(context)) {}
112110
113- void _process_if_rpc_failed () override {
114- Defer defer {[&]() {
115- Base::_process_if_rpc_failed ();
111+ void call () override {
112+ // On error: disable the wrapper and sub the dependency here, because set_synced_size()
113+ // will never be called (the merge node won't respond with a sync).
114+ // On success: do NOT sub here. The merge node will respond with sync_filter_size,
115+ // which calls set_synced_size() -> _dependency->sub().
116+ if (cntl_->Failed ()) {
117+ if (auto w = _wrapper.lock ()) {
118+ w->set_state (RuntimeFilterWrapper::State::DISABLED, cntl_->ErrorText ());
119+ }
120+ if (auto ctx = _context.lock ()) {
121+ if (!ctx->ignore_runtime_filter_error ()) {
122+ ctx->cancel (Status::NetworkError (" RPC meet failed: {}" ErrorText ()));
123+ }
124+ }
116125 ((CountedFinishDependency*)_dependency.get ())->sub ();
117- }};
118- auto wrapper = _wrapper.lock ();
119- if (!wrapper) {
120126 return ;
121127 }
122128
123- wrapper->set_state (RuntimeFilterWrapper::State::DISABLED, cntl_->ErrorText ());
124- }
125-
126- void _process_if_meet_error_status (const Status& status) override {
127- Defer defer {[&]() {
128- Base::_process_if_meet_error_status (status);
129+ Status status = Status::create (response_->status ());
130+ if (!status.ok ()) {
131+ if (auto w = _wrapper.lock ()) {
132+ w->set_state (RuntimeFilterWrapper::State::DISABLED, status.to_string ());
133+ }
134+ if (auto ctx = _context.lock ()) {
135+ if (!ctx->ignore_runtime_filter_error ()) {
136+ ctx->cancel (Status::NetworkError (" RPC meet failed: {}" to_string ()));
137+ }
138+ }
129139 ((CountedFinishDependency*)_dependency.get ())->sub ();
130- }};
131- auto wrapper = _wrapper.lock ();
132- if (!wrapper) {
133- return ;
134140 }
135-
136- wrapper->set_state (RuntimeFilterWrapper::State::DISABLED, status.to_string ());
137141 }
138142
139- public :
140- SyncSizeClosure ( std::shared_ptr<PSendFilterSizeRequest> req,
141- std::shared_ptr<DummyBrpcCallback<PSendFilterSizeResponse>> callback,
142- std::shared_ptr<Dependency> dependency,
143- std::shared_ptr<RuntimeFilterWrapper> wrapper,
144- std::weak_ptr<QueryContext> context)
145- : Base(req, callback, context), _dependency( std::move(dependency)), _wrapper(wrapper) {}
143+ private :
144+ std::shared_ptr<Dependency> _dependency;
145+ // Should use weak ptr here, because when query context deconstructs, should also delete runtime filter
146+ // context, it not the memory is not released. And rpc is in another thread, it will hold rf context
147+ // after query context because the rpc is not returned.
148+ std::weak_ptr<RuntimeFilterWrapper> _wrapper;
149+ std::weak_ptr<QueryContext> _context;
146150};
147151
148152void RuntimeFilterProducer::latch_dependency (
@@ -208,14 +212,16 @@ Status RuntimeFilterProducer::send_size(RuntimeState* state, uint64_t local_filt
208212
209213 auto request = std::make_shared<PSendFilterSizeRequest>();
210214 request->set_stage (_stage);
211-
212- auto callback = DummyBrpcCallback<PSendFilterSizeResponse>::
create_shared ();
215+ auto callback = SyncSizeCallback::create_shared (_dependency, _wrapper,
216+ state->get_query_ctx ()->weak_from_this ());
217+ // Store callback in the producer to keep it alive until the RPC completes.
218+ // AutoReleaseClosure holds callbacks via weak_ptr, so without this the callback
219+ // would be destroyed when this function returns and error-path sub() would never fire.
220+ _sync_size_callback = callback;
213221 // RuntimeFilter maybe deconstructed before the rpc finished, so that could not use
214222 // a raw pointer in closure. Has to use the context's shared ptr.
215- auto closure = SyncSizeClosure::create_unique (request, callback, _dependency, _wrapper,
216- state->query_options ().ignore_runtime_filter_error
217- ? std::weak_ptr<QueryContext> {}
218- : state->get_query_ctx_weak ());
223+ auto closure = AutoReleaseClosure<PSendFilterSizeRequest, SyncSizeCallback>::
create_unique (
224+ request, callback);
219225 auto * pquery_id = request->mutable_query_id ();
220226 pquery_id->set_hi (state->get_query_ctx ()->query_id ().hi );
221227 pquery_id->set_lo (state->get_query_ctx ()->query_id ().lo );
0 commit comments