@@ -94,51 +94,43 @@ Status RuntimeFilterProducer::publish(RuntimeState* state, bool build_hash_table
9494 return Status::OK ();
9595}
9696
97- class SyncSizeClosure :
public AutoReleaseClosure <PSendFilterSizeRequest,
98- DummyBrpcCallback<PSendFilterSizeResponse>> {
99- std::shared_ptr<Dependency> _dependency;
100- // Should use weak ptr here, because when query context deconstructs, should also delete runtime filter
101- // context, it not the memory is not released. And rpc is in another thread, it will hold rf context
102- // after query context because the rpc is not returned.
103- std::weak_ptr<RuntimeFilterWrapper> _wrapper;
104- using Base =
105- AutoReleaseClosure<PSendFilterSizeRequest, DummyBrpcCallback<PSendFilterSizeResponse>>;
106- friend class RuntimeFilterProducer ;
107- ENABLE_FACTORY_CREATOR (SyncSizeClosure);
108-
109- void _process_if_rpc_failed () override {
110- Defer defer {[&]() {
111- Base::_process_if_rpc_failed ();
97+ // Callback for sync-size RPCs. Handles errors (disable wrapper + sub dependency) in call().
98+ class SyncSizeCallback :
public DummyBrpcCallback <PSendFilterSizeResponse> {
99+ ENABLE_FACTORY_CREATOR (SyncSizeCallback);
100+
101+ public:
102+ SyncSizeCallback (std::shared_ptr<Dependency> dependency,
103+ std::shared_ptr<RuntimeFilterWrapper> wrapper)
104+ : _dependency(std::move(dependency)), _wrapper(wrapper) {}
105+
106+ void call () override {
107+ // On error: disable the wrapper and sub the dependency here, because set_synced_size()
108+ // will never be called (the merge node won't respond with a sync).
109+ // On success: do NOT sub here. The merge node will respond with sync_filter_size,
110+ // which calls set_synced_size() -> _dependency->sub().
111+ if (cntl_->Failed ()) {
112+ if (auto w = _wrapper.lock ()) {
113+ w->set_state (RuntimeFilterWrapper::State::DISABLED, cntl_->ErrorText ());
114+ }
112115 ((CountedFinishDependency*)_dependency.get ())->sub ();
113- }};
114- auto wrapper = _wrapper.lock ();
115- if (!wrapper) {
116116 return ;
117117 }
118118
119- wrapper->set_state (RuntimeFilterWrapper::State::DISABLED, cntl_->ErrorText ());
120- }
121-
122- void _process_if_meet_error_status (const Status& status) override {
123- Defer defer {[&]() {
124- Base::_process_if_meet_error_status (status);
119+ Status status = Status::create (response_->status ());
120+ if (!status.ok ()) {
121+ if (auto w = _wrapper.lock ()) {
122+ w->set_state (RuntimeFilterWrapper::State::DISABLED, status.to_string ());
123+ }
125124 ((CountedFinishDependency*)_dependency.get ())->sub ();
126- }};
127- auto wrapper = _wrapper.lock ();
128- if (!wrapper) {
129- return ;
130125 }
131-
132- wrapper->set_state (RuntimeFilterWrapper::State::DISABLED, status.to_string ());
133126 }
134127
135- public:
136- SyncSizeClosure (std::shared_ptr<PSendFilterSizeRequest> req,
137- std::shared_ptr<DummyBrpcCallback<PSendFilterSizeResponse>> callback,
138- std::shared_ptr<Dependency> dependency,
139- std::shared_ptr<RuntimeFilterWrapper> wrapper,
140- std::weak_ptr<QueryContext> context)
141- : Base(req, callback, context), _dependency(std::move(dependency)), _wrapper(wrapper) {}
128+ private:
129+ std::shared_ptr<Dependency> _dependency;
130+ // Should use weak ptr here, because when query context deconstructs, should also delete runtime filter
131+ // context, it not the memory is not released. And rpc is in another thread, it will hold rf context
132+ // after query context because the rpc is not returned.
133+ std::weak_ptr<RuntimeFilterWrapper> _wrapper;
142134};
143135
144136void RuntimeFilterProducer::latch_dependency (
@@ -199,13 +191,15 @@ Status RuntimeFilterProducer::send_size(RuntimeState* state, uint64_t local_filt
199191 }
200192
201193 auto request = std::make_shared<PSendFilterSizeRequest>();
202- auto callback = DummyBrpcCallback<PSendFilterSizeResponse>::
create_shared ();
194+ auto callback = SyncSizeCallback::create_shared (_dependency, _wrapper);
195+ // Store callback in the producer to keep it alive until the RPC completes.
196+ // AutoReleaseClosure holds callbacks via weak_ptr, so without this the callback
197+ // would be destroyed when this function returns and error-path sub() would never fire.
198+ _sync_size_callback = callback;
203199 // RuntimeFilter maybe deconstructed before the rpc finished, so that could not use
204200 // a raw pointer in closure. Has to use the context's shared ptr.
205- auto closure = SyncSizeClosure::create_unique (request, callback, _dependency, _wrapper,
206- state->query_options ().ignore_runtime_filter_error
207- ? std::weak_ptr<QueryContext> {}
208- : state->get_query_ctx_weak ());
201+ auto closure = AutoReleaseClosure<PSendFilterSizeRequest, SyncSizeCallback>::
create_unique (
202+ request, callback);
209203 auto * pquery_id = request->mutable_query_id ();
210204 pquery_id->set_hi (state->get_query_ctx ()->query_id ().hi );
211205 pquery_id->set_lo (state->get_query_ctx ()->query_id ().lo );
0 commit comments