[Improvement](hash) opt for pack_fixeds (#59410)

<img width="298" height="2142" alt="图片"
src="https://github.com/user-attachments/assets/3be70146-c4bd-4ff7-ac96-2645f89fed14"
/>

This pull request refactors and optimizes the handling of null maps and
key packing in hash join and hash table code, with a focus on improving
SIMD (Single Instruction, Multiple Data) usage and simplifying null
bitmap logic. The changes replace older byte-searching utilities with
new, more efficient SIMD-based functions, update how null bitmaps are
packed and processed, and streamline column null data replacement.
Additionally, the logic for determining hash key types and handling
fixed key serialization is improved for better correctness and
performance.

Key improvements and changes:

### SIMD utilities and null map handling

* Introduced new SIMD-based functions `contain_one` and `contain_zero`
in `simd/bits.h`, replacing the older `contain_byte` and related logic
for checking the presence of ones or zeros in null maps, resulting in
more efficient null detection.
* Updated all usages of null map checks throughout the codebase to use
the new `contain_one` and `contain_zero` functions, simplifying and
unifying the logic for detecting nulls in columns and filters.
[[1]](diffhunk://#diff-0732e01c1a3f38997ada381c43aff98286e86ca7519db5469a6e4dcdec5bce44L195-L200)
[[2]](diffhunk://#diff-3110bab7d558f46b88ae1958b09ac369a92cac4bff98b280b2cf83d2d7aecbf4L117-R118)
[[3]](diffhunk://#diff-3110bab7d558f46b88ae1958b09ac369a92cac4bff98b280b2cf83d2d7aecbf4L369-R371)
[[4]](diffhunk://#diff-8981dd2e1f08aaa46a97aeef27bd906c64d1bb08deedc0fe1d94c1c49dc064ceL100-R100)
[[5]](diffhunk://#diff-9fd61a223bcb3b7a9cb93c2d26c9364d8cce2131673fe286f22a80b09c6fd2c6L283-R283)
[[6]](diffhunk://#diff-9fd61a223bcb3b7a9cb93c2d26c9364d8cce2131673fe286f22a80b09c6fd2c6L601-R605)

### Hash key and null bitmap packing

* Refactored the logic for packing null maps into hash keys in
`MethodKeysFixed`, introducing new templates and helper functions for
interleaved null map packing, and replacing the old bitmap size
calculation with a simplified approach. This improves both performance
and maintainability.
[[1]](diffhunk://#diff-b8623712a5a1728bb77cc67b6ee1bbf16ef2b842044f6f6bab64c3fc5c4575f3R478-R540)
[[2]](diffhunk://#diff-b8623712a5a1728bb77cc67b6ee1bbf16ef2b842044f6f6bab64c3fc5c4575f3L500-R611)
* Updated the logic for initializing and inserting keys, ensuring
correct handling of nulls and simplifying offset calculations for key
data.
[[1]](diffhunk://#diff-b8623712a5a1728bb77cc67b6ee1bbf16ef2b842044f6f6bab64c3fc5c4575f3R653)
[[2]](diffhunk://#diff-b8623712a5a1728bb77cc67b6ee1bbf16ef2b842044f6f6bab64c3fc5c4575f3L619-R692)
[[3]](diffhunk://#diff-b8623712a5a1728bb77cc67b6ee1bbf16ef2b842044f6f6bab64c3fc5c4575f3L645-R712)

### Column null data replacement

* Simplified the `replace_column_null_data` methods for vector and
decimal columns by removing unnecessary null count checks and optimizing
the replacement logic.
[[1]](diffhunk://#diff-3fa47f544ff08bb2c8232af99312c0bbf2c58cac9da7a2b06473282b99ad5aa4L528-R530)
[[2]](diffhunk://#diff-5fdf450def955da3201cc889aa870d94def054d1168f1ef3def32e8f009dc65aL526-L529)

### Hash key type logic

* Improved the logic for determining the hash key type in
`hash_key_type.h` to handle cases where the number of data types exceeds
the bit size, defaulting to serialized keys as needed.
[[1]](diffhunk://#diff-4f1fb8a89cd0e13a719c3427b1ae7581b42cb7325755a3ceac4c44bdc64bd144R83-R86)
[[2]](diffhunk://#diff-4f1fb8a89cd0e13a719c3427b1ae7581b42cb7325755a3ceac4c44bdc64bd144L97-R101)

### Code cleanup and dependency updates

* Removed unused functions and updated includes to ensure all SIMD
utilities are properly imported where needed.
[[1]](diffhunk://#diff-b8623712a5a1728bb77cc67b6ee1bbf16ef2b842044f6f6bab64c3fc5c4575f3R20-R26)
[[2]](diffhunk://#diff-b8623712a5a1728bb77cc67b6ee1bbf16ef2b842044f6f6bab64c3fc5c4575f3R36)
[[3]](diffhunk://#diff-b8623712a5a1728bb77cc67b6ee1bbf16ef2b842044f6f6bab64c3fc5c4575f3L292-L295)

These changes collectively improve performance, maintainability, and
correctness in hash join operations, especially in handling nullable
columns and SIMD optimizations.

Author: BiteTheDDDDt

be/src/pipeline/exec/hashjoin_build_sink.h

@@ -192,11 +192,8 @@ struct ProcessHashTableBuild {
                bool* has_null_key) {
         if (null_map) {
             // first row is mocked and is null
-            // TODO: Need to test the for loop. break may better
-            for (uint32_t i = 1; i < _rows; i++) {
-                if ((*null_map)[i]) {
-                    *has_null_key = true;
-                }
+            if (simd::contain_one(null_map->data() + 1, _rows - 1)) {
+                *has_null_key = true;
             }
             if (short_circuit_for_null && *has_null_key) {
                 return Status::OK();
@@ -208,7 +205,7 @@ struct ProcessHashTableBuild {
                 _rows, _batch_size, *has_null_key, hash_table_ctx.direct_mapping_range());
 
         // In order to make the null keys equal when using single null eq, all null keys need to be set to default value.
-        if (_build_raw_ptrs.size() == 1 && null_map) {
+        if (_build_raw_ptrs.size() == 1 && null_map && *has_null_key) {
             _build_raw_ptrs[0]->assume_mutable()->replace_column_null_data(null_map->data());
         }
 

be/src/pipeline/exec/join/process_hash_table_probe_impl.h

@@ -114,8 +114,8 @@ void ProcessHashTableProbe<JoinOpType>::build_side_output_column(vectorized::Mut
             _build_column_has_null[i] = false;
             if (_right_output_slot_flags[i] && column.is_nullable()) {
                 const auto& nullable = assert_cast<const vectorized::ColumnNullable&>(column);
-                _build_column_has_null[i] = !simd::contain_byte(
-                        nullable.get_null_map_data().data() + 1, nullable.size() - 1, 1);
+                _build_column_has_null[i] = !simd::contain_one(
+                        nullable.get_null_map_data().data() + 1, nullable.size() - 1);
             }
         }
     }
@@ -192,7 +192,9 @@ typename HashTableType::State ProcessHashTableProbe<JoinOpType>::_init_probe_sid
         hash_table_ctx.arena.clear();
         // In order to make the null keys equal when using single null eq, all null keys need to be set to default value.
         if (_parent->_probe_columns.size() == 1 && null_map) {
-            _parent->_probe_columns[0]->assume_mutable()->replace_column_null_data(null_map);
+            if (simd::contain_one(null_map, probe_rows)) {
+                _parent->_probe_columns[0]->assume_mutable()->replace_column_null_data(null_map);
+            }
         }
 
         hash_table_ctx.init_serialized_keys(_parent->_probe_columns, probe_rows, null_map, true,
@@ -366,8 +368,7 @@ Status ProcessHashTableProbe<JoinOpType>::finalize_block_with_filter(
         }
         const auto& column_filter =
                 assert_cast<const vectorized::ColumnUInt8*>(filter_ptr.get())->get_data();
-        bool need_filter =
-                simd::count_zero_num((int8_t*)column_filter.data(), column_filter.size()) != 0;
+        bool need_filter = simd::contain_zero(column_filter.data(), column_filter.size());
         if (need_filter) {
             row_indexs.filter(column_filter);
         }

be/src/pipeline/exec/nested_loop_join_probe_operator.h

@@ -97,9 +97,7 @@ class NestedLoopJoinProbeLocalState final
                     }
                     if (!_cur_probe_row_visited_flags[i]) {
                         _cur_probe_row_visited_flags[i] =
-                                simd::contain_byte<uint8_t>(filter.data() + offset, end - offset, 1)
-                                        ? 1
-                                        : 0;
+                                simd::contain_one(filter.data() + offset, end - offset);
                     }
                     end = offset;
                 }

be/src/util/simd/bits.h

@@ -261,14 +261,6 @@ static size_t find_byte(const T* data, size_t start, size_t end, T byte) {
     return (T*)p - data;
 }
 
-template <typename T>
-bool contain_byte(const T* __restrict data, const size_t length, const signed char byte) {
-    if (length == 0) {
-        return false;
-    }
-    return nullptr != std::memchr(reinterpret_cast<const void*>(data), byte, length);
-}
-
 inline size_t find_one(const std::vector<uint8_t>& vec, size_t start) {
     return find_byte<uint8_t>(vec, start, 1);
 }
@@ -281,5 +273,58 @@ inline size_t find_zero(const std::vector<uint8_t>& vec, size_t start) {
     return find_byte<uint8_t>(vec, start, 0);
 }
 
+inline bool contain_one(const uint8_t* __restrict data, size_t size) {
+    size_t i = 0;
+#if defined(__AVX2__)
+    for (; i + 32 <= size; i += 32) {
+        __m256i chunk = _mm256_loadu_si256(reinterpret_cast<const __m256i*>(data + i));
+        if (!_mm256_testz_si256(chunk, chunk)) {
+            return true;
+        }
+    }
+#elif defined(__SSE2__)
+    const __m128i zero = _mm_setzero_si128();
+    for (; i + 16 <= size; i += 16) {
+        __m128i chunk = _mm_loadu_si128(reinterpret_cast<const __m128i*>(data + i));
+        if (_mm_movemask_epi8(_mm_cmpeq_epi8(chunk, zero)) != 0xFFFF) {
+            return true;
+        }
+    }
+#endif
+    for (; i < size; ++i) {
+        if (data[i]) {
+            return true;
+        }
+    }
+    return false;
+}
+
+inline bool contain_zero(const uint8_t* __restrict data, size_t size) {
+    size_t i = 0;
+#if defined(__AVX2__)
+    const __m256i zero = _mm256_setzero_si256();
+    for (; i + 32 <= size; i += 32) {
+        __m256i chunk = _mm256_loadu_si256(reinterpret_cast<const __m256i*>(data + i));
+        if (_mm256_movemask_epi8(_mm256_cmpeq_epi8(chunk, zero)) != 0) {
+            return true;
+        }
+    }
+#elif defined(__SSE2__)
+    const __m128i zero = _mm_setzero_si128();
+    for (; i + 16 <= size; i += 16) {
+        __m128i chunk = _mm_loadu_si128(reinterpret_cast<const __m128i*>(data + i));
+        if (_mm_movemask_epi8(_mm_cmpeq_epi8(chunk, zero)) != 0) {
+            return true;
+        }
+    }
+#endif
+    for (; i < size; ++i) {
+        if (!data[i]) {
+            return true;
+        }
+    }
+    return false;
+}
+
 } // namespace doris::simd
 #include "common/compile_check_end.h"

be/src/vec/columns/column_decimal.cpp

@@ -521,10 +521,6 @@ void ColumnDecimal<T>::compare_internal(size_t rhs_row_id, const IColumn& rhs,
 template <PrimitiveType T>
 void ColumnDecimal<T>::replace_column_null_data(const uint8_t* __restrict null_map) {
     auto s = size();
-    size_t null_count = s - simd::count_zero_num((const int8_t*)null_map, s);
-    if (0 == null_count) {
-        return;
-    }
     for (size_t i = 0; i < s; ++i) {
         data[i] = null_map[i] ? value_type() : data[i];
     }

be/src/vec/columns/column_nullable.cpp

@@ -280,7 +280,7 @@ size_t ColumnNullable::serialize_impl(char* pos, const size_t row) const {
 }
 
 void ColumnNullable::serialize(StringRef* keys, size_t num_rows) const {
-    const bool has_null = simd::contain_byte(get_null_map_data().data(), num_rows, 1);
+    const bool has_null = simd::contain_one(get_null_map_data().data(), num_rows);
     const auto* __restrict null_map =
             assert_cast<const ColumnUInt8&>(get_null_map_column()).get_data().data();
     _nested_column->serialize_with_nullable(keys, num_rows, has_null, null_map);
@@ -598,11 +598,11 @@ void ColumnNullable::sort_column(const ColumnSorter* sorter, EqualFlags& flags,
 }
 
 bool ColumnNullable::only_null() const {
-    return !simd::contain_byte(get_null_map_data().data(), size(), 0);
+    return !simd::contain_zero(get_null_map_data().data(), size());
 }
 
 bool ColumnNullable::has_null(size_t begin, size_t end) const {
-    return simd::contain_byte(get_null_map_data().data() + begin, end - begin, 1);
+    return simd::contain_one(get_null_map_data().data() + begin, end - begin);
 }
 
 bool ColumnNullable::has_null() const {

be/src/vec/columns/column_vector.cpp

@@ -525,12 +525,9 @@ MutableColumnPtr ColumnVector<T>::permute(const IColumn::Permutation& perm, size
 template <PrimitiveType T>
 void ColumnVector<T>::replace_column_null_data(const uint8_t* __restrict null_map) {
     auto s = size();
-    size_t null_count = s - simd::count_zero_num((const int8_t*)null_map, s);
-    if (0 == null_count) {
-        return;
-    }
+    auto value = default_value();
     for (size_t i = 0; i < s; ++i) {
-        data[i] = null_map[i] ? default_value() : data[i];
+        data[i] = null_map[i] ? value : data[i];
     }
 }
 

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

@@ -80,6 +80,10 @@ inline HashKeyType get_hash_key_type_with_fixed(size_t size) {
 }
 
 inline HashKeyType get_hash_key_type_fixed(const std::vector<vectorized::DataTypePtr>& data_types) {
+    if (data_types.size() >= vectorized::BITSIZE) {
+        return HashKeyType::serialized;
+    }
+
     bool has_null = false;
     size_t key_byte_size = 0;
 
@@ -94,8 +98,7 @@ inline HashKeyType get_hash_key_type_fixed(const std::vector<vectorized::DataTyp
         }
     }
 
-    size_t bitmap_size = has_null ? vectorized::get_bitmap_size(data_types.size()) : 0;
-    return get_hash_key_type_with_fixed(bitmap_size + key_byte_size);
+    return get_hash_key_type_with_fixed(has_null + key_byte_size);
 }
 
 inline HashKeyType get_hash_key_type(const std::vector<vectorized::DataTypePtr>& data_types) {