(doris) 01/01: fix publish not inited rf when broadcast join meet wake up by downsteam

[panxiaolei]

This is an automated email from the ASF dual-hosted git repository.

panxiaolei pushed a commit to branch cp_1202_2
in repository https://gitbox.apache.org/repos/asf/doris.git

commit c25a6d60cf564c38889380ce559462f41b7ac9d1
Author: BiteTheDDDDt <pxl...@qq.com>
AuthorDate: Mon Dec 2 16:24:51 2024 +0800

    fix publish not inited rf when broadcast join meet wake up by downsteam
---
 be/src/pipeline/exec/hashjoin_build_sink.cpp | 31 ++++++++++++++++------------
 1 file changed, 18 insertions(+), 13 deletions(-)

diff --git a/be/src/pipeline/exec/hashjoin_build_sink.cpp 
b/be/src/pipeline/exec/hashjoin_build_sink.cpp
index 4deed04fa2b..991127d83fe 100644
--- a/be/src/pipeline/exec/hashjoin_build_sink.cpp
+++ b/be/src/pipeline/exec/hashjoin_build_sink.cpp
@@ -127,24 +127,29 @@ Status HashJoinBuildSinkLocalState::close(RuntimeState* 
state, Status exec_statu
         return Base::close(state, exec_status);
     }
 
-    if (_should_build_hash_table) {
-        if (state->get_task()->wake_up_by_downstream()) {
+    if (state->get_task()->wake_up_by_downstream()) {
+        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 {
-            auto* block = _shared_state->build_block.get();
-            uint64_t hash_table_size = block ? block->rows() : 0;
-            {
-                SCOPED_TIMER(_runtime_filter_init_timer);
-                RETURN_IF_ERROR(_runtime_filter_slots->init_filters(state, 
hash_table_size));
-                RETURN_IF_ERROR(_runtime_filter_slots->ignore_filters(state));
-            }
-            if (hash_table_size > 1) {
-                SCOPED_TIMER(_runtime_filter_compute_timer);
-                _runtime_filter_slots->insert(block);
-            }
+            // 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);
+            RETURN_IF_ERROR(_runtime_filter_slots->init_filters(state, 
hash_table_size));
+            RETURN_IF_ERROR(_runtime_filter_slots->ignore_filters(state));
+        }
+        if (hash_table_size > 1) {
+            SCOPED_TIMER(_runtime_filter_compute_timer);
+            _runtime_filter_slots->insert(block);
         }
     }
+
     SCOPED_TIMER(_publish_runtime_filter_timer);
     RETURN_IF_ERROR(_runtime_filter_slots->publish(state, 
!_should_build_hash_table));
     return Base::close(state, exec_status);


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org