[fix](runtime filter) do not publish an uninitialized (UNINITED) filter as disabled

be/src/exec/runtime_filter/runtime_filter.h

@@ -22,6 +22,7 @@
 #include <vector>
 
 #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());
 

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<std::recursive_mutex> 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<std::recursive_mutex> l(_rmtx);

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;

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 {