[fix](join) core caused by null-safe-equal join

be/src/pipeline/exec/hashjoin_build_sink.cpp

@@ -361,8 +361,22 @@ void HashJoinBuildSinkLocalState::_hash_table_init(RuntimeState* state) {
                     }
                     return;
                 }
+
+                std::vector<vectorized::DataTypePtr> data_types;
+                for (size_t i = 0; i != _build_expr_ctxs.size(); ++i) {
+                    auto& ctx = _build_expr_ctxs[i];
+                    auto data_type = ctx->root()->data_type();
+
+                    /// For 'null safe equal' join,
+                    /// the build key column maybe be converted to nullable from non-nullable.
+                    if (p._should_convert_to_nullable[i]) {
+                        data_type = vectorized::make_nullable(data_type);
+                    }
+                    data_types.emplace_back(std::move(data_type));
+                }
+
                 if (!try_get_hash_map_context_fixed<JoinHashMap, HashCRC32>(
-                            *_shared_state->hash_table_variants, _build_expr_ctxs)) {
+                            *_shared_state->hash_table_variants, data_types)) {
                     _shared_state->hash_table_variants
                             ->emplace<vectorized::SerializedHashTableContext>();
                 }

be/src/vec/common/hash_table/hash_map_context_creator.h

@@ -55,14 +55,23 @@ void get_hash_map_context_fixed(Variant& variant, size_t size, bool has_nullable
 template <template <typename... Args> typename HashMap, template <typename> typename Hash,
           typename... Mapped, typename Variant>
 bool try_get_hash_map_context_fixed(Variant& variant, const VExprContextSPtrs& expr_ctxs) {
+    std::vector<DataTypePtr> data_types;
+    for (const auto& ctx : expr_ctxs) {
+        data_types.emplace_back(ctx->root()->data_type());
+    }
+    return try_get_hash_map_context_fixed<HashMap, Hash, Mapped...>(variant, data_types);
+}
+
+template <template <typename... Args> typename HashMap, template <typename> typename Hash,
+          typename... Mapped, typename Variant>
+bool try_get_hash_map_context_fixed(Variant& variant, const std::vector<DataTypePtr>& data_types) {
     Sizes key_sizes;
 
     bool use_fixed_key = true;
     bool has_null = false;
     size_t key_byte_size = 0;
 
-    for (auto ctx : expr_ctxs) {
-        const auto& data_type = ctx->root()->data_type();
+    for (const auto& data_type : data_types) {
         if (!data_type->have_maximum_size_of_value()) {
             use_fixed_key = false;
             break;
@@ -73,7 +82,7 @@ bool try_get_hash_map_context_fixed(Variant& variant, const VExprContextSPtrs& e
         key_byte_size += key_sizes.back();
     }
 
-    size_t bitmap_size = has_null ? get_bitmap_size(expr_ctxs.size()) : 0;
+    size_t bitmap_size = has_null ? get_bitmap_size(data_types.size()) : 0;
     if (bitmap_size + key_byte_size > sizeof(UInt256)) {
         use_fixed_key = false;
     }

be/src/vec/exec/join/vhash_join_node.cpp

@@ -1053,8 +1053,18 @@ void HashJoinNode::_hash_table_init(RuntimeState* state) {
                     return;
                 }
 
+                std::vector<DataTypePtr> data_types;
+                for (size_t i = 0; i != _build_expr_ctxs.size(); ++i) {
+                    auto& ctx = _build_expr_ctxs[i];
+                    auto data_type = ctx->root()->data_type();
+                    if (_should_convert_build_side_to_nullable[i]) {
+                        data_type = make_nullable(data_type);
+                    }
+                    data_types.emplace_back(std::move(data_type));
+                }
+
                 if (!try_get_hash_map_context_fixed<JoinHashMap, HashCRC32>(*_hash_table_variants,
-                                                                            _build_expr_ctxs)) {
+                                                                            data_types)) {
                     _hash_table_variants->emplace<SerializedHashTableContext>();
                 }
             },

regression-test/data/query_p0/join/test_half_join_nullable_build_side.out

@@ -126,3 +126,11 @@
 2	111	111
 3	1111	1111
 
+-- !sql30 --
+1	11	11	1	11	11
+2	111	111	4	111	111
+2	111	111	2	111	111
+3	1111	1111	3	1111	1111
+4	\N	\N	\N	\N	\N
+5	1111	1111	3	1111	1111
+

regression-test/suites/query_p0/join/test_half_join_nullable_build_side.groovy

@@ -277,4 +277,13 @@ suite("test_half_join_nullable_build_side", "query,p0") {
             test_half_join_nullable_build_side_l2 l right semi join test_half_join_nullable_build_side_r2 r on  l.v2 = r.v2
         order by 1, 2, 3;
     """
+
+    qt_sql30 """
+        select
+            *
+        from
+            test_half_join_nullable_build_side_l2 l
+            left join test_half_join_nullable_build_side_l r on  l.v2 <=> r.v2
+        order by 1, 2, 3;
+    """
 }