diff --git a/be/CMakeLists.txt b/be/CMakeLists.txt
index 524d0322873ec3..f07dba5c073be0 100644
--- a/be/CMakeLists.txt
+++ b/be/CMakeLists.txt
@@ -448,6 +448,10 @@ if (COMPILER_CLANG)
                         -Wunused-macros
                         -Wconversion
                         -Wthread-safety)
+    # Clang >= 17 flags namespace-scope shadowing (e.g. enum constants vs
+    # protobuf-generated enums) that older toolchains accepted; keep the
+    # warnings visible but non-fatal so newer compilers can build.
+    add_compile_options(-Wno-error=shadow)
     add_compile_options(-Wno-gnu-statement-expression
                         -Wno-implicit-float-conversion
                         -Wno-sign-conversion
@@ -941,6 +945,9 @@ if (MAKE_TEST)
         add_compile_options(
         -Wno-implicit-int-conversion
         -Wno-shorten-64-to-32
+        # Newer libstdc++ (>= 12) marks std::get_temporary_buffer deprecated;
+        # test code using std::stable_sort trips over it under -Werror.
+        -Wno-deprecated-declarations
         )
     endif()
 endif ()
diff --git a/be/src/common/signal_handler.h b/be/src/common/signal_handler.h
index 334b2ea99af6ce..360b7bece4c425 100644
--- a/be/src/common/signal_handler.h
+++ b/be/src/common/signal_handler.h
@@ -337,40 +337,28 @@ void InvokeDefaultSignalHandler(int signal_number) {
 // See also comments in FailureSignalHandler().
 static pthread_t* g_entered_thread_id_pointer = nullptr;
 
-// Wrapper of __sync_val_compare_and_swap. If the GCC extension isn't
-// defined, we try the CPU specific logics (we only support x86 and
-// x86_64 for now) first, then use a naive implementation, which has a
-// race condition.
+// Wrapper of the compiler's atomic compare-and-swap builtin.
+// __atomic_compare_exchange_n is available on every architecture supported
+// by GCC/Clang (x86, aarch64, ...). The previous fallback chain
+// (HAVE___SYNC_VAL_COMPARE_AND_SWAP, which nothing in this CMake-based
+// project ever defines, then x86-only inline asm) ended in a naive
+// non-atomic read-check-write on aarch64. That races when several threads
+// crash at the same time on many-core ARM machines: multiple threads win
+// the FailureSignalHandler election below and dump concurrently.
 template <typename T>
 T sync_val_compare_and_swap(T* ptr, T oldval, T newval) {
-#if defined(HAVE___SYNC_VAL_COMPARE_AND_SWAP)
-    return __sync_val_compare_and_swap(ptr, oldval, newval);
-#elif defined(__GNUC__) && (defined(__i386__) || defined(__x86_64__))
-    T ret;
-    __asm__ __volatile__("lock; cmpxchg %1, (%2);"
-                         : "=a"(ret)
-                         // GCC may produces %sil or %dil for
-                         // constraint "r", but some of apple's gas
-                         // dosn't know the 8 bit registers.
-                         // We use "q" to avoid these registers.
-                         : "q"(newval), "q"(ptr), "a"(oldval)
-                         : "memory", "cc");
-    return ret;
-#else
-    T ret = *ptr;
-    if (ret == oldval) {
-        *ptr = newval;
-    }
-    return ret;
-#endif
+    T expected = oldval;
+    __atomic_compare_exchange_n(ptr, &expected, newval, false, __ATOMIC_SEQ_CST,
+                                __ATOMIC_SEQ_CST);
+    return expected;
 }
 
 // Dumps signal and stack frame information, and invokes the default
 // signal handler once our job is done.
 void FailureSignalHandler(int signal_number, siginfo_t* signal_info, void* ucontext) {
-    // First check if we've already entered the function.  We use an atomic
-    // compare and swap operation for platforms that support it.  For other
-    // platforms, we use a naive method that could lead to a subtle race.
+    // First check if we've already entered the function.  The election uses
+    // the compiler's atomic compare-and-swap builtin, which is available on
+    // every supported platform (x86_64 and aarch64 alike).
 
     // We assume pthread_self() is async signal safe, though it's not
     // officially guaranteed.
diff --git a/be/src/core/column/columns_common.cpp b/be/src/core/column/columns_common.cpp
index bfaee36a8e0058..53d61b2bce7b8f 100644
--- a/be/src/core/column/columns_common.cpp
+++ b/be/src/core/column/columns_common.cpp
@@ -45,7 +45,10 @@ size_t count_bytes_in_filter(const IColumn::Filter& filt) {
     const Int8* pos = reinterpret_cast<const Int8*>(filt.data());
     const Int8* end = pos + filt.size();
 
-#if defined(__SSE2__) || defined(__aarch64__) && defined(__POPCNT__)
+// NOTE: the old guard already parsed as `__SSE2__ || (__aarch64__ && __POPCNT__)`
+// (`&&` binds tighter), and ARM toolchains never define __POPCNT__, so this SIMD
+// block was compiled out on aarch64; dropping the __POPCNT__ gate enables it.
+#if defined(__SSE2__) || defined(__aarch64__)
     const __m128i zero16 = _mm_setzero_si128();
     const Int8* end64 = pos + filt.size() / 64 * 64;
 
diff --git a/be/src/core/value/bitmap_value.h b/be/src/core/value/bitmap_value.h
index ae4249c5b19cb5..1ce317442a65ae 100644
--- a/be/src/core/value/bitmap_value.h
+++ b/be/src/core/value/bitmap_value.h
@@ -2971,12 +2971,23 @@ class BitmapValue {
         _set.clear();
     }
 
+    // NOTE: the enumerators (EMPTY/SINGLE/BITMAP/SET) collide with protobuf
+    // enum values exported at namespace scope by olap_file.pb.h (e.g.
+    // doris::BITMAP); clang >= 17 -Wshadow flags the shadowing as an error
+    // under -Werror, so suppress it just for this declaration.
+#if defined(__clang__)
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wshadow"
+#endif
     enum BitmapDataType {
         EMPTY = 0,
         SINGLE = 1, // single element
         BITMAP = 2, // more than one elements
         SET = 3     // elements count less or equal than 32
     };
+#if defined(__clang__)
+#pragma clang diagnostic pop
+#endif
     uint64_t _sv = 0; // store the single value when _type == SINGLE
     // !FIXME: We should rethink the logic about _bitmap and _is_shared
     mutable std::shared_ptr<detail::Roaring64Map> _bitmap; // used when _type == BITMAP
diff --git a/be/src/exec/operator/bucketed_aggregation_source_operator.cpp b/be/src/exec/operator/bucketed_aggregation_source_operator.cpp
index 993f1868e17a3e..7128c50f7d8ac5 100644
--- a/be/src/exec/operator/bucketed_aggregation_source_operator.cpp
+++ b/be/src/exec/operator/bucketed_aggregation_source_operator.cpp
@@ -499,7 +499,10 @@ Status BucketedAggLocalState::_output_bucket(RuntimeState* state, Block* block,
 Status BucketedAggLocalState::_merge_and_output_null_keys(RuntimeState* state, Block* block) {
     auto& shared_state = *_shared_state;
     size_t key_size = shared_state.probe_expr_ctxs.size();
-    int merge_target = shared_state.merge_target_instance.load(std::memory_order_relaxed);
+    // acquire: the loaded index is used to dereference per-instance data
+    // published by other threads; relaxed would rely on data-dependency
+    // ordering, which the C++ memory model does not guarantee (aarch64).
+    int merge_target = shared_state.merge_target_instance.load(std::memory_order_acquire);
 
     // Merge null keys from all 256 buckets (in merge target) into bucket 0.
     // After per-bucket merge, each bucket in merge target may have its own null key data.
diff --git a/be/src/exprs/function/function_string_misc.cpp b/be/src/exprs/function/function_string_misc.cpp
index 663fa0fe018591..f4cd21e8320db6 100644
--- a/be/src/exprs/function/function_string_misc.cpp
+++ b/be/src/exprs/function/function_string_misc.cpp
@@ -31,7 +31,6 @@
 #include <cstdint>
 #include <cstdlib>
 #include <cstring>
-#include <format>
 #include <iomanip>
 #include <memory>
 #include <random>
@@ -229,7 +228,7 @@ class FunctionAutoPartitionName : public IFunction {
             // check the name of length
             int len = res_p.size();
             if (len > 50) {
-                res_p = std::format("{}_{:08x}", res_p.substr(0, 50), to_hash_code(res_p));
+                res_p = fmt::format("{}_{:08x}", res_p.substr(0, 50), to_hash_code(res_p));
                 len = res_p.size();
             }
             curr_len += len;
diff --git a/be/src/format/parquet/parquet_column_convert.h b/be/src/format/parquet/parquet_column_convert.h
index 5ec05d450d8b31..3a6306affb362c 100644
--- a/be/src/format/parquet/parquet_column_convert.h
+++ b/be/src/format/parquet/parquet_column_convert.h
@@ -22,6 +22,7 @@
 #include <libdivide.h>
 
 #include <chrono>
+#include <cmath> // std::pow (used by the half-float decoding path)
 #include <limits>
 
 #include "common/cast_set.h"
@@ -593,7 +594,7 @@ class Float16PhysicalConverter : public PhysicalToLogicalConverter {
                 // half subnormal:
                 //    value = (-1)^sign * (mant / 2^10) * 2^(1 - bias)
                 // half bias = 15 → exponent = 1 - 15 = -14
-                float f = (static_cast<float>(mant) / 1024.0F) * std::powf(2.0F, -14.0F);
+                float f = (static_cast<float>(mant) / 1024.0F) * std::pow(2.0F, -14.0F);
                 return sign ? -f : f;
             }
         } else if (exp == 0x1F) {
diff --git a/be/src/glibc-compatibility/CMakeLists.txt b/be/src/glibc-compatibility/CMakeLists.txt
index 370d73466918ae..6ebae6f2000199 100644
--- a/be/src/glibc-compatibility/CMakeLists.txt
+++ b/be/src/glibc-compatibility/CMakeLists.txt
@@ -57,6 +57,9 @@ if (GLIBC_COMPATIBILITY)
     # libcalls. Workaround: Use object file so that linker will always take a
     # look at its symbol table.
     list(REMOVE_ITEM glibc_compatibility_sources musl/getrandom.c)
+    # NOTE: the OBJECT lib must always provide the resolv_shim symbol where it
+    # exists (resolv_shim.c is a no-op on glibc < 2.34); keep it out of the archive.
+    list(REMOVE_ITEM glibc_compatibility_sources resolv_shim.c)
     # NOTE(amos): sanitizers might generate memcpy references that are too late to
     # refer. Let's also extract memcpy definitions explicitly to avoid UNDEF GLIBC 2.14.
     #
@@ -65,9 +68,9 @@ if (GLIBC_COMPATIBILITY)
     # before ASAN shadow memory is initialized, causing SIGSEGV. Skip custom memcpy in
     # this case and fall back to glibc's memcpy.
     if (ARCH_ARM AND (CMAKE_BUILD_TYPE STREQUAL "ASAN_UT" OR CMAKE_BUILD_TYPE STREQUAL "ASAN"))
-        add_library(glibc-compatibility-explicit OBJECT musl/getrandom.c)
+        add_library(glibc-compatibility-explicit OBJECT musl/getrandom.c resolv_shim.c)
     else()
-        add_library(glibc-compatibility-explicit OBJECT musl/getrandom.c ${MEMCPY_SOURCE})
+        add_library(glibc-compatibility-explicit OBJECT musl/getrandom.c resolv_shim.c ${MEMCPY_SOURCE})
     endif()
     target_compile_options(glibc-compatibility-explicit PRIVATE -fPIC)
     add_library(glibc-compatibility STATIC ${glibc_compatibility_sources})
diff --git a/be/src/glibc-compatibility/resolv_shim.c b/be/src/glibc-compatibility/resolv_shim.c
new file mode 100644
index 00000000000000..6c75c99ab21682
--- /dev/null
+++ b/be/src/glibc-compatibility/resolv_shim.c
@@ -0,0 +1,48 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// glibc >= 2.34 demoted the double-underscore resolver entry points
+// (__res_nsearch & friends) to non-default compat versions, so newly linked
+// binaries cannot bind them anymore. The prebuilt thirdparty krb5 archive
+// (dnsglue.o) still references __res_nsearch, which breaks the doris_be link
+// on Ubuntu 22.04 (glibc 2.35). Provide a thin forwarder to the public
+// res_nsearch entry point, which is the identical implementation (same
+// symbol address in libc).
+//
+// The shim must only exist where it is needed: on glibc < 2.34,
+// __res_nsearch is still a default-versioned libc symbol, and <resolv.h>
+// there #defines res_nsearch as __res_nsearch, which would fold the
+// forwarder below into infinite self-recursion (clang -Winfinite-recursion
+// errors out under -Werror, e.g. on the AlmaLinux 8 / glibc 2.28 CI image).
+
+#include <resolv.h>
+#include <sys/types.h>
+
+#if defined(__GLIBC__) && __GLIBC_PREREQ(2, 34)
+
+int __res_nsearch(res_state statp, const char* dname, int class_, int type,
+                  unsigned char* answer, int anslen) {
+    return res_nsearch(statp, dname, class_, type, answer, anslen);
+}
+
+#else
+
+// Keep the translation unit non-empty (-Wpedantic forbids an empty one);
+// no shim is required on glibc < 2.34.
+typedef int doris_resolv_shim_unused_t;
+
+#endif
diff --git a/be/src/io/cache/block_file_cache_profile.cpp b/be/src/io/cache/block_file_cache_profile.cpp
index 6e4222dbc11852..a3ce8ae2fefa1f 100644
--- a/be/src/io/cache/block_file_cache_profile.cpp
+++ b/be/src/io/cache/block_file_cache_profile.cpp
@@ -28,31 +28,23 @@ namespace doris::io {
 
 std::shared_ptr<AtomicStatistics> FileCacheMetrics::report() {
     std::shared_ptr<AtomicStatistics> output_stats = std::make_shared<AtomicStatistics>();
-    std::lock_guard lock(_mtx);
-    output_stats->num_io_bytes_read_from_cache += _statistics->num_io_bytes_read_from_cache;
-    output_stats->num_io_bytes_read_from_remote += _statistics->num_io_bytes_read_from_remote;
-    output_stats->num_io_bytes_read_from_peer += _statistics->num_io_bytes_read_from_peer;
+    output_stats->num_io_bytes_read_from_cache += _statistics.num_io_bytes_read_from_cache;
+    output_stats->num_io_bytes_read_from_remote += _statistics.num_io_bytes_read_from_remote;
+    output_stats->num_io_bytes_read_from_peer += _statistics.num_io_bytes_read_from_peer;
     output_stats->inverted_index_bytes_read_from_remote +=
-            _statistics->inverted_index_bytes_read_from_remote;
+            _statistics.inverted_index_bytes_read_from_remote;
     output_stats->segment_footer_index_bytes_read_from_remote +=
-            _statistics->segment_footer_index_bytes_read_from_remote;
+            _statistics.segment_footer_index_bytes_read_from_remote;
     return output_stats;
 }
 
 void FileCacheMetrics::update(FileCacheStatistics* input_stats) {
-    if (_statistics == nullptr) {
-        std::lock_guard<std::mutex> lock(_mtx);
-        if (_statistics == nullptr) {
-            _statistics = std::make_shared<AtomicStatistics>();
-            register_entity();
-        }
-    }
-    _statistics->num_io_bytes_read_from_cache += input_stats->bytes_read_from_local;
-    _statistics->num_io_bytes_read_from_remote += input_stats->bytes_read_from_remote;
-    _statistics->num_io_bytes_read_from_peer += input_stats->bytes_read_from_peer;
-    _statistics->inverted_index_bytes_read_from_remote +=
+    _statistics.num_io_bytes_read_from_cache += input_stats->bytes_read_from_local;
+    _statistics.num_io_bytes_read_from_remote += input_stats->bytes_read_from_remote;
+    _statistics.num_io_bytes_read_from_peer += input_stats->bytes_read_from_peer;
+    _statistics.inverted_index_bytes_read_from_remote +=
             input_stats->inverted_index_bytes_read_from_remote;
-    _statistics->segment_footer_index_bytes_read_from_remote +=
+    _statistics.segment_footer_index_bytes_read_from_remote +=
             input_stats->segment_footer_index_bytes_read_from_remote;
 }
 
diff --git a/be/src/io/cache/block_file_cache_profile.h b/be/src/io/cache/block_file_cache_profile.h
index 8b594add9977b8..6462cda9a7f3db 100644
--- a/be/src/io/cache/block_file_cache_profile.h
+++ b/be/src/io/cache/block_file_cache_profile.h
@@ -23,7 +23,6 @@
 #include <atomic>
 #include <cstdint>
 #include <memory>
-#include <mutex>
 #include <unordered_map>
 
 #include "common/metrics/doris_metrics.h"
@@ -49,21 +48,23 @@ class FileCacheMetrics {
         return s_metrics;
     }
 
-    FileCacheMetrics() {
-        FileCacheStatistics stats;
-        update(&stats);
-    }
+    // The counters are value members, so they are fully constructed before this
+    // body runs; registering here (instead of lazily on first update) is safe
+    // even if a metrics callback fires right after registration, since report()
+    // only reads the counters. There is no publication race: instance() is a
+    // magic static and returns only after construction completes.
+    FileCacheMetrics() { register_entity(); }
 
     void update(FileCacheStatistics* stats);
     std::shared_ptr<AtomicStatistics> report();
+    // Public for tests: pushes the current counters into the DorisMetrics
+    // gauges without waiting for the periodic metrics hook.
+    void update_metrics_callback();
 
 private:
     void register_entity();
-    void update_metrics_callback();
 
-    std::mutex _mtx;
-    // use shared_ptr for concurrent
-    std::shared_ptr<AtomicStatistics> _statistics;
+    AtomicStatistics _statistics;
 };
 
 FileCacheStatistics diff_file_cache_statistics(const FileCacheStatistics& current,
diff --git a/be/src/service/doris_main.cpp b/be/src/service/doris_main.cpp
index dc55d6ce14228b..fdd9a6e1ee3116 100644
--- a/be/src/service/doris_main.cpp
+++ b/be/src/service/doris_main.cpp
@@ -239,9 +239,15 @@ void check_required_instructions_impl(volatile InstructionFail& fail) {
     __asm__ volatile("vpabsw %%zmm0, %%zmm0" : : : "zmm0");
 #endif
 
-#if defined(__ARM_NEON__)
+// GCC aarch64 defines __ARM_NEON, Clang additionally defines __ARM_NEON__ on
+// some targets (e.g. Apple). The 32-bit "vadd.i32 q8,..." syntax is AArch32
+// only and does not assemble on AArch64, which needs "add v8.4s,...". Pick
+// the spelling per architecture so every ARM toolchain passes this check.
+#if defined(__ARM_NEON__) || defined(__ARM_NEON)
     fail = InstructionFail::ARM_NEON;
-#ifndef __APPLE__
+#if defined(__aarch64__)
+    __asm__ volatile("add v8.4s, v8.4s, v8.4s" : : : "v8");
+#elif !defined(__APPLE__)
     __asm__ volatile("vadd.i32  q8, q8, q8" : : : "q8");
 #endif
 #endif
diff --git a/be/src/service/http/action/be_thread_stack_action.cpp b/be/src/service/http/action/be_thread_stack_action.cpp
index a7d89d743f32ea..0312cdfeb1573b 100644
--- a/be/src/service/http/action/be_thread_stack_action.cpp
+++ b/be/src/service/http/action/be_thread_stack_action.cpp
@@ -133,7 +133,9 @@ pid_t get_current_tid() {
     return static_cast<pid_t>(syscall(SYS_gettid));
 }
 
-void append_frame(SignalContextCapture* capture, uintptr_t pc) {
+// Only called from the x86_64 libunwind path below; on aarch64 it is
+// intentionally unused, so keep -Wunused-function quiet.
+[[maybe_unused]] void append_frame(SignalContextCapture* capture, uintptr_t pc) {
     if (pc == 0 || capture->size >= capture->frame_pointers.size()) {
         return;
     }
diff --git a/be/src/storage/index/bloom_filter/ngram_bloom_filter.cpp b/be/src/storage/index/bloom_filter/ngram_bloom_filter.cpp
index 0a7ee243b36251..f3ac312e097cdd 100644
--- a/be/src/storage/index/bloom_filter/ngram_bloom_filter.cpp
+++ b/be/src/storage/index/bloom_filter/ngram_bloom_filter.cpp
@@ -20,6 +20,8 @@
 #include <gen_cpp/segment_v2.pb.h>
 #include <glog/logging.h>
 
+#include <cstring>
+
 #include "absl/strings/substitute.h"
 #include "util/hash/city.h"
 
@@ -40,11 +42,12 @@ Status NGramBloomFilter::init(const char* buf, size_t size, HashStrategyPB strat
         return Status::InvalidArgument(absl::Substitute("invalid strategy:$0", strategy));
     }
     words = (_size + sizeof(UnderType) - 1) / sizeof(UnderType);
-    filter.reserve(words);
-    const auto* from = reinterpret_cast<const UnderType*>(buf);
-    for (size_t i = 0; i < words; ++i) {
-        filter[i] = from[i];
-    }
+    filter.assign(words, 0);
+    // buf points into an arbitrarily-offset page buffer; a plain
+    // reinterpret_cast<const uint64_t*> read would be misaligned UB.
+    // Copy exactly size bytes: the tail bytes of the last word must stay
+    // zero so contains() bit comparisons match query-side filters.
+    memcpy(filter.data(), buf, size);
 
     return Status::OK();
 }
diff --git a/be/src/storage/index/snii/encoding/crc32c.cpp b/be/src/storage/index/snii/encoding/crc32c.cpp
index 39d7c6f58fe487..8271cd63c81b0d 100644
--- a/be/src/storage/index/snii/encoding/crc32c.cpp
+++ b/be/src/storage/index/snii/encoding/crc32c.cpp
@@ -38,6 +38,10 @@
 #define SNII_CRC32C_X86 1
 #include <cpuid.h>     // __get_cpuid, bit_SSE4_2
 #include <nmmintrin.h> // _mm_crc32_u8/u32/u64 (SSE4.2)
+#else
+// Keep the #if SNII_CRC32C_X86 uses below -Wundef-clean on non-x86 (aarch64
+// UT builds compile this BE_TEST-only TU with -Werror).
+#define SNII_CRC32C_X86 0
 #endif
 
 namespace doris::snii {
diff --git a/be/src/storage/olap_common.h b/be/src/storage/olap_common.h
index 55d84154e68b38..d8d275ca4f7437 100644
--- a/be/src/storage/olap_common.h
+++ b/be/src/storage/olap_common.h
@@ -472,16 +472,31 @@ struct MowContext {
 // used for controll compaction
 struct VersionWithTime {
     std::atomic<int64_t> version;
-    int64_t update_ts;
+    // Written by a single writer thread (heartbeat) and read lock-free by
+    // compaction selection; both must be atomic — on weakly-ordered aarch64
+    // plain stores could be observed reordered/torn.
+    //
+    // Invariant: update_ts is stored BEFORE the release CAS that publishes a
+    // new version. A reader that acquire-loads version and then update_ts is
+    // guaranteed a timestamp at least as new as the one stored for that
+    // version's publication; "new version + stale timestamp" never happens.
+    // The residual combination "old version + newer ts" is harmless: it only
+    // biases readers toward the conservative max retention count.
+    std::atomic<int64_t> update_ts;
 
     VersionWithTime() : version(0), update_ts(MonotonicMillis()) {}
 
     void update_version_monoto(int64_t new_version) {
         int64_t cur_version = version.load(std::memory_order_relaxed);
         while (cur_version < new_version) {
-            if (version.compare_exchange_strong(cur_version, new_version, std::memory_order_relaxed,
+            // Store the timestamp before the release CAS that publishes the
+            // version; relaxed suffices because the release/acquire chain on
+            // version orders this store ahead of any reader that observes the
+            // new version.
+            update_ts.store(MonotonicMillis(), std::memory_order_relaxed);
+            if (version.compare_exchange_strong(cur_version, new_version,
+                                                std::memory_order_release,
                                                 std::memory_order_relaxed)) {
-                update_ts = MonotonicMillis();
                 break;
             }
         }
diff --git a/be/src/storage/tablet/tablet.cpp b/be/src/storage/tablet/tablet.cpp
index 3aec7583b2e218..f4e0b1d011f525 100644
--- a/be/src/storage/tablet/tablet.cpp
+++ b/be/src/storage/tablet/tablet.cpp
@@ -1402,8 +1402,8 @@ std::tuple<int64_t, int64_t> Tablet::get_visible_version_and_time() const {
     // so let this tablet's visible version become int64 max.
     auto version_info = _visible_version.load();
     if (version_info != nullptr && partition_id() != 0) {
-        return std::make_tuple(version_info->version.load(std::memory_order_relaxed),
-                               version_info->update_ts);
+        return std::make_tuple(version_info->version.load(std::memory_order_acquire),
+                               version_info->update_ts.load(std::memory_order_acquire));
     } else {
         return std::make_tuple(std::numeric_limits<int64_t>::max(),
                                std::numeric_limits<int64_t>::max());
diff --git a/be/src/storage/types.h b/be/src/storage/types.h
index 7d66631ef16180..a6cd86906e19f9 100644
--- a/be/src/storage/types.h
+++ b/be/src/storage/types.h
@@ -36,6 +36,7 @@
 #include "core/decimal12.h"
 #include "core/extended_types.h"
 #include "core/packed_int128.h"
+#include "util/unaligned.h"
 #include "core/type_limit.h"
 #include "core/uint24.h"
 #include "core/value/ipv4_value.h"
@@ -238,7 +239,11 @@ struct BaseFieldTypeTraits : public CppTypeTraits<field_type> {
         if constexpr (field_type == FieldType::OLAP_FIELD_TYPE_LARGEINT) {
             return get_int128_from_unalign(address);
         }
-        return *reinterpret_cast<const CppType*>(address);
+        // Row-buffer fields are packed without padding, so 'address' is only
+        // byte-aligned for many (type, offset) combinations. Load through
+        // memcpy instead of a casted dereference (misaligned UB, e.g. an
+        // int64 field right after a tinyint in old row storage).
+        return unaligned_load<CppType>(address);
     }
 
     static inline void set_cpp_type_value(void* address, const CppType& value) {
diff --git a/be/src/util/bfd_parser.cpp b/be/src/util/bfd_parser.cpp
index 8213d468c2365e..b03b6ab50eafd3 100644
--- a/be/src/util/bfd_parser.cpp
+++ b/be/src/util/bfd_parser.cpp
@@ -94,8 +94,18 @@ void BfdParser::init_bfd() {
     }
     std::lock_guard<std::mutex> lock(_bfd_mutex);
     bfd_init();
-    if (!bfd_set_default_target("elf64-x86-64")) {
-        LOG(ERROR) << "set default target to elf64-x86-64 failed.";
+    // The default target must match the host architecture: hardcoding
+    // "elf64-x86-64" fails on aarch64 (libbfd there does not bundle the x86
+    // backend) and logs a spurious error on every BE start.
+#if defined(__aarch64__)
+    static constexpr const char* kDefaultTarget = "elf64-littleaarch64";
+#elif defined(__x86_64__)
+    static constexpr const char* kDefaultTarget = "elf64-x86-64";
+#else
+    static constexpr const char* kDefaultTarget = nullptr;
+#endif
+    if (kDefaultTarget != nullptr && !bfd_set_default_target(kDefaultTarget)) {
+        LOG(ERROR) << "set default target to " << kDefaultTarget << " failed.";
     }
     _is_bfd_inited = true;
 }
diff --git a/be/src/util/bit_packing.inline.h b/be/src/util/bit_packing.inline.h
index d0de57a836b4be..8ab704543632a7 100644
--- a/be/src/util/bit_packing.inline.h
+++ b/be/src/util/bit_packing.inline.h
@@ -23,6 +23,7 @@
 #if defined(__x86_64__) && (defined(__GNUC__) || defined(__clang__))
 #include "util/pdep_unpack.h"
 #endif
+#include "util/unaligned.h"
 
 namespace doris {
 inline int64_t BitPacking::NumValuesToUnpack(int bit_width, int64_t in_bytes, int64_t num_values) {
@@ -192,7 +193,7 @@ std::pair<const uint8_t*, int64_t> BitPacking::UnpackAndDecodeValues(
 // avoid buffer overflow (if we are unpacking 32 values, we can safely assume an input
 // buffer of length 32 * BIT_WIDTH).
 template <int BIT_WIDTH, int VALUE_IDX, bool FULL_BATCH>
-uint64_t NO_SANITIZE_UNDEFINED UnpackValue(const uint8_t* __restrict__ in_buf) {
+uint64_t UnpackValue(const uint8_t* __restrict__ in_buf) {
     if (BIT_WIDTH == 0) return 0;
 
     constexpr int FIRST_BIT_IDX = VALUE_IDX * BIT_WIDTH;
@@ -204,7 +205,11 @@ uint64_t NO_SANITIZE_UNDEFINED UnpackValue(const uint8_t* __restrict__ in_buf) {
 
     constexpr int FIRST_BIT_OFFSET = FIRST_BIT_IDX - FIRST_WORD_IDX * 32;
     constexpr uint64_t mask = GetMask(BIT_WIDTH);
-    const uint32_t* const in = reinterpret_cast<const uint32_t*>(in_buf);
+    // in_buf is an arbitrary byte pointer into a page buffer; word loads must
+    // go through unaligned_load (memcpy-based, compiles to the same single
+    // load instruction) instead of reinterpret_cast dereferences, which are
+    // misaligned UB whenever in_buf is not word-aligned.
+    const uint8_t* const in = in_buf;
 
     // Avoid reading past the end of the buffer. We can safely read 64 bits if we know that
     // this is a full batch read (so the input buffer is 32 * BIT_WIDTH long) and there is
@@ -222,17 +227,17 @@ uint64_t NO_SANITIZE_UNDEFINED UnpackValue(const uint8_t* __restrict__ in_buf) {
             WORDS_TO_READ == 1 && (!CAN_SAFELY_READ_64_BITS || BitUtil::IsPowerOf2(BIT_WIDTH));
 
     if (READ_32_BITS) {
-        uint32_t word = in[FIRST_WORD_IDX];
+        uint32_t word = unaligned_load<uint32_t>(in + 4 * FIRST_WORD_IDX);
         word >>= FIRST_BIT_OFFSET < 32 ? FIRST_BIT_OFFSET : 0;
         return word & mask;
     }
 
-    uint64_t word = *reinterpret_cast<const uint64_t*>(in + FIRST_WORD_IDX);
+    uint64_t word = unaligned_load<uint64_t>(in + 4 * FIRST_WORD_IDX);
     word >>= FIRST_BIT_OFFSET;
 
     if (WORDS_TO_READ > 2) {
         constexpr int USEFUL_BITS = FIRST_BIT_OFFSET == 0 ? 0 : 64 - FIRST_BIT_OFFSET;
-        uint64_t extra_word = in[FIRST_WORD_IDX + 2];
+        uint64_t extra_word = unaligned_load<uint32_t>(in + 4 * (FIRST_WORD_IDX + 2));
         word |= extra_word << USEFUL_BITS;
     }
 
diff --git a/be/src/util/bitmap_intersect.h b/be/src/util/bitmap_intersect.h
index 7e9d0308843338..881babb5ed793a 100644
--- a/be/src/util/bitmap_intersect.h
+++ b/be/src/util/bitmap_intersect.h
@@ -20,6 +20,7 @@
 #include "common/cast_set.h"
 #include "core/string_ref.h"
 #include "core/value/bitmap_value.h"
+#include "util/unaligned.h"
 
 namespace doris {
 
@@ -56,9 +57,11 @@ class Helper {
 
 template <>
 inline char* Helper::write_to<VecDateTimeValue>(const VecDateTimeValue& v, char* dest) {
-    *(int64_t*)dest = v.to_int64_datetime_packed();
+    // dest may be arbitrarily aligned (variable-length keys precede us), use
+    // memcpy-based unaligned stores instead of casted stores (UB).
+    unaligned_store<int64_t>(dest, v.to_int64_datetime_packed());
     dest += DATETIME_PACKED_TIME_BYTE_SIZE;
-    *(int*)dest = v.type();
+    unaligned_store<int>(dest, v.type());
     dest += DATETIME_TYPE_BYTE_SIZE;
     return dest;
 }
@@ -73,7 +76,7 @@ inline char* Helper::write_to<DecimalV2Value>(const DecimalV2Value& v, char* des
 
 template <>
 inline char* Helper::write_to<StringRef>(const StringRef& v, char* dest) {
-    *(int32_t*)dest = cast_set<int32_t>(v.size);
+    unaligned_store<int32_t>(dest, cast_set<int32_t>(v.size));
     dest += 4;
     memcpy(dest, v.data, v.size);
     dest += v.size;
@@ -82,7 +85,7 @@ inline char* Helper::write_to<StringRef>(const StringRef& v, char* dest) {
 
 template <>
 inline char* Helper::write_to<std::string>(const std::string& v, char* dest) {
-    *(uint32_t*)dest = cast_set<uint32_t>(v.size());
+    unaligned_store<uint32_t>(dest, cast_set<uint32_t>(v.size()));
     dest += 4;
     memcpy(dest, v.c_str(), v.size());
     dest += v.size();
@@ -113,9 +116,9 @@ inline int32_t Helper::serialize_size<std::string>(const std::string& v) {
 
 template <>
 inline void Helper::read_from<VecDateTimeValue>(const char** src, VecDateTimeValue* result) {
-    result->from_packed_time(*(int64_t*)(*src));
+    result->from_packed_time(unaligned_load<int64_t>(*src));
     *src += DATETIME_PACKED_TIME_BYTE_SIZE;
-    if (*(int*)(*src) == TIME_DATE) {
+    if (unaligned_load<int>(*src) == TIME_DATE) {
         result->cast_to_date();
     }
     *src += DATETIME_TYPE_BYTE_SIZE;
@@ -131,7 +134,7 @@ inline void Helper::read_from<DecimalV2Value>(const char** src, DecimalV2Value*
 
 template <>
 inline void Helper::read_from<StringRef>(const char** src, StringRef* result) {
-    int32_t length = *(int32_t*)(*src);
+    int32_t length = unaligned_load<int32_t>(*src);
     *src += 4;
     *result = StringRef((char*)*src, length);
     *src += length;
@@ -139,7 +142,7 @@ inline void Helper::read_from<StringRef>(const char** src, StringRef* result) {
 
 template <>
 inline void Helper::read_from<std::string>(const char** src, std::string* result) {
-    int32_t length = *(int32_t*)(*src);
+    int32_t length = unaligned_load<int32_t>(*src);
     *src += 4;
     *result = std::string((char*)*src, length);
     *src += length;
@@ -215,7 +218,7 @@ struct BitmapIntersect {
     //must call size() first
     void serialize(char* dest) {
         char* writer = dest;
-        *(int32_t*)writer = cast_set<int32_t>(_bitmaps.size());
+        unaligned_store<int32_t>(writer, cast_set<int32_t>(_bitmaps.size()));
         writer += 4;
         for (auto& kv : _bitmaps) {
             writer = detail::Helper::write_to(kv.first, writer);
@@ -226,7 +229,7 @@ struct BitmapIntersect {
 
     void deserialize(const char* src) {
         const char* reader = src;
-        int32_t bitmaps_size = *(int32_t*)reader;
+        int32_t bitmaps_size = unaligned_load<int32_t>(reader);
         reader += 4;
         for (int32_t i = 0; i < bitmaps_size; i++) {
             T key;
@@ -302,7 +305,7 @@ struct BitmapIntersect<std::string_view> {
     //must call size() first
     void serialize(char* dest) {
         char* writer = dest;
-        *(int32_t*)writer = cast_set<int32_t>(_bitmaps.size());
+        unaligned_store<int32_t>(writer, cast_set<int32_t>(_bitmaps.size()));
         writer += 4;
         for (auto& kv : _bitmaps) {
             writer = detail::Helper::write_to(kv.first, writer);
@@ -313,7 +316,7 @@ struct BitmapIntersect<std::string_view> {
 
     void deserialize(const char* src) {
         const char* reader = src;
-        int32_t bitmaps_size = *(int32_t*)reader;
+        int32_t bitmaps_size = unaligned_load<int32_t>(reader);
         reader += 4;
         for (int32_t i = 0; i < bitmaps_size; i++) {
             std::string key;
diff --git a/be/src/util/hash_util.hpp b/be/src/util/hash_util.hpp
index 49c01e4a176a26..4c00a95e63f5ed 100644
--- a/be/src/util/hash_util.hpp
+++ b/be/src/util/hash_util.hpp
@@ -35,6 +35,7 @@
 #include "util/hash/city.h"
 #include "util/hash/murmur_hash3.h"
 #include "util/sse_util.hpp"
+#include "util/unaligned.h"
 
 namespace doris {
 namespace detail {
@@ -161,14 +162,16 @@ class HashUtil {
         uint32_t words = bytes / sizeof(uint32_t);
         bytes = bytes % sizeof(uint32_t);
 
-        const uint32_t* p = reinterpret_cast<const uint32_t*>(data);
+        // 'data' is an arbitrary caller buffer; word loads must tolerate
+        // misalignment (crc intrinsics consume values, so load via memcpy).
+        const uint8_t* p = static_cast<const uint8_t*>(data);
 
         while (words--) {
-            hash = _mm_crc32_u32(hash, *p);
-            ++p;
+            hash = _mm_crc32_u32(hash, unaligned_load<uint32_t>(p));
+            p += sizeof(uint32_t);
         }
 
-        const uint8_t* s = reinterpret_cast<const uint8_t*>(p);
+        const uint8_t* s = p;
 
         while (bytes--) {
             hash = _mm_crc32_u8(hash, *s);
@@ -188,13 +191,14 @@ class HashUtil {
         uint32_t h1 = hash >> 32;
         uint32_t h2 = (hash << 32) >> 32;
 
-        const uint32_t* p = reinterpret_cast<const uint32_t*>(data);
+        const uint8_t* p = static_cast<const uint8_t*>(data);
         while (words--) {
-            (words & 1) ? (h1 = _mm_crc32_u32(h1, *p)) : (h2 = _mm_crc32_u32(h2, *p));
-            ++p;
+            (words & 1) ? (h1 = _mm_crc32_u32(h1, unaligned_load<uint32_t>(p)))
+                        : (h2 = _mm_crc32_u32(h2, unaligned_load<uint32_t>(p)));
+            p += sizeof(uint32_t);
         }
 
-        const uint8_t* s = reinterpret_cast<const uint8_t*>(p);
+        const uint8_t* s = p;
         while (bytes--) {
             (bytes & 1) ? (h1 = _mm_crc32_u8(h1, *s)) : (h2 = _mm_crc32_u8(h2, *s));
             ++s;
@@ -240,11 +244,15 @@ class HashUtil {
     static uint64_t murmur_hash2_64(const void* input, int len, uint64_t seed) {
         uint64_t h = seed ^ (len * MURMUR_PRIME);
 
-        const uint64_t* data = reinterpret_cast<const uint64_t*>(input);
-        const uint64_t* end = data + (len / sizeof(uint64_t));
+        // 'input' is an arbitrary caller buffer and is not guaranteed to be
+        // 8-byte aligned; load words via memcpy (unaligned_load) instead of
+        // casted dereferences, which are misaligned UB.
+        const uint8_t* data = static_cast<const uint8_t*>(input);
+        const uint8_t* end = data + (len / sizeof(uint64_t)) * sizeof(uint64_t);
 
         while (data != end) {
-            uint64_t k = *data++;
+            uint64_t k = unaligned_load<uint64_t>(data);
+            data += sizeof(uint64_t);
             k *= MURMUR_PRIME;
             k ^= k >> MURMUR_R;
             k *= MURMUR_PRIME;
@@ -252,7 +260,7 @@ class HashUtil {
             h *= MURMUR_PRIME;
         }
 
-        const uint8_t* data2 = reinterpret_cast<const uint8_t*>(data);
+        const uint8_t* data2 = data;
         switch (len & 7) {
         case 7:
             h ^= uint64_t(data2[6]) << 48;
diff --git a/be/src/util/histogram.cpp b/be/src/util/histogram.cpp
index 8cdcc42b7d991a..19e4749151ab18 100644
--- a/be/src/util/histogram.cpp
+++ b/be/src/util/histogram.cpp
@@ -25,6 +25,8 @@
 #include <limits>
 #include <utility>
 
+#include "util/sse_util.hpp"
+
 namespace doris {
 
 HistogramBucketMapper::HistogramBucketMapper() {
@@ -121,12 +123,16 @@ void HistogramStat::merge(const HistogramStat& other) {
     // requires no lock and value update can still happen concurrently
     uint64_t old_min = min();
     uint64_t other_min = other.min();
+    // Pause on CAS failure: a bare retry loop floods the interconnect with
+    // LL/SC attempts on many-core aarch64 machines.
     while (other_min < old_min && !_min.compare_exchange_weak(old_min, other_min)) {
+        _mm_pause();
     }
 
     uint64_t old_max = max();
     uint64_t other_max = other.max();
     while (other_max > old_max && !_max.compare_exchange_weak(old_max, other_max)) {
+        _mm_pause();
     }
 
     _num.fetch_add(other.num(), std::memory_order_relaxed);
diff --git a/be/test/common/signal_handler_test.cpp b/be/test/common/signal_handler_test.cpp
new file mode 100644
index 00000000000000..6492442351305e
--- /dev/null
+++ b/be/test/common/signal_handler_test.cpp
@@ -0,0 +1,85 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gen_cpp/Types_types.h> // TUniqueId, normally pulled in via pch.h
+#include <gen_cpp/types.pb.h>  // PUniqueId, normally pulled in via pch.h
+#include <iostream>
+
+#include "common/signal_handler.h"
+
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <thread>
+#include <vector>
+
+namespace doris {
+
+// T1: On aarch64, HAVE___SYNC_VAL_COMPARE_AND_SWAP is never defined and the
+// x86 inline-asm branch is compiled out, so sync_val_compare_and_swap falls
+// back to a non-atomic read-check-write implementation. FailureSignalHandler
+// uses it to elect the single thread that dumps crash info. On weakly-ordered
+// ARM with many cores, multiple crashing threads can all "win" the election
+// and dump concurrently, corrupting the crash log or crashing again inside
+// the handler. This test reproduces the race: exactly one thread must win.
+TEST(SignalHandlerTest, CasRaceSingleWinner) {
+    constexpr int kThreads = 128;
+    for (int round = 0; round < 5; ++round) {
+        pthread_t* entered = nullptr;
+        std::vector<pthread_t> ids(kThreads);
+        std::atomic<int> ready {0};
+        std::atomic<bool> go {false};
+        std::atomic<int> winners {0};
+        std::vector<std::thread> threads;
+        threads.reserve(kThreads);
+        for (int i = 0; i < kThreads; ++i) {
+            threads.emplace_back([&, i] {
+                ids[i] = pthread_self();
+                ready.fetch_add(1, std::memory_order_relaxed);
+                while (!go.load(std::memory_order_acquire)) {
+                }
+                pthread_t* old = signal::sync_val_compare_and_swap(
+                        &entered, static_cast<pthread_t*>(nullptr), &ids[i]);
+                if (old == nullptr) {
+                    winners.fetch_add(1, std::memory_order_relaxed);
+                }
+            });
+        }
+        while (ready.load(std::memory_order_acquire) != kThreads) {
+        }
+        go.store(true, std::memory_order_release);
+        for (auto& t : threads) {
+            t.join();
+        }
+        EXPECT_EQ(1, winners.load())
+                << "round " << round << ": " << winners.load()
+                << " threads won the FailureSignalHandler election, expected exactly 1";
+    }
+}
+
+// Basic single-threaded CAS semantics must hold on every platform.
+TEST(SignalHandlerTest, CasBasicSemantics) {
+    int value = 1;
+    // oldval matches: swap happens, old value returned
+    EXPECT_EQ(1, signal::sync_val_compare_and_swap(&value, 1, 2));
+    EXPECT_EQ(2, value);
+    // oldval does not match: no swap, current value returned
+    EXPECT_EQ(2, signal::sync_val_compare_and_swap(&value, 1, 3));
+    EXPECT_EQ(2, value);
+}
+
+} // namespace doris
diff --git a/be/test/core/column/columns_common_test.cpp b/be/test/core/column/columns_common_test.cpp
new file mode 100644
index 00000000000000..2c8c99e185f64e
--- /dev/null
+++ b/be/test/core/column/columns_common_test.cpp
@@ -0,0 +1,72 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <cstdint>
+#include <random>
+
+#include "core/column/columns_common.h"
+
+namespace doris {
+
+// H3: columns_common.cpp had `#if defined(__SSE2__) || defined(__aarch64__)
+// && defined(__POPCNT__)` — && binds tighter than ||, and ARM toolchains
+// never define __POPCNT__, so the SIMD block was silently compiled out on
+// aarch64. These tests pin the semantics — the function deliberately
+// compares SIGNED bytes (`> 0`, see the NOTE in the implementation), so
+// bytes 128..255 (negative int8) must NOT be counted — and the scalar and
+// SIMD paths must always agree.
+namespace {
+size_t ref_count(const IColumn::Filter& filt) {
+    size_t n = 0;
+    for (auto v : filt) {
+        n += static_cast<int8_t>(v) > 0;
+    }
+    return n;
+}
+} // namespace
+
+TEST(ColumnsCommonTest, CountBytesInFilterMatchesReference) {
+    std::mt19937 rng(20260815);
+    for (size_t size : {0, 1, 7, 63, 64, 65, 127, 128, 129, 1000, 4096, 65537}) {
+        IColumn::Filter filt(size);
+        for (auto& v : filt) {
+            // values beyond {0,1} on purpose: covers positive and negative
+            // int8 alike; only strictly-positive int8 may be counted
+            v = static_cast<uint8_t>(rng() % 4 == 0 ? 0 : (rng() % 255 + 1));
+        }
+        ASSERT_EQ(ref_count(filt), count_bytes_in_filter(filt)) << "size=" << size;
+    }
+}
+
+TEST(ColumnsCommonTest, CountBytesInFilterEdgePatterns) {
+    IColumn::Filter all_zero(512, 0);
+    EXPECT_EQ(0, count_bytes_in_filter(all_zero));
+    IColumn::Filter all_one(512, 1);
+    EXPECT_EQ(512, count_bytes_in_filter(all_one));
+    // 127 is the largest positive int8: counted.
+    IColumn::Filter all_max_positive(513, 127);
+    EXPECT_EQ(513, count_bytes_in_filter(all_max_positive));
+    // 128..255 are negative int8 and must NOT be counted (signed compare).
+    IColumn::Filter all_negative(511, 255);
+    EXPECT_EQ(0, count_bytes_in_filter(all_negative));
+    IColumn::Filter all_msb_set(64, 128);
+    EXPECT_EQ(0, count_bytes_in_filter(all_msb_set));
+}
+
+} // namespace doris
diff --git a/be/test/storage/index/bloom_filter/ngram_bloom_filter_test.cpp b/be/test/storage/index/bloom_filter/ngram_bloom_filter_test.cpp
new file mode 100644
index 00000000000000..63835c86176a66
--- /dev/null
+++ b/be/test/storage/index/bloom_filter/ngram_bloom_filter_test.cpp
@@ -0,0 +1,113 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "storage/index/bloom_filter/ngram_bloom_filter.h"
+
+#include <gen_cpp/segment_v2.pb.h>
+#include <gtest/gtest.h>
+
+#include <cstring>
+#include <vector>
+
+namespace doris::segment_v2 {
+
+// H1: NGramBloomFilter::init() reads the on-disk bitset through a
+// reinterpret_cast<const uint64_t*> of an arbitrarily-offset page buffer.
+// Feed it a deliberately misaligned buffer (offset by 1 byte) and verify the
+// filter still round-trips correctly. This guards the memcpy-based hardening
+// and is UBSan-ready.
+TEST(NGramBloomFilterTest, InitFromUnalignedBuffer) {
+    constexpr size_t kBfSize = 1024; // bytes
+    NGramBloomFilter writer(kBfSize);
+    const char* added[] = {"hello", "doris", "arm64", "bloom-filter"};
+    for (const char* s : added) {
+        writer.add_bytes(s, strlen(s));
+    }
+
+    // Copy the serialized bitset into a buffer shifted by 1 byte from an
+    // 8-aligned base (vector<uint8_t> storage is not guaranteed aligned).
+    std::vector<uint8_t> raw(kBfSize + 16, 0);
+    uint8_t* aligned_base =
+            reinterpret_cast<uint8_t*>((reinterpret_cast<uintptr_t>(raw.data()) + 7) & ~7ULL);
+    memcpy(aligned_base + 1, writer.data(), kBfSize);
+    const char* unaligned_buf = reinterpret_cast<const char*>(aligned_base + 1);
+    ASSERT_EQ(1, static_cast<int>(reinterpret_cast<uintptr_t>(unaligned_buf) % alignof(uint64_t)));
+
+    NGramBloomFilter reader(kBfSize);
+    ASSERT_TRUE(reader.init(unaligned_buf, kBfSize, CITY_HASH_64).ok());
+
+    // Every added entry must be found in the filter read back from the
+    // unaligned buffer.
+    for (const char* s : added) {
+        NGramBloomFilter query(kBfSize);
+        query.add_bytes(s, strlen(s));
+        EXPECT_TRUE(reader.contains(query)) << "missing added entry: " << s;
+    }
+    // A never-added entry should (with overwhelming probability) be absent.
+    NGramBloomFilter absent(kBfSize);
+    const char* not_added = "definitely-not-added-string";
+    absent.add_bytes(not_added, strlen(not_added));
+    EXPECT_FALSE(reader.contains(absent));
+}
+
+// Content read back via init() must be identical to the source filter.
+TEST(NGramBloomFilterTest, InitPreservesContent) {
+    constexpr size_t kBfSize = 512;
+    NGramBloomFilter writer(kBfSize);
+    for (int i = 0; i < 100; ++i) {
+        std::string s = "key_" + std::to_string(i);
+        writer.add_bytes(s.data(), s.size());
+    }
+    NGramBloomFilter reader(kBfSize);
+    ASSERT_TRUE(reader.init(writer.data(), kBfSize, CITY_HASH_64).ok());
+    EXPECT_EQ(0, memcmp(writer.data(), reader.data(), kBfSize));
+}
+
+// ASan guard for the rounded-up tail over-read in init(): for bf sizes not
+// divisible by sizeof(UnderType), init() must copy exactly size bytes from
+// the input. The source vector is allocated with NO slack, so any read past
+// size bytes lands in an ASan redzone.
+TEST(NGramBloomFilterTest, InitFromExactSizeNonMultipleOfEight) {
+    const char* added[] = {"hello", "doris", "arm64", "bloom-filter"};
+    for (size_t bf_size : {65, 67, 100, 511, 65535}) {
+        NGramBloomFilter writer(bf_size);
+        for (const char* s : added) {
+            writer.add_bytes(s, strlen(s));
+        }
+
+        // Exactly bf_size bytes, no padding.
+        std::vector<uint8_t> exact(bf_size);
+        memcpy(exact.data(), writer.data(), bf_size);
+
+        NGramBloomFilter reader(bf_size);
+        ASSERT_TRUE(reader.init(reinterpret_cast<const char*>(exact.data()), bf_size, CITY_HASH_64)
+                            .ok())
+                << "bf_size=" << bf_size;
+
+        // The deserialized filter must contain the writer filter itself;
+        // this only holds if the tail bytes of the last word stayed zero.
+        EXPECT_TRUE(reader.contains(writer)) << "bf_size=" << bf_size;
+        // Every added entry must be found, queried the same way as above.
+        for (const char* s : added) {
+            NGramBloomFilter query(bf_size);
+            query.add_bytes(s, strlen(s));
+            EXPECT_TRUE(reader.contains(query)) << "bf_size=" << bf_size << " missing: " << s;
+        }
+    }
+}
+
+} // namespace doris::segment_v2
diff --git a/be/test/storage/version_with_time_test.cpp b/be/test/storage/version_with_time_test.cpp
new file mode 100644
index 00000000000000..61acda4d578ce8
--- /dev/null
+++ b/be/test/storage/version_with_time_test.cpp
@@ -0,0 +1,103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "storage/olap_common.h"
+
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <cstdint>
+#include <thread>
+#include <vector>
+
+#include "util/time.h"
+
+namespace doris {
+
+// Regression test for the (version, update_ts) publication order in
+// VersionWithTime::update_version_monoto. update_ts must be stored before the
+// release CAS that publishes the new version; otherwise a reader that
+// acquire-loads version and then update_ts can observe "new version + stale
+// timestamp", skewing MonotonicMillis()-update_ts upward and wrongly selecting
+// compaction_keep_invisible_version_min_count. With the correct order the
+// release/acquire chain on version guarantees ts >= the timestamp stored for
+// that version's publication, so this test passes deterministically.
+TEST(VersionWithTimeTest, ReadersNeverSeeStaleTimestamp) {
+    constexpr int64_t kMaxVersion = 20000;
+    constexpr int kReaders = 4;
+    VersionWithTime vwt;
+    // lower_bound[v]: sampled right before publishing v, so the timestamp
+    // stored for v's publication is guaranteed to be >= lower_bound[v].
+    std::vector<std::atomic<int64_t>> lower_bound(kMaxVersion + 1);
+    for (auto& bound : lower_bound) {
+        bound.store(0, std::memory_order_relaxed);
+    }
+
+    std::atomic<bool> stop {false};
+    std::atomic<int64_t> violations {0};
+    std::atomic<int64_t> first_offending_version {0};
+
+    auto reader = [&] {
+        while (!stop.load(std::memory_order_relaxed)) {
+            int64_t v = vwt.version.load(std::memory_order_acquire);
+            int64_t ts = vwt.update_ts.load(std::memory_order_acquire);
+            if (v > 0 && ts < lower_bound[v].load(std::memory_order_acquire)) {
+                violations.fetch_add(1, std::memory_order_relaxed);
+                int64_t expected = 0;
+                first_offending_version.compare_exchange_strong(expected, v,
+                                                                std::memory_order_relaxed);
+            }
+        }
+    };
+
+    std::vector<std::thread> readers;
+    readers.reserve(kReaders);
+    for (int i = 0; i < kReaders; ++i) {
+        readers.emplace_back(reader);
+    }
+
+    for (int64_t v = 1; v <= kMaxVersion; ++v) {
+        lower_bound[v].store(MonotonicMillis(), std::memory_order_release);
+        vwt.update_version_monoto(v);
+    }
+
+    stop.store(true, std::memory_order_relaxed);
+    for (auto& t : readers) {
+        t.join();
+    }
+
+    EXPECT_EQ(violations.load(std::memory_order_relaxed), 0)
+            << "first offending version: "
+            << first_offending_version.load(std::memory_order_relaxed);
+}
+
+TEST(VersionWithTimeTest, MonotoneVersionRejectsOlder) {
+    VersionWithTime vwt;
+    vwt.update_version_monoto(10);
+    ASSERT_EQ(vwt.version.load(std::memory_order_relaxed), 10);
+    int64_t ts = vwt.update_ts.load(std::memory_order_relaxed);
+
+    // Older or equal versions must be a no-op: version and ts stay unchanged.
+    vwt.update_version_monoto(5);
+    EXPECT_EQ(vwt.version.load(std::memory_order_relaxed), 10);
+    EXPECT_EQ(vwt.update_ts.load(std::memory_order_relaxed), ts);
+    vwt.update_version_monoto(10);
+    EXPECT_EQ(vwt.version.load(std::memory_order_relaxed), 10);
+    EXPECT_EQ(vwt.update_ts.load(std::memory_order_relaxed), ts);
+}
+
+} // namespace doris
diff --git a/be/test/util/bit_packing_unaligned_test.cpp b/be/test/util/bit_packing_unaligned_test.cpp
new file mode 100644
index 00000000000000..4f58e45b4418b7
--- /dev/null
+++ b/be/test/util/bit_packing_unaligned_test.cpp
@@ -0,0 +1,88 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <cstdint>
+#include <cstring>
+#include <random>
+#include <vector>
+
+#include "util/bit_packing.h"
+#include "util/bit_packing.inline.h"
+
+namespace doris {
+
+// H2: UnpackValue() in bit_packing.inline.h loads 32/64-bit words through
+// reinterpret_cast of an arbitrary byte pointer. Feed UnpackValues buffers at
+// every possible misalignment (offset 1..7) and check decoded values against
+// a straightforward reference implementation. Guards the unaligned_load
+// hardening; also directly exercisable under -fsanitize=alignment.
+namespace {
+
+// Reference bit unpacker: reads the stream one bit at a time (LSB-first
+// within each byte, matching the layout documented in bit_packing.h).
+uint64_t ref_unpack(const uint8_t* in, int bit_width, int64_t value_idx) {
+    uint64_t result = 0;
+    int64_t first_bit = value_idx * bit_width;
+    for (int b = 0; b < bit_width; ++b) {
+        int64_t bit = first_bit + b;
+        uint64_t v = (in[bit / 8] >> (bit % 8)) & 1;
+        result |= v << b;
+    }
+    return result;
+}
+
+void run_unaligned_case(int bit_width, int offset, std::mt19937_64* rng) {
+    constexpr int64_t kNumValues = 64;
+    const int64_t in_bytes = (bit_width * kNumValues + 7) / 8;
+    std::vector<uint8_t> backing(in_bytes + 16, 0);
+    // vector<uint8_t> storage is not guaranteed 8-aligned; find an aligned
+    // base inside it first so that 'in' lands exactly at 'offset' mod 8.
+    uint8_t* aligned_base =
+            reinterpret_cast<uint8_t*>((reinterpret_cast<uintptr_t>(backing.data()) + 7) & ~7ULL);
+    for (int i = 0; i < in_bytes; ++i) {
+        aligned_base[offset + i] = static_cast<uint8_t>((*rng)());
+    }
+    const uint8_t* in = aligned_base + offset;
+    ASSERT_EQ(offset, static_cast<int>(reinterpret_cast<uintptr_t>(in) % 8));
+
+    std::vector<uint64_t> out(kNumValues, 0);
+    auto [end, read] =
+            BitPacking::UnpackValues<uint64_t>(bit_width, in, in_bytes, kNumValues, out.data());
+    ASSERT_EQ(kNumValues, read);
+    for (int64_t i = 0; i < kNumValues; ++i) {
+        uint64_t expected = ref_unpack(in, bit_width, i);
+        EXPECT_EQ(expected, out[i]) << "bit_width=" << bit_width << " offset=" << offset
+                                    << " value_idx=" << i;
+    }
+}
+
+} // namespace
+
+TEST(BitPackingUnalignedTest, UnpackFromMisalignedBuffers) {
+    std::mt19937_64 rng(20260815);
+    // Widths that exercise the 32-bit path, the 64-bit path and the
+    // three-word path (e.g. width 63 spans words) in UnpackValue.
+    for (int bit_width : {1, 3, 7, 8, 12, 16, 21, 31, 32, 33, 48, 63, 64}) {
+        for (int offset = 1; offset < 8; ++offset) {
+            run_unaligned_case(bit_width, offset, &rng);
+        }
+    }
+}
+
+} // namespace doris
diff --git a/be/test/util/bitmap_intersect_test.cpp b/be/test/util/bitmap_intersect_test.cpp
new file mode 100644
index 00000000000000..44accab0f54765
--- /dev/null
+++ b/be/test/util/bitmap_intersect_test.cpp
@@ -0,0 +1,146 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "util/bitmap_intersect.h"
+
+#include <gtest/gtest.h>
+
+#include <cstdint>
+#include <cstring>
+#include <string>
+#include <vector>
+
+namespace doris {
+
+// H2: BitmapIntersect serialization places a 4-byte length + variable-length
+// key bytes back to back; after an odd-length string key the following reads
+// (*(int32_t*) / BitmapValue header) land on odd addresses. aarch64 scalar
+// loads tolerate this, but it is C++ UB. This roundtrip test with odd-length
+// keys guards the memcpy-based hardening and is UBSan-ready.
+TEST(BitmapIntersectTest, RoundtripWithOddLengthKeys) {
+    BitmapIntersect<std::string> writer;
+    // Odd-length keys make every following field misaligned.
+    std::vector<std::string> keys = {"a", "bbb", "ccccc", "x", "odd_key_9"};
+    BitmapValue bv1;
+    for (uint32_t i = 0; i < 100; i += 2) {
+        bv1.add(i);
+    }
+    BitmapValue bv2;
+    for (uint32_t i = 0; i < 100; i += 3) {
+        bv2.add(i);
+    }
+    for (size_t i = 0; i < keys.size(); ++i) {
+        writer.add_key(keys[i]);
+        writer.update(keys[i], i % 2 == 0 ? bv1 : bv2);
+    }
+
+    const size_t ser_size = writer.size();
+    // Serialize into a buffer shifted by 1 byte from an 8-aligned base so the
+    // whole stream is misaligned, then also misalign the read side.
+    std::vector<char> raw(ser_size + 16, 0);
+    char* aligned_base =
+            reinterpret_cast<char*>((reinterpret_cast<uintptr_t>(raw.data()) + 7) & ~7ULL);
+    char* dest = aligned_base + 1;
+    writer.serialize(dest);
+
+    BitmapIntersect<std::string> reader(dest);
+    EXPECT_EQ(writer.intersect_count(), reader.intersect_count());
+
+    // Expected intersection: bv1 (even) keys & bv2 (odd) keys.
+    // keys with even index hold bv1, odd index hold bv2.
+    BitmapValue expect = bv1;
+    expect &= bv2;
+    EXPECT_EQ(expect.cardinality(), reader.intersect_count());
+}
+
+// int32 keys: with the 4-byte count prefix the first key starts at offset 4
+// (only 4-aligned), and keys are read back through fixed-width
+// unaligned_load<int32_t>. Guards the count-prefix + fixed-width key reads at
+// only-4-aligned offsets.
+TEST(BitmapIntersectTest, RoundtripIntKeysMisaligned) {
+    BitmapIntersect<int32_t> writer;
+    BitmapValue bv;
+    for (uint32_t i = 1; i <= 10; ++i) {
+        bv.add(i * 7);
+    }
+    for (int32_t k = -5; k <= 5; ++k) {
+        writer.add_key(k);
+        writer.update(k, bv);
+    }
+    const size_t ser_size = writer.size();
+    std::vector<char> raw(ser_size + 16, 0);
+    char* aligned_base =
+            reinterpret_cast<char*>((reinterpret_cast<uintptr_t>(raw.data()) + 7) & ~7ULL);
+    char* dest = aligned_base + 3; // 8k+3 offset: every int32 read misaligned
+    writer.serialize(dest);
+
+    BitmapIntersect<int32_t> reader(dest);
+    EXPECT_EQ(bv.cardinality(), reader.intersect_count());
+}
+
+// VecDateTimeValue keys: write_to stores packed int64 + 4-byte type tag and
+// read_from does two sequential word reads; with the 4-byte count prefix the
+// first key's packed int64 lands at offset 4 (4-aligned but not 8-aligned).
+// Guards Helper::write_to/read_from<VecDateTimeValue> hardening.
+TEST(BitmapIntersectTest, RoundtripDateTimeKeysMisaligned) {
+    BitmapIntersect<VecDateTimeValue> writer;
+    auto make_datetime = [](int64_t v) {
+        VecDateTimeValue d;
+        d.from_date_int64(v);
+        return d;
+    };
+    // Include a TIME_DATE key to cover the cast_to_date branch of read_from.
+    VecDateTimeValue date_key = make_datetime(20240404);
+    date_key.cast_to_date();
+    std::vector<VecDateTimeValue> keys = {make_datetime(20240101123000),
+                                          make_datetime(20240202123000),
+                                          make_datetime(20240303123000), date_key};
+    BitmapValue bv1;
+    for (uint32_t i = 0; i < 100; i += 2) {
+        bv1.add(i);
+    }
+    BitmapValue bv2;
+    for (uint32_t i = 0; i < 100; i += 3) {
+        bv2.add(i);
+    }
+    for (size_t i = 0; i < keys.size(); ++i) {
+        writer.add_key(keys[i]);
+        writer.update(keys[i], i % 2 == 0 ? bv1 : bv2);
+    }
+
+    const size_t ser_size = writer.size();
+    std::vector<char> raw(ser_size + 16, 0);
+    char* aligned_base =
+            reinterpret_cast<char*>((reinterpret_cast<uintptr_t>(raw.data()) + 7) & ~7ULL);
+    char* dest = aligned_base + 3; // 8k+3 offset: every packed int64 read misaligned
+    writer.serialize(dest);
+
+    BitmapIntersect<VecDateTimeValue> reader(dest);
+
+    // Expected intersection: even-index keys hold bv1, odd-index keys hold bv2.
+    BitmapValue expect = bv1;
+    expect &= bv2;
+    BitmapValue got = reader.intersect();
+    EXPECT_EQ(expect.cardinality(), got.cardinality());
+    for (uint32_t i = 0; i < 100; ++i) {
+        if (i % 6 == 0) { // present in both bv1 and bv2
+            EXPECT_TRUE(got.contains(i));
+        }
+    }
+}
+
+} // namespace doris
diff --git a/be/test/util/hash_util_unaligned_test.cpp b/be/test/util/hash_util_unaligned_test.cpp
new file mode 100644
index 00000000000000..11facd01d2e7cb
--- /dev/null
+++ b/be/test/util/hash_util_unaligned_test.cpp
@@ -0,0 +1,106 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <cstdint>
+#include <cstring>
+#include <random>
+#include <vector>
+
+#include "util/hash_util.hpp"
+
+namespace doris {
+
+// H2: HashUtil::murmur_hash2_64 loads uint64_t words through
+// reinterpret_cast of the caller buffer. Hashing the same content at every
+// byte misalignment must yield the reference value (computed from an aligned
+// copy). Guards the unaligned_load hardening; UBSan-ready.
+TEST(HashUtilUnalignedTest, MurmurHash2FromMisalignedBuffers) {
+    std::mt19937_64 rng(20260815);
+    for (int len : {1, 7, 8, 9, 15, 16, 17, 64, 100, 1000}) {
+        std::vector<uint8_t> backing(len + 16);
+        uint8_t* aligned_base =
+                reinterpret_cast<uint8_t*>((reinterpret_cast<uintptr_t>(backing.data()) + 7) &
+                                           ~7ULL);
+        std::vector<uint8_t> content(len);
+        for (auto& b : content) {
+            b = static_cast<uint8_t>(rng());
+        }
+        const uint64_t seed = 0x9e3779b97f4a7c15ULL;
+        // reference on aligned copy
+        memcpy(aligned_base, content.data(), len);
+        uint64_t ref = HashUtil::murmur_hash2_64(aligned_base, len, seed);
+        for (int offset = 1; offset < 8; ++offset) {
+            memcpy(aligned_base + offset, content.data(), len);
+            uint64_t got = HashUtil::murmur_hash2_64(aligned_base + offset, len, seed);
+            EXPECT_EQ(ref, got) << "len=" << len << " offset=" << offset;
+        }
+    }
+}
+
+// Same guard for HashUtil::crc_hash, which feeds word loads to _mm_crc32_u32
+// (sse2neon on aarch64). Reference is computed on the aligned copy.
+TEST(HashUtilUnalignedTest, CrcHashFromMisalignedBuffers) {
+    std::mt19937_64 rng(20260815);
+    for (int len : {1, 7, 8, 9, 15, 16, 17, 64, 100, 1000}) {
+        std::vector<uint8_t> backing(len + 16);
+        uint8_t* aligned_base =
+                reinterpret_cast<uint8_t*>((reinterpret_cast<uintptr_t>(backing.data()) + 7) &
+                                           ~7ULL);
+        std::vector<uint8_t> content(len);
+        for (auto& b : content) {
+            b = static_cast<uint8_t>(rng());
+        }
+        const uint32_t seed = 0x9e3779b9U;
+        // reference on aligned copy
+        memcpy(aligned_base, content.data(), len);
+        uint32_t ref = HashUtil::crc_hash(aligned_base, len, seed);
+        for (int offset = 1; offset < 8; ++offset) {
+            memcpy(aligned_base + offset, content.data(), len);
+            uint32_t got = HashUtil::crc_hash(aligned_base + offset, len, seed);
+            EXPECT_EQ(ref, got) << "len=" << len << " offset=" << offset;
+        }
+    }
+}
+
+// Same guard for HashUtil::crc_hash64, the 64-bit variant built on the same
+// _mm_crc32_* word loads.
+TEST(HashUtilUnalignedTest, CrcHash64FromMisalignedBuffers) {
+    std::mt19937_64 rng(20260815);
+    for (int len : {1, 7, 8, 9, 15, 16, 17, 64, 100, 1000}) {
+        std::vector<uint8_t> backing(len + 16);
+        uint8_t* aligned_base =
+                reinterpret_cast<uint8_t*>((reinterpret_cast<uintptr_t>(backing.data()) + 7) &
+                                           ~7ULL);
+        std::vector<uint8_t> content(len);
+        for (auto& b : content) {
+            b = static_cast<uint8_t>(rng());
+        }
+        const uint64_t seed = 0x9e3779b97f4a7c15ULL;
+        // reference on aligned copy
+        memcpy(aligned_base, content.data(), len);
+        uint64_t ref = HashUtil::crc_hash64(aligned_base, len, seed);
+        for (int offset = 1; offset < 8; ++offset) {
+            memcpy(aligned_base + offset, content.data(), len);
+            uint64_t got = HashUtil::crc_hash64(aligned_base + offset, len, seed);
+            EXPECT_EQ(ref, got) << "len=" << len << " offset=" << offset;
+        }
+    }
+}
+
+} // namespace doris
diff --git a/thirdparty/build-thirdparty.sh b/thirdparty/build-thirdparty.sh
index df4bfe6d1fa173..43d0365bc7e921 100755
--- a/thirdparty/build-thirdparty.sh
+++ b/thirdparty/build-thirdparty.sh
@@ -1404,10 +1404,22 @@ build_bitshuffle() {
 # croaring bitmap
 build_croaringbitmap() {
     avx_flag=''
-    if [[ -n "${USE_AVX2}" && "${USE_AVX2}" -eq 0 ]]; then
+    # USE_AVX2 accepts common CMake boolean spellings, case-insensitively:
+    # 0/OFF/FALSE/NO disable AVX2, 1/ON/TRUE/YES or empty/unset keep it enabled.
+    # build.sh defaults USE_AVX2=ON and forwards it verbatim to BE's CMake, so
+    # croaring must agree with BE on whether AVX2 is enabled. String matching is
+    # used instead of `-eq`: in bash arithmetic context the string "ON"
+    # evaluates to 0, so exporting USE_AVX2=ON would silently DISABLE AVX2.
+    case "${USE_AVX2}" in
+    0 | [oO][fF][fF] | [fF][aA][lL][sS][eE] | [nN][oO])
         echo "set USE_AVX2=${USE_AVX2} to FORCE disable AVX2 in croaringbitmap"
         avx_flag="-DROARING_DISABLE_AVX=ON"
-    fi
+        ;;
+    1 | [oO][nN] | [tT][rR][uU][eE] | [yY][eE][sS] | '') ;;
+    *)
+        echo "WARNING: unrecognized USE_AVX2=${USE_AVX2}, AVX2 is left enabled in croaringbitmap"
+        ;;
+    esac
 
     check_if_source_exist "${CROARINGBITMAP_SOURCE}"
     cd "${TP_SOURCE_DIR}/${CROARINGBITMAP_SOURCE}"