[Bug](runtime-filter) fix publish not inited rf when broadcast join m…

…eet wake up by downsteam #44823 (#44859)

pick from #44823

be/src/pipeline/exec/hashjoin_build_sink.cpp

@@ -142,19 +142,23 @@ Status HashJoinBuildSinkLocalState::close(RuntimeState* state, Status exec_statu
     }
 
     if (state->get_task()->wake_up_by_downstream()) {
-        RETURN_IF_ERROR(_runtime_filter_slots->send_filter_size(state, 0, _finish_dependency));
-        RETURN_IF_ERROR(_runtime_filter_slots->ignore_all_filters());
-    } else {
+        if (_should_build_hash_table) {
+            // partitial ignore rf to make global rf work
+            RETURN_IF_ERROR(_runtime_filter_slots->send_filter_size(state, 0, _finish_dependency));
+            RETURN_IF_ERROR(_runtime_filter_slots->ignore_all_filters());
+        } else {
+            // do not publish filter coz local rf not inited and useless
+            return Base::close(state, exec_status);
+        }
+    } else if (_should_build_hash_table) {
         auto* block = _shared_state->build_block.get();
         uint64_t hash_table_size = block ? block->rows() : 0;
         {
             SCOPED_TIMER(_runtime_filter_init_timer);
-            if (_should_build_hash_table) {
-                RETURN_IF_ERROR(_runtime_filter_slots->init_filters(state, hash_table_size));
-            }
+            RETURN_IF_ERROR(_runtime_filter_slots->init_filters(state, hash_table_size));
             RETURN_IF_ERROR(_runtime_filter_slots->ignore_filters(state));
         }
-        if (_should_build_hash_table && hash_table_size > 1) {
+        if (hash_table_size > 1) {
             SCOPED_TIMER(_runtime_filter_compute_timer);
             _runtime_filter_slots->insert(block);
         }