[Improvement](hash) opt for pack_fixeds

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

@@ -523,10 +523,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) {