From 81270342f7312c5daeae65e29922cdfa47573ddc Mon Sep 17 00:00:00 2001
From: obdev <obdev@oceanbase.com>
Date: Sat, 10 Feb 2024 06:53:44 +0000
Subject: [PATCH] fix bug: runtime filter core because of data racing

---
 src/sql/engine/px/p2p_datahub/ob_p2p_dh_mgr.cpp | 1 +
 src/sql/engine/px/p2p_datahub/ob_p2p_dh_msg.cpp | 7 ++++---
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/src/sql/engine/px/p2p_datahub/ob_p2p_dh_mgr.cpp b/src/sql/engine/px/p2p_datahub/ob_p2p_dh_mgr.cpp
index 5ccfac9b2a..6719ffad28 100644
--- a/src/sql/engine/px/p2p_datahub/ob_p2p_dh_mgr.cpp
+++ b/src/sql/engine/px/p2p_datahub/ob_p2p_dh_mgr.cpp
@@ -418,6 +418,7 @@ int ObP2PDatahubManager::P2PMsgSetCall::operator() (const common::hash::HashMapP
         K(dh_key_), K(dh_msg_.get_dm_cb_node_seq_id()));
     if (OB_FAIL(dh_msg_.regenerate())) {
       LOG_WARN("failed to do regen_call", K(dh_key_));
+    } else if (FALSE_IT(dh_msg_.check_finish_receive())) {
     }
   }
   if (OB_FAIL(ret)) {
diff --git a/src/sql/engine/px/p2p_datahub/ob_p2p_dh_msg.cpp b/src/sql/engine/px/p2p_datahub/ob_p2p_dh_msg.cpp
index 1f60548dc3..8d8baa9af5 100644
--- a/src/sql/engine/px/p2p_datahub/ob_p2p_dh_msg.cpp
+++ b/src/sql/engine/px/p2p_datahub/ob_p2p_dh_msg.cpp
@@ -126,6 +126,9 @@ int ObP2PDatahubMsgBase::process_msg_internal(bool &need_free)
   ObP2PDatahubMsgGuard guard(this);
 
   bool need_merge = true;
+  // to avoid data racing in the process of check_finish_receive, protect it in hashmap lock
+  // if set succ, check_finish_receive() in P2PMsgSetCall
+  // if set failed(with OB_HASH_EXIST, need merge), check_finish_receive() in P2PMsgMergeCall
   if (OB_FAIL(map.set_refactored(dh_key, this, 0/*flag*/, 0/*broadcast*/, 0/*overwrite_key*/, &set_call))) {
     if (OB_HASH_EXIST == ret) {
       ret = OB_SUCCESS;
@@ -145,9 +148,7 @@ int ObP2PDatahubMsgBase::process_msg_internal(bool &need_free)
       }
     }
   }
-  if (OB_SUCC(ret) && !need_merge) {
-    (void)check_finish_receive();
-  }
+
   if (need_free) {
     // msg not in map, dec ref count
     guard.dec_msg_ref_count();