From fa60c60b420868b16d4d7c2d5f2c1ba297c945c6 Mon Sep 17 00:00:00 2001
From: Jerry Hu <hushenggang@selectdb.com>
Date: Mon, 13 Jan 2025 14:21:43 +0800
Subject: [PATCH] [fix](pipelien) should not finalize probe when wake up early
 in SetProbeSinkOperatorX (#46706) (#46832)

```
*** Query id: 80819fcc223e4a45-b46155de6e0c4eee ***
*** is nereids: 1 ***
*** tablet id: 0 ***
*** Aborted at 1736352810 (unix time) try "date -d @1736352810" if you are using GNU date ***
*** Current BE git commitID: 08683cbaf5 ***
*** SIGSEGV address not mapped to object (@0x38) received by PID 8736 (TID 11549 OR 0x7f8dd0922640) from PID 56; stack trace: ***
 0# doris::signal::(anonymous namespace)::FailureSignalHandler(int, siginfo_t*, void*) at /root/doris_branch-3.0/doris/be/src/common/signal_handler.h:421
 1# PosixSignals::chained_handler(int, siginfo*, void*) [clone .part.0] in /usr/lib/jvm/java-17-openjdk-amd64/lib/server/libjvm.so
 2# JVM_handle_linux_signal in /usr/lib/jvm/java-17-openjdk-amd64/lib/server/libjvm.so
 3# 0x00007F92019CA520 in /lib/x86_64-linux-gnu/libc.so.6
 4# auto doris::pipeline::SetProbeSinkOperatorX::_refresh_hash_table(doris::pipeline::SetProbeSinkLocalState&)::{lambda(auto:1&&)#1}::operator(), HashTableNoState>, DefaultHash, HashTableGrower<10ul>, Allocator > >&>(doris::vectorized::MethodSerialized, HashTableNoState>, DefaultHash, HashTableGrower<10ul>, Allocator > >&) const at /root/doris_branch-3.0/doris/be/src/pipeline/exec/set_probe_sink_operator.cpp:213
 5# doris::pipeline::SetProbeSinkOperatorX::_finalize_probe(doris::pipeline::SetProbeSinkLocalState&) at /root/doris_branch-3.0/doris/be/src/pipeline/exec/set_probe_sink_operator.cpp:184
 6# doris::pipeline::SetProbeSinkOperatorX::sink(doris::RuntimeState*, doris::vectorized::Block*, bool) at /root/doris_branch-3.0/doris/be/src/pipeline/exec/set_probe_sink_operator.cpp:98
 7# doris::pipeline::PipelineTask::execute(bool*) at /root/doris_branch-3.0/doris/be/src/pipeline/pipeline_task.cpp:387
 8# doris::pipeline::TaskScheduler::_do_work(unsigned long) at /root/doris_branch-3.0/doris/be/src/pipeline/task_scheduler.cpp:138
 9# doris::ThreadPool::dispatch_thread() in /mnt/ssd01/doris-branch40preview/NEREIDS_ASAN/be/lib/doris_be
10# doris::Thread::supervise_thread(void*) at /root/doris_branch-3.0/doris/be/src/util/thread.cpp:499
11# start_thread at ./nptl/pthread_create.c:442
12# 0x00007F9201AAE850 at ../sysdeps/unix/sysv/linux/x86_64/clone3.S:83
```
---
 be/src/pipeline/exec/set_probe_sink_operator.cpp | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/be/src/pipeline/exec/set_probe_sink_operator.cpp b/be/src/pipeline/exec/set_probe_sink_operator.cpp
index 7688bce00ef28c..857df17227dc2c 100644
--- a/be/src/pipeline/exec/set_probe_sink_operator.cpp
+++ b/be/src/pipeline/exec/set_probe_sink_operator.cpp
@@ -22,6 +22,7 @@
 #include <memory>
 
 #include "pipeline/exec/operator.h"
+#include "pipeline/pipeline_x/pipeline_x_task.h"
 #include "vec/common/hash_table/hash_table_set_probe.h"
 #include "vec/exec/vset_operation_node.h"
 
@@ -130,7 +131,7 @@ Status SetProbeSinkOperatorX<is_intersect>::sink(RuntimeState* state, vectorized
                 *local_state._shared_state->hash_table_variants));
     }
 
-    if (eos) {
+    if (eos && !state->get_task()->wake_up_early()) {
         _finalize_probe(local_state);
     }
     return Status::OK();