diff --git a/be/src/exec/runtime_filter/runtime_filter.h b/be/src/exec/runtime_filter/runtime_filter.h index c1e80dd2b88c9e..0a69ca92c9acbb 100644 --- a/be/src/exec/runtime_filter/runtime_filter.h +++ b/be/src/exec/runtime_filter/runtime_filter.h @@ -22,6 +22,7 @@ #include #include "common/exception.h" +#include "common/logging.h" #include "common/status.h" #include "exec/runtime_filter/runtime_filter_definitions.h" #include "exec/runtime_filter/runtime_filter_wrapper.h" @@ -67,10 +68,34 @@ class RuntimeFilter { request->set_filter_id(_wrapper->filter_id()); auto state = _wrapper->get_state(); - if (state != RuntimeFilterWrapper::State::READY) { + // Align with the legacy (branch-3.1) semantics: only a *really* disabled filter + // (reach max_in_num / join spill / rpc error, i.e. State::DISABLED) should be + // published as `disabled` to consumers. A filter that is merely NOT ready yet + // (State::UNINITED) must NOT be turned into a disabled filter, otherwise the + // consumer permanently gives up this filter ("get disabled from remote") and a + // large table degrades to a full scan. + // + // Callers are expected to skip publishing an UNINITED filter (see + // is_wrapper_uninited() checks in the producer / merge controller) and let the + // consumer wait until timeout instead. Reaching here with UNINITED therefore means + // an unexpected code path. We surface it via DCHECK (fail fast in debug builds) but + // in release builds we degrade *safely*: neither mark the filter as disabled nor + // return an error. Serializing "nothing" leaves the request without filter content + // and without the disabled flag, so the consumer keeps waiting until timeout and at + // worst loses the filter effect (falling back to a full scan) instead of failing the + // whole query. + if (state == RuntimeFilterWrapper::State::DISABLED) { request->set_disabled(true); return Status::OK(); } + if (state == RuntimeFilterWrapper::State::UNINITED) { + DCHECK(false) << "Try to serialize an uninitialized(not ready) runtime filter, " + << "filter_id=" << _wrapper->filter_id(); + LOG(WARNING) << "Try to serialize an uninitialized(not ready) runtime filter, " + "skip publishing to avoid disabling it on consumers, filter_id=" + << _wrapper->filter_id(); + return Status::OK(); + } request->set_contain_null(_wrapper->contain_null()); diff --git a/be/src/exec/runtime_filter/runtime_filter_merger.h b/be/src/exec/runtime_filter/runtime_filter_merger.h index 4e1d19dabc3a7b..6a82925ae362d4 100644 --- a/be/src/exec/runtime_filter/runtime_filter_merger.h +++ b/be/src/exec/runtime_filter/runtime_filter_merger.h @@ -115,6 +115,16 @@ class RuntimeFilterMerger : public RuntimeFilter { bool ready() const { return _rf_state == State::READY; } + // Returns true when the merged wrapper is still UNINITED, i.e. the merger has + // collected all products by count but there is no really produced filter content + // (nor a real DISABLED). In this case we must NOT publish it as a disabled filter; + // instead skip publishing so that consumers fall back to waiting until timeout, + // which aligns with the legacy (branch-3.1) runtime filter semantics. + bool is_wrapper_uninited() { + std::unique_lock l(_rmtx); + return _wrapper->get_state() == RuntimeFilterWrapper::State::UNINITED; + } + void set_wrapper_state_and_ready_to_apply(RuntimeFilterWrapper::State state, std::string reason = "") { std::unique_lock l(_rmtx); diff --git a/be/src/exec/runtime_filter/runtime_filter_mgr.cpp b/be/src/exec/runtime_filter/runtime_filter_mgr.cpp index f69efa855a95b4..135d6b4c42ee9a 100644 --- a/be/src/exec/runtime_filter/runtime_filter_mgr.cpp +++ b/be/src/exec/runtime_filter/runtime_filter_mgr.cpp @@ -631,6 +631,17 @@ Status RuntimeFilterMergeControllerEntity::_send_rf_to_target( return Status::InternalError("Runtime filter has been sent", cnt_val.merger->debug_string()); } + + // Align with the legacy (branch-3.1) semantics: if all products have arrived (by count) + // but the merged filter content is still not ready (UNINITED, not a real DISABLED), + // do NOT broadcast it. Publishing here would serialize it as `disabled` and every + // consumer would permanently give up the filter ("get disabled from remote"), causing + // e.g. a large probe-side table to degrade into a full table scan. Skip publishing and + // let consumers wait until their runtime_filter_wait_time_ms timeout instead. + if (cnt_val.merger->is_wrapper_uninited()) { + cnt_val.done = true; + return Status::OK(); + } cnt_val.done = true; butil::IOBuf request_attachment; diff --git a/be/src/exec/runtime_filter/runtime_filter_producer.cpp b/be/src/exec/runtime_filter/runtime_filter_producer.cpp index b3684a774ab650..eec80d43379b8e 100644 --- a/be/src/exec/runtime_filter/runtime_filter_producer.cpp +++ b/be/src/exec/runtime_filter/runtime_filter_producer.cpp @@ -64,6 +64,14 @@ Status RuntimeFilterProducer::publish(RuntimeState* state, bool build_hash_table bool ready = false; RETURN_IF_ERROR(context->merger->merge_from(this, &ready)); if (ready) { + // Align with the legacy (branch-3.1) semantics: only send the merged filter when + // there is real content (READY) or it is really disabled (DISABLED). If the merger + // has collected all products by count but the content is still UNINITED, skip + // sending so that consumers fall back to waiting until timeout, instead of being + // told "disabled from remote" and degrading a large table to a full scan. + if (context->merger->is_wrapper_uninited()) { + return Status::OK(); + } if (_has_remote_target) { RETURN_IF_ERROR(_send_to_remote_targets(state, context->merger.get())); } else {