placeholder for real size runtime bloom filter

src/sql/CMakeLists.txt

@@ -781,6 +781,7 @@ ob_set_subtarget(ob_sql engine_join
   engine/join/ob_join_op.cpp
   engine/join/ob_merge_join_op.cpp
   engine/join/ob_nested_loop_join_op.cpp
+  engine/join/ob_join_filter_material_control_info.cpp
 )
 
 ob_set_subtarget(ob_sql engine_pdml

src/sql/engine/join/hash_join/ob_hash_join_vec_op.cpp

@@ -124,7 +124,8 @@ OB_SERIALIZE_MEMBER((ObHashJoinVecSpec, ObJoinVecSpec),
                     is_naaj_,
                     is_sna_,
                     is_shared_ht_,
-                    is_ns_equal_cond_);
+                    is_ns_equal_cond_,
+                    jf_material_control_info_);
 
 ObHashJoinVecOp::ObHashJoinVecOp(ObExecContext &ctx_, const ObOpSpec &spec, ObOpInput *input)
   : ObJoinVecOp(ctx_, spec, input),

src/sql/engine/join/hash_join/ob_hash_join_vec_op.h

@@ -22,6 +22,7 @@
 #include "sql/engine/ob_sql_mem_mgr_processor.h"
 #include "sql/engine/aggregate/ob_exec_hash_struct.h"
 #include "sql/engine/aggregate/ob_adaptive_bypass_ctrl.h"
+#include "sql/engine/join/ob_join_filter_material_control_info.h"
 
 namespace oceanbase
 {
@@ -256,6 +257,7 @@ public:
   bool is_shared_ht_;
   // record which equal cond is null safe equal
   common::ObFixedArray<bool, common::ObIAllocator> is_ns_equal_cond_;
+  ObJoinFilterMaterialControlInfo jf_material_control_info_;
 };
 
 // hash join has no expression result overwrite problem:

src/sql/engine/join/ob_join_filter_material_control_info.cpp

@@ -0,0 +1,25 @@
+/** * Copyright (c) 2024 OceanBase
+ * OceanBase CE is licensed under Mulan PubL v2.
+ * You can use this software according to the terms and conditions of the Mulan PubL v2.
+ * You may obtain a copy of Mulan PubL v2 at:
+ *          http://license.coscl.org.cn/MulanPubL-2.0
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PubL v2 for more details.
+ */
+
+#include "sql/engine/join/ob_join_filter_material_control_info.h"
+#include "lib/utility/ob_unify_serialize.h"
+
+using namespace oceanbase::lib;
+using namespace oceanbase;
+
+namespace oceanbase
+{
+namespace sql
+{
+OB_SERIALIZE_MEMBER(ObJoinFilterMaterialControlInfo, enable_material_, hash_id_, is_controller_,
+                    join_filter_count_, extra_hash_count_, each_sqc_has_full_data_);
+}
+}

src/sql/engine/join/ob_join_filter_material_control_info.h

@@ -0,0 +1,43 @@
+/** * Copyright (c) 2024 OceanBase
+ * OceanBase CE is licensed under Mulan PubL v2.
+ * You can use this software according to the terms and conditions of the Mulan PubL v2.
+ * You may obtain a copy of Mulan PubL v2 at:
+ *          http://license.coscl.org.cn/MulanPubL-2.0
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PubL v2 for more details.
+ */
+
+#pragma once
+
+#include "lib/container/ob_fixed_array.h"
+#include <utility>
+
+namespace oceanbase
+{
+namespace sql
+{
+
+struct ObJoinFilterMaterialControlInfo
+{
+  OB_UNIS_VERSION_V(1);
+public:
+  TO_STRING_KV(K(enable_material_), K(hash_id_), K(is_controller_), K(join_filter_count_),
+               K(extra_hash_count_), K(each_sqc_has_full_data_));
+
+public:
+  // these variables for all join filter
+  bool enable_material_{false};
+  int16_t hash_id_{-1}; // mark the hash value position in compact row
+
+  // these variables for the controller and hash join
+  bool is_controller_{false}; // only the top join filter become the controller
+  uint16_t join_filter_count_{0}; // total join filter count in the left side of a hash join
+  uint16_t extra_hash_count_{0}; // hash value count(one for hash join, several for join filter)
+  bool each_sqc_has_full_data_{false}; // mark whether each sqc has complete data
+};
+
+
+}
+}

src/sql/engine/join/ob_join_filter_op.cpp

@@ -40,7 +40,8 @@ OB_SERIALIZE_MEMBER(ObJoinFilterShareInfo,
     ch_provider_ptr_,
     release_ref_ptr_,
     filter_ptr_,
-    shared_msgs_);
+    shared_msgs_,
+    ser_shared_jf_constructor_);
 
 OB_SERIALIZE_MEMBER(ObJoinFilterRuntimeConfig,
     bloom_filter_ratio_,
@@ -74,7 +75,11 @@ OB_SERIALIZE_MEMBER((ObJoinFilterSpec, ObOpSpec),
                     rf_probe_cmp_infos_,
                     px_query_range_info_,
                     bloom_filter_ratio_,
-                    send_bloom_filter_size_);
+                    send_bloom_filter_size_,
+                    jf_material_control_info_,
+                    join_type_,
+                    full_hash_join_keys_,
+                    hash_join_is_ns_equal_cond_);
 
 OB_SERIALIZE_MEMBER(ObJoinFilterOpInput,
     share_info_,

src/sql/engine/join/ob_join_filter_op.h

@@ -27,6 +27,7 @@
 #include "sql/engine/px/p2p_datahub/ob_runtime_filter_vec_msg.h"
 #include "share/detect/ob_detectable_id.h"
 #include "sql/engine/px/p2p_datahub/ob_runtime_filter_query_range.h"
+#include "sql/engine/join/ob_join_filter_material_control_info.h"
 
 
 namespace oceanbase
@@ -36,6 +37,19 @@ namespace sql
 
 class ObPxSQCProxy;
 
+class SharedJoinFilterConstructor
+{
+public:
+  inline bool snatch_the_leader() { return !ATOMIC_CAS(&leader_snatched_, false, true); }
+  int init();
+  int wait_bloom_filter_init(ObOperator *join_filter_op, ObRFBloomFilterMsg *bf_msg);
+  int notify_bloom_filter_inited();
+private:
+  static constexpr uint64_t COND_WAIT_TIME_USEC = 10; // 10 us
+  ObThreadCond cond_;
+  bool leader_snatched_{false};
+  bool is_shared_bloom_filter_inited_{false};
+} CACHE_ALIGNED;
 
 struct ObJoinFilterShareInfo
 {
@@ -48,6 +62,10 @@ struct ObJoinFilterShareInfo
   uint64_t release_ref_ptr_; // 释放内存引用计数, 初始值为worker个数.
   uint64_t filter_ptr_;   //此指针将作为PX JOIN FILTER CREATE算子共享内存.
   uint64_t shared_msgs_;  //sqc-shared dh msgs
+  union {
+    SharedJoinFilterConstructor *shared_jf_constructor_;
+    uint64_t ser_shared_jf_constructor_;
+  };
   OB_UNIS_VERSION_V(1);
 public:
   TO_STRING_KV(KP(unfinished_count_ptr_), KP(ch_provider_ptr_), KP(release_ref_ptr_), KP(filter_ptr_), K(shared_msgs_));
@@ -214,6 +232,10 @@ public:
   ObPxQueryRangeInfo px_query_range_info_;
   int64_t bloom_filter_ratio_;
   int64_t send_bloom_filter_size_; // how many KB a piece bloom filter has
+  ObJoinFilterMaterialControlInfo jf_material_control_info_;
+  ObJoinType join_type_ {UNKNOWN_JOIN};
+  ExprFixedArray full_hash_join_keys_;
+  common::ObFixedArray<bool, common::ObIAllocator> hash_join_is_ns_equal_cond_;
 };
 
 class ObJoinFilterOp : public ObOperator