[Bug](runtime-filter) fix missing rf when target not exist on backend where the producer is…

be/src/exprs/runtime_filter.cpp

@@ -323,25 +323,31 @@ Status IRuntimeFilter::publish(RuntimeState* state, bool publish_local) {
         return Status::OK();
     };
     auto do_merge = [&]() {
-        if (!_state->global_runtime_filter_mgr()->get_consume_filters(_filter_id).empty()) {
-            LocalMergeFilters* local_merge_filters = nullptr;
-            RETURN_IF_ERROR(_state->global_runtime_filter_mgr()->get_local_merge_producer_filters(
-                    _filter_id, &local_merge_filters));
-            local_merge_filters->merge_watcher.start();
-            std::lock_guard l(*local_merge_filters->lock);
-            RETURN_IF_ERROR(local_merge_filters->filters[0]->merge_from(_wrapper.get()));
-            local_merge_filters->merge_time--;
-            local_merge_filters->merge_watcher.stop();
-            if (local_merge_filters->merge_time == 0) {
-                if (_has_local_target) {
-                    RETURN_IF_ERROR(send_to_local_targets(
-                            local_merge_filters->filters[0]->_wrapper, true,
-                            local_merge_filters->merge_watcher.elapsed_time()));
-                } else {
-                    RETURN_IF_ERROR(send_to_remote_targets(
-                            local_merge_filters->filters[0].get(),
-                            local_merge_filters->merge_watcher.elapsed_time()));
-                }
+        // two case we need do local merge:
+        // 1. has remote target
+        // 2. has local target and has global consumer (means target scan has local shuffle)
+        if (_has_local_target &&
+            _state->global_runtime_filter_mgr()->get_consume_filters(_filter_id).empty()) {
+            // when global consumer not exist, send_to_local_targets will do nothing, so merge rf is useless
+            return Status::OK();
+        }
+        LocalMergeFilters* local_merge_filters = nullptr;
+        RETURN_IF_ERROR(_state->global_runtime_filter_mgr()->get_local_merge_producer_filters(
+                _filter_id, &local_merge_filters));
+        local_merge_filters->merge_watcher.start();
+        std::lock_guard l(*local_merge_filters->lock);
+        RETURN_IF_ERROR(local_merge_filters->filters[0]->merge_from(_wrapper.get()));
+        local_merge_filters->merge_time--;
+        local_merge_filters->merge_watcher.stop();
+        if (local_merge_filters->merge_time == 0) {
+            if (_has_local_target) {
+                RETURN_IF_ERROR(
+                        send_to_local_targets(local_merge_filters->filters[0]->_wrapper, true,
+                                              local_merge_filters->merge_watcher.elapsed_time()));
+            } else {
+                RETURN_IF_ERROR(
+                        send_to_remote_targets(local_merge_filters->filters[0].get(),
+                                               local_merge_filters->merge_watcher.elapsed_time()));
             }
         }
         return Status::OK();
@@ -417,7 +423,11 @@ class SyncSizeClosure : public AutoReleaseClosure<PSendFilterSizeRequest,
 Status IRuntimeFilter::send_filter_size(RuntimeState* state, uint64_t local_filter_size) {
     DCHECK(is_producer());
 
-    if (!_state->global_runtime_filter_mgr()->get_consume_filters(_filter_id).empty()) {
+    // two case we need do local merge:
+    // 1. has remote target
+    // 2. has local target and has global consumer (means target scan has local shuffle)
+    if (_has_remote_target ||
+        !_state->global_runtime_filter_mgr()->get_consume_filters(_filter_id).empty()) {
         LocalMergeFilters* local_merge_filters = nullptr;
         RETURN_IF_ERROR(_state->global_runtime_filter_mgr()->get_local_merge_producer_filters(
                 _filter_id, &local_merge_filters));

be/src/runtime/runtime_filter_mgr.cpp

@@ -170,16 +170,6 @@ Status RuntimeFilterMgr::get_local_merge_producer_filters(
     return Status::OK();
 }
 
-doris::LocalMergeFilters* RuntimeFilterMgr::get_local_merge_producer_filters(int filter_id) {
-    DCHECK(_is_global);
-    std::lock_guard<std::mutex> l(_lock);
-    auto iter = _local_merge_producer_map.find(filter_id);
-    if (iter == _local_merge_producer_map.end()) {
-        return nullptr;
-    }
-    return &iter->second;
-}
-
 Status RuntimeFilterMgr::register_producer_filter(
         const TRuntimeFilterDesc& desc, const TQueryOptions& options,
         std::shared_ptr<IRuntimeFilter>* producer_filter) {

be/src/runtime/runtime_filter_mgr.h

@@ -107,7 +107,6 @@ class RuntimeFilterMgr {
                                                 std::shared_ptr<IRuntimeFilter> producer_filter);
 
     Status get_local_merge_producer_filters(int filter_id, LocalMergeFilters** local_merge_filters);
-    LocalMergeFilters* get_local_merge_producer_filters(int filter_id);
 
     Status register_producer_filter(const TRuntimeFilterDesc& desc, const TQueryOptions& options,
                                     std::shared_ptr<IRuntimeFilter>* producer_filter);

regression-test/data/query_p0/join/test_low_bucket/test_low_bucket.out

@@ -0,0 +1,151 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+
+-- !test --
+480
+