diff --git a/be/benchmark/benchmark_hybrid_set.hpp b/be/benchmark/benchmark_hybrid_set.hpp
index fee58c56013521..a407ef1e856c44 100644
--- a/be/benchmark/benchmark_hybrid_set.hpp
+++ b/be/benchmark/benchmark_hybrid_set.hpp
@@ -26,11 +26,22 @@
 
 #include <benchmark/benchmark.h>
 
+#include <algorithm>
 #include <cstdint>
+#include <limits>
+#include <memory>
 #include <string>
+#include <utility>
 #include <vector>
 
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/field.h"
+#include "exprs/create_predicate_function.h"
+#include "exprs/expr_zonemap_filter.h"
 #include "exprs/hybrid_set.h"
+#include "exprs/vliteral.h"
 
 namespace doris {
 
@@ -74,6 +85,105 @@ std::vector<std::string> generate_values<std::string>(size_t n) {
 // Number of find() calls per iteration to amortize loop overhead.
 static constexpr size_t FIND_ITERS = 10000;
 
+enum class StringSetWorkload { SHORT, LONG, EMBEDDED_NUL };
+enum class StringSetLookup { HIT, MISS };
+
+template <StringSetWorkload workload>
+static std::vector<std::string> generate_string_set_values(size_t n, bool misses) {
+    std::vector<std::string> values;
+    values.reserve(n);
+    for (size_t i = 0; i < n; ++i) {
+        const auto key = std::to_string(i);
+        if constexpr (workload == StringSetWorkload::SHORT) {
+            values.emplace_back((misses ? "miss_" : "hit_") + key);
+        } else if constexpr (workload == StringSetWorkload::LONG) {
+            std::string value(192, static_cast<char>('a' + i % 26));
+            value.append(misses ? "_miss_" : "_hit_");
+            value.append(key);
+            values.emplace_back(std::move(value));
+        } else {
+            std::string value = misses ? "miss" : "hit";
+            value.push_back('\0');
+            value.append("binary_");
+            value.append(key);
+            value.push_back('\0');
+            value.append(32, static_cast<char>('a' + i % 26));
+            values.emplace_back(std::move(value));
+        }
+    }
+    return values;
+}
+
+template <StringSetWorkload workload, StringSetLookup lookup>
+static void BM_StringSet_StringRefFind(benchmark::State& state) {
+    const auto set_size = static_cast<size_t>(state.range(0));
+    const auto hit_values = generate_string_set_values<workload>(set_size, false);
+    const auto miss_values = generate_string_set_values<workload>(set_size, true);
+
+    StringSet<> set(false);
+    std::vector<StringRef> hit_refs;
+    std::vector<StringRef> miss_refs;
+    hit_refs.reserve(set_size);
+    miss_refs.reserve(set_size);
+    for (size_t i = 0; i < set_size; ++i) {
+        StringRef value(hit_values[i]);
+        set.insert(&value);
+        hit_refs.emplace_back(hit_values[i]);
+        miss_refs.emplace_back(miss_values[i]);
+    }
+
+    const auto& lookup_refs = lookup == StringSetLookup::HIT ? hit_refs : miss_refs;
+    const size_t lookup_iters = std::max(FIND_ITERS, set_size);
+    const size_t expected_found = lookup == StringSetLookup::HIT ? lookup_iters : 0;
+    for (auto _ : state) {
+        size_t found = 0;
+        for (size_t i = 0; i < lookup_iters; ++i) {
+            const auto index = i % set_size;
+            found += set.find(&lookup_refs[index]);
+        }
+        benchmark::DoNotOptimize(found);
+        if (found != expected_found) {
+            state.SkipWithError("StringSet lookup returned an unexpected result");
+            break;
+        }
+    }
+    state.SetItemsProcessed(state.iterations() * lookup_iters);
+}
+
+static void BM_StringSet_StringRefFindShortHit(benchmark::State& state) {
+    BM_StringSet_StringRefFind<StringSetWorkload::SHORT, StringSetLookup::HIT>(state);
+}
+
+static void BM_StringSet_StringRefFindShortMiss(benchmark::State& state) {
+    BM_StringSet_StringRefFind<StringSetWorkload::SHORT, StringSetLookup::MISS>(state);
+}
+
+static void BM_StringSet_StringRefFindLongHit(benchmark::State& state) {
+    BM_StringSet_StringRefFind<StringSetWorkload::LONG, StringSetLookup::HIT>(state);
+}
+
+static void BM_StringSet_StringRefFindLongMiss(benchmark::State& state) {
+    BM_StringSet_StringRefFind<StringSetWorkload::LONG, StringSetLookup::MISS>(state);
+}
+
+static void BM_StringSet_StringRefFindEmbeddedNulHit(benchmark::State& state) {
+    BM_StringSet_StringRefFind<StringSetWorkload::EMBEDDED_NUL, StringSetLookup::HIT>(state);
+}
+
+static void BM_StringSet_StringRefFindEmbeddedNulMiss(benchmark::State& state) {
+    BM_StringSet_StringRefFind<StringSetWorkload::EMBEDDED_NUL, StringSetLookup::MISS>(state);
+}
+
+#define REGISTER_STRING_SET_LOOKUP(NAME) \
+    BENCHMARK(NAME)->Arg(64)->Arg(1024)->Arg(40960)->Unit(benchmark::kMicrosecond)
+
+REGISTER_STRING_SET_LOOKUP(BM_StringSet_StringRefFindShortHit);
+REGISTER_STRING_SET_LOOKUP(BM_StringSet_StringRefFindShortMiss);
+REGISTER_STRING_SET_LOOKUP(BM_StringSet_StringRefFindLongHit);
+REGISTER_STRING_SET_LOOKUP(BM_StringSet_StringRefFindLongMiss);
+REGISTER_STRING_SET_LOOKUP(BM_StringSet_StringRefFindEmbeddedNulHit);
+REGISTER_STRING_SET_LOOKUP(BM_StringSet_StringRefFindEmbeddedNulMiss);
+
 // ============================================================
 // FixedContainer benchmark: insert N values, then find them
 // ============================================================
@@ -130,11 +240,13 @@ static void BM_DynamicContainer_Find(benchmark::State& state) {
 // Register benchmarks for int32_t
 // ============================================================
 
-#define REGISTER_FIXED_INT32(N)                                                            \
-    BENCHMARK(BM_FixedContainer_Find<int32_t, N>)->Name("Fixed_Int32_N" #N)->Unit(         \
-            benchmark::kMicrosecond);                                                      \
-    BENCHMARK(BM_DynamicContainer_Find<int32_t, N>)->Name("Dynamic_Int32_N" #N)->Unit(     \
-            benchmark::kMicrosecond);
+#define REGISTER_FIXED_INT32(N)                     \
+    BENCHMARK(BM_FixedContainer_Find<int32_t, N>)   \
+            ->Name("Fixed_Int32_N" #N)              \
+            ->Unit(benchmark::kMicrosecond);        \
+    BENCHMARK(BM_DynamicContainer_Find<int32_t, N>) \
+            ->Name("Dynamic_Int32_N" #N)            \
+            ->Unit(benchmark::kMicrosecond);
 
 REGISTER_FIXED_INT32(1)
 REGISTER_FIXED_INT32(2)
@@ -149,11 +261,13 @@ REGISTER_FIXED_INT32(8)
 // Register benchmarks for int64_t
 // ============================================================
 
-#define REGISTER_FIXED_INT64(N)                                                            \
-    BENCHMARK(BM_FixedContainer_Find<int64_t, N>)->Name("Fixed_Int64_N" #N)->Unit(         \
-            benchmark::kMicrosecond);                                                      \
-    BENCHMARK(BM_DynamicContainer_Find<int64_t, N>)->Name("Dynamic_Int64_N" #N)->Unit(     \
-            benchmark::kMicrosecond);
+#define REGISTER_FIXED_INT64(N)                     \
+    BENCHMARK(BM_FixedContainer_Find<int64_t, N>)   \
+            ->Name("Fixed_Int64_N" #N)              \
+            ->Unit(benchmark::kMicrosecond);        \
+    BENCHMARK(BM_DynamicContainer_Find<int64_t, N>) \
+            ->Name("Dynamic_Int64_N" #N)            \
+            ->Unit(benchmark::kMicrosecond);
 
 REGISTER_FIXED_INT64(1)
 REGISTER_FIXED_INT64(2)
@@ -168,11 +282,13 @@ REGISTER_FIXED_INT64(8)
 // Register benchmarks for std::string
 // ============================================================
 
-#define REGISTER_FIXED_STRING(N)                                                               \
-    BENCHMARK(BM_FixedContainer_Find<std::string, N>)->Name("Fixed_String_N" #N)->Unit(        \
-            benchmark::kMicrosecond);                                                          \
-    BENCHMARK(BM_DynamicContainer_Find<std::string, N>)->Name("Dynamic_String_N" #N)->Unit(    \
-            benchmark::kMicrosecond);
+#define REGISTER_FIXED_STRING(N)                        \
+    BENCHMARK(BM_FixedContainer_Find<std::string, N>)   \
+            ->Name("Fixed_String_N" #N)                 \
+            ->Unit(benchmark::kMicrosecond);            \
+    BENCHMARK(BM_DynamicContainer_Find<std::string, N>) \
+            ->Name("Dynamic_String_N" #N)               \
+            ->Unit(benchmark::kMicrosecond);
 
 REGISTER_FIXED_STRING(1)
 REGISTER_FIXED_STRING(2)
@@ -183,8 +299,309 @@ REGISTER_FIXED_STRING(6)
 REGISTER_FIXED_STRING(7)
 REGISTER_FIXED_STRING(8)
 
+// ============================================================
+// Metadata-pruning snapshot benchmark
+// ============================================================
+
+struct LegacyInZonemapSnapshot {
+    bool contains_null = false;
+    std::vector<Field> values;
+    Field min_value;
+    Field max_value;
+};
+
+static void materialize_hybrid_set_legacy(HybridSetBase& set, const DataTypePtr& data_type,
+                                          LegacyInZonemapSnapshot* result) {
+    DORIS_CHECK(result != nullptr);
+    DORIS_CHECK(data_type != nullptr);
+    const auto value_type = remove_nullable(data_type);
+    DORIS_CHECK(value_type != nullptr);
+
+    result->contains_null = set.contain_null();
+    result->values.clear();
+    result->min_value = Field();
+    result->max_value = Field();
+
+    auto* iterator = set.begin();
+    while (iterator->has_next()) {
+        const void* value = iterator->get_value();
+        if (value != nullptr) {
+            TExprNode literal_node = expr_zonemap::create_texpr_node_from_hybrid_set_value(
+                    value, value_type->get_primitive_type(), value_type->get_precision(),
+                    value_type->get_scale());
+            auto literal = VLiteral::create_shared(literal_node);
+            Field field;
+            literal->get_column_ptr()->get(0, field);
+            result->values.emplace_back(std::move(field));
+        }
+        iterator->next();
+    }
+
+    if (!result->values.empty()) {
+        const auto minmax = std::ranges::minmax_element(
+                result->values, [](const Field& lhs, const Field& rhs) { return lhs < rhs; });
+        result->min_value = *minmax.min;
+        result->max_value = *minmax.max;
+    }
+}
+
+static void BM_HybridSet_GetMinMaxInt32(benchmark::State& state) {
+    HybridSet<TYPE_INT> set(false);
+    const auto values = generate_values<int32_t>(state.range(0));
+    for (const auto value : values) {
+        set.insert(&value);
+    }
+
+    for (auto _ : state) {
+        Field min_value;
+        Field max_value;
+        set.get_min_max(min_value, max_value);
+        benchmark::DoNotOptimize(min_value);
+        benchmark::DoNotOptimize(max_value);
+    }
+    state.SetItemsProcessed(state.iterations() * state.range(0));
+}
+
+BENCHMARK(BM_HybridSet_GetMinMaxInt32)
+        ->Arg(64)
+        ->Arg(65)
+        ->Arg(1024)
+        ->Arg(40960)
+        ->Unit(benchmark::kMicrosecond);
+
+static void BM_HybridSet_LegacyMaterializeInt32(benchmark::State& state) {
+    HybridSet<TYPE_INT> set(false);
+    const auto values = generate_values<int32_t>(state.range(0));
+    for (const auto value : values) {
+        set.insert(&value);
+    }
+    const auto data_type = std::make_shared<DataTypeInt32>();
+
+    for (auto _ : state) {
+        LegacyInZonemapSnapshot materialized;
+        materialize_hybrid_set_legacy(set, data_type, &materialized);
+        benchmark::DoNotOptimize(materialized.values);
+        benchmark::DoNotOptimize(materialized.min_value);
+        benchmark::DoNotOptimize(materialized.max_value);
+    }
+    state.SetItemsProcessed(state.iterations() * state.range(0));
+}
+
+BENCHMARK(BM_HybridSet_LegacyMaterializeInt32)
+        ->Arg(64)
+        ->Arg(65)
+        ->Arg(1024)
+        ->Arg(40960)
+        ->Unit(benchmark::kMicrosecond);
+
+static void BM_StringSet_GetMinMaxLong(benchmark::State& state) {
+    StringSet<> set(false);
+    const auto values = generate_string_set_values<StringSetWorkload::LONG>(
+            static_cast<size_t>(state.range(0)), false);
+    for (const auto& value : values) {
+        StringRef string_ref(value);
+        set.insert(&string_ref);
+    }
+
+    for (auto _ : state) {
+        Field min_value;
+        Field max_value;
+        set.get_min_max(min_value, max_value);
+        benchmark::DoNotOptimize(min_value);
+        benchmark::DoNotOptimize(max_value);
+    }
+    state.SetItemsProcessed(state.iterations() * state.range(0));
+}
+
+BENCHMARK(BM_StringSet_GetMinMaxLong)
+        ->Arg(65)
+        ->Arg(1024)
+        ->Arg(40960)
+        ->Unit(benchmark::kMicrosecond);
+
+static void BM_StringSet_LegacyMaterializeLong(benchmark::State& state) {
+    StringSet<> set(false);
+    const auto values = generate_string_set_values<StringSetWorkload::LONG>(
+            static_cast<size_t>(state.range(0)), false);
+    for (const auto& value : values) {
+        StringRef string_ref(value);
+        set.insert(&string_ref);
+    }
+    const auto data_type = std::make_shared<DataTypeString>();
+
+    for (auto _ : state) {
+        LegacyInZonemapSnapshot materialized;
+        materialize_hybrid_set_legacy(set, data_type, &materialized);
+        benchmark::DoNotOptimize(materialized.values);
+        benchmark::DoNotOptimize(materialized.min_value);
+        benchmark::DoNotOptimize(materialized.max_value);
+    }
+    state.SetItemsProcessed(state.iterations() * state.range(0));
+}
+
+BENCHMARK(BM_StringSet_LegacyMaterializeLong)
+        ->Arg(65)
+        ->Arg(1024)
+        ->Arg(40960)
+        ->Unit(benchmark::kMicrosecond);
+
+// ============================================================
+// Typed range lookup benchmark
+// ============================================================
+
+static void BM_HybridSet_ContainsAnyInRangeInt32(benchmark::State& state) {
+    const auto set_size = static_cast<size_t>(state.range(0));
+    HybridSet<TYPE_INT> set(false);
+    const auto values = generate_values<int32_t>(set_size);
+    for (const auto value : values) {
+        set.insert(&value);
+    }
+
+    const int32_t hit_value = values[set_size / 2];
+    const auto hit = Field::create_field<TYPE_INT>(hit_value);
+    const auto miss_min = Field::create_field<TYPE_INT>(hit_value + 1);
+    const auto miss_max = Field::create_field<TYPE_INT>(hit_value + 6);
+    DORIS_CHECK(set.contains_any_in_range(hit, hit));
+    DORIS_CHECK(!set.contains_any_in_range(miss_min, miss_max));
+
+    for (auto _ : state) {
+        bool hit_result = set.contains_any_in_range(hit, hit);
+        bool miss_result = set.contains_any_in_range(miss_min, miss_max);
+        benchmark::DoNotOptimize(hit_result);
+        benchmark::DoNotOptimize(miss_result);
+    }
+    state.SetItemsProcessed(state.iterations() * 2);
+}
+
+BENCHMARK(BM_HybridSet_ContainsAnyInRangeInt32)->Arg(8)->Arg(64)->Unit(benchmark::kNanosecond);
+
+static void BM_BitSetSmallInt_ContainsAnyInWideHole(benchmark::State& state) {
+    HybridSet<TYPE_SMALLINT, BitSetContainer<int16_t>> set(false);
+    const int16_t min_value = std::numeric_limits<int16_t>::min();
+    const int16_t max_value = std::numeric_limits<int16_t>::max();
+    set.insert(&min_value);
+    set.insert(&max_value);
+
+    const auto hole_min = Field::create_field<TYPE_SMALLINT>(static_cast<int16_t>(min_value + 1));
+    const auto hole_max = Field::create_field<TYPE_SMALLINT>(static_cast<int16_t>(max_value - 1));
+    DORIS_CHECK(!set.contains_any_in_range(hole_min, hole_max));
+
+    for (auto _ : state) {
+        bool found = set.contains_any_in_range(hole_min, hole_max);
+        benchmark::DoNotOptimize(found);
+    }
+    state.SetItemsProcessed(state.iterations());
+}
+
+BENCHMARK(BM_BitSetSmallInt_ContainsAnyInWideHole)->Unit(benchmark::kNanosecond);
+
+// Measure only the native HybridSet traversal used by Bloom pruning. The lightweight fingerprint
+// predicate keeps storage Bloom-filter implementation costs outside this microbenchmark; both the
+// predicate and its captured fingerprint are constructed outside the timed loop.
+
+static uint64_t raw_bloom_fingerprint(const char* data, size_t size) {
+    uint64_t fingerprint = 14695981039346656037ULL;
+    for (size_t i = 0; i < size; ++i) {
+        fingerprint ^= static_cast<uint8_t>(data[i]);
+        fingerprint *= 1099511628211ULL;
+    }
+    return fingerprint;
+}
+
+enum class RawBloomProbeLookup { FULL_MISS, HIT };
+
+template <RawBloomProbeLookup lookup>
+static void BM_HybridSet_AnyMatchRawInt32(benchmark::State& state) {
+    const auto set_size = static_cast<size_t>(state.range(0));
+    std::shared_ptr<HybridSetBase> set(create_set(TYPE_INT, false));
+    const auto values = generate_values<int32_t>(set_size);
+    for (const auto value : values) {
+        set->insert(&value);
+    }
+
+    const int32_t target =
+            lookup == RawBloomProbeLookup::HIT ? values[set_size / 2] : values.back() + 1;
+    const auto target_fingerprint =
+            raw_bloom_fingerprint(reinterpret_cast<const char*>(&target), sizeof(target));
+    const auto predicate = [target_fingerprint](const char* data, size_t size) {
+        return raw_bloom_fingerprint(data, size) == target_fingerprint;
+    };
+    constexpr bool expected = lookup == RawBloomProbeLookup::HIT;
+    DORIS_CHECK_EQ(set->any_match_raw(TYPE_INT, predicate), expected);
+
+    for (auto _ : state) {
+        bool found = set->any_match_raw(TYPE_INT, predicate);
+        benchmark::DoNotOptimize(found);
+    }
+    state.SetItemsProcessed(state.iterations());
+}
+
+static void BM_HybridSet_AnyMatchRawInt32FullMiss(benchmark::State& state) {
+    BM_HybridSet_AnyMatchRawInt32<RawBloomProbeLookup::FULL_MISS>(state);
+}
+
+static void BM_HybridSet_AnyMatchRawInt32Hit(benchmark::State& state) {
+    BM_HybridSet_AnyMatchRawInt32<RawBloomProbeLookup::HIT>(state);
+}
+
+template <StringSetWorkload workload, RawBloomProbeLookup lookup>
+static void BM_StringSet_AnyMatchRaw(benchmark::State& state) {
+    const auto set_size = static_cast<size_t>(state.range(0));
+    std::shared_ptr<HybridSetBase> set(create_set(TYPE_STRING, false));
+    const auto values = generate_string_set_values<workload>(set_size, false);
+    const auto misses = generate_string_set_values<workload>(set_size, true);
+    for (const auto& value : values) {
+        StringRef string_ref(value);
+        set->insert(&string_ref);
+    }
+
+    const auto& target =
+            lookup == RawBloomProbeLookup::HIT ? values[set_size / 2] : misses[set_size / 2];
+    const auto target_fingerprint = raw_bloom_fingerprint(target.data(), target.size());
+    const auto predicate = [target_fingerprint](const char* data, size_t size) {
+        return raw_bloom_fingerprint(data, size) == target_fingerprint;
+    };
+    constexpr bool expected = lookup == RawBloomProbeLookup::HIT;
+    DORIS_CHECK_EQ(set->any_match_raw(TYPE_STRING, predicate), expected);
+
+    for (auto _ : state) {
+        bool found = set->any_match_raw(TYPE_STRING, predicate);
+        benchmark::DoNotOptimize(found);
+    }
+    state.SetItemsProcessed(state.iterations());
+}
+
+static void BM_StringSet_AnyMatchRawLongFullMiss(benchmark::State& state) {
+    BM_StringSet_AnyMatchRaw<StringSetWorkload::LONG, RawBloomProbeLookup::FULL_MISS>(state);
+}
+
+static void BM_StringSet_AnyMatchRawLongHit(benchmark::State& state) {
+    BM_StringSet_AnyMatchRaw<StringSetWorkload::LONG, RawBloomProbeLookup::HIT>(state);
+}
+
+static void BM_StringSet_AnyMatchRawEmbeddedNulFullMiss(benchmark::State& state) {
+    BM_StringSet_AnyMatchRaw<StringSetWorkload::EMBEDDED_NUL, RawBloomProbeLookup::FULL_MISS>(
+            state);
+}
+
+static void BM_StringSet_AnyMatchRawEmbeddedNulHit(benchmark::State& state) {
+    BM_StringSet_AnyMatchRaw<StringSetWorkload::EMBEDDED_NUL, RawBloomProbeLookup::HIT>(state);
+}
+
+#define REGISTER_RAW_BLOOM_PROBE(NAME) \
+    BENCHMARK(NAME)->Arg(64)->Arg(1024)->Arg(40960)->Unit(benchmark::kMicrosecond)
+
+REGISTER_RAW_BLOOM_PROBE(BM_HybridSet_AnyMatchRawInt32FullMiss);
+REGISTER_RAW_BLOOM_PROBE(BM_HybridSet_AnyMatchRawInt32Hit);
+REGISTER_RAW_BLOOM_PROBE(BM_StringSet_AnyMatchRawLongFullMiss);
+REGISTER_RAW_BLOOM_PROBE(BM_StringSet_AnyMatchRawLongHit);
+REGISTER_RAW_BLOOM_PROBE(BM_StringSet_AnyMatchRawEmbeddedNulFullMiss);
+REGISTER_RAW_BLOOM_PROBE(BM_StringSet_AnyMatchRawEmbeddedNulHit);
+
 #undef REGISTER_FIXED_INT32
 #undef REGISTER_FIXED_INT64
 #undef REGISTER_FIXED_STRING
+#undef REGISTER_STRING_SET_LOOKUP
+#undef REGISTER_RAW_BLOOM_PROBE
 
 } // namespace doris
diff --git a/be/benchmark/benchmark_main.cpp b/be/benchmark/benchmark_main.cpp
index 7cce267d01e94e..9a11387d1961f5 100644
--- a/be/benchmark/benchmark_main.cpp
+++ b/be/benchmark/benchmark_main.cpp
@@ -29,6 +29,7 @@
 #include "benchmark_fastunion.hpp"
 #include "benchmark_fmod.hpp"
 #include "benchmark_hll_merge.hpp"
+#include "benchmark_hybrid_set.hpp"
 #include "benchmark_json_extract.hpp"
 #include "benchmark_zone_map_index.hpp"
 #include "binary_cast_benchmark.hpp"
diff --git a/be/src/exprs/bitset_container.h b/be/src/exprs/bitset_container.h
index a230d3b98eb6cc..c30aa6371556ad 100644
--- a/be/src/exprs/bitset_container.h
+++ b/be/src/exprs/bitset_container.h
@@ -17,6 +17,7 @@
 
 #pragma once
 
+#include <algorithm>
 #include <array>
 #include <cstddef>
 #include <cstdint>
@@ -100,7 +101,7 @@ class BitSetContainer {
         mutable T _cached_value = T();
     };
 
-    BitSetContainer() { _data.fill(false); }
+    BitSetContainer() = default;
 
     ~BitSetContainer() = default;
 
@@ -108,6 +109,16 @@ class BitSetContainer {
         auto idx = _to_index(value);
         if (!_data[idx]) {
             _data[idx] = true;
+            const size_t block = idx / VALUES_PER_BLOCK;
+            _nonempty_blocks[block / SUMMARY_WORD_BITS] |= uint64_t {1}
+                                                           << (block % SUMMARY_WORD_BITS);
+            if (_size == 0) {
+                _min_value = value;
+                _max_value = value;
+            } else {
+                _min_value = std::min(_min_value, value);
+                _max_value = std::max(_max_value, value);
+            }
             _size++;
         }
     }
@@ -116,10 +127,33 @@ class BitSetContainer {
 
     void clear() {
         _data.fill(false);
+        _nonempty_blocks.fill(0);
         _size = 0;
     }
 
     size_t size() const { return _size; }
+    T min_value() const { return _min_value; }
+    T max_value() const { return _max_value; }
+
+    template <typename Less>
+    bool contains_any_in_range(T min_value, T max_value, Less&& less) const {
+        if (_size == 0) {
+            return false;
+        }
+        min_value = std::max(min_value, _min_value);
+        max_value = std::min(max_value, _max_value);
+        if (less(max_value, min_value)) {
+            return false;
+        }
+
+        const size_t min_index = _to_index(min_value);
+        const size_t max_index = _to_index(max_value);
+        if (min_value < 0 && max_value >= 0) {
+            return _has_value_in_raw_index_range(min_index, RANGE - 1) ||
+                   _has_value_in_raw_index_range(0, max_index);
+        }
+        return _has_value_in_raw_index_range(min_index, max_index);
+    }
 
     Iterator begin() const { return Iterator(&_data, 0); }
     Iterator end() const { return Iterator(&_data, RANGE); }
@@ -141,6 +175,63 @@ class BitSetContainer {
     }
 
 private:
+    static constexpr size_t VALUES_PER_BLOCK = 64;
+    static constexpr size_t SUMMARY_WORD_BITS = 64;
+    static constexpr size_t BLOCK_COUNT = (RANGE + VALUES_PER_BLOCK - 1) / VALUES_PER_BLOCK;
+    static constexpr size_t SUMMARY_WORD_COUNT =
+            (BLOCK_COUNT + SUMMARY_WORD_BITS - 1) / SUMMARY_WORD_BITS;
+
+    bool _has_nonempty_block_in_range(size_t first_block, size_t last_block) const {
+        const size_t first_word = first_block / SUMMARY_WORD_BITS;
+        const size_t last_word = last_block / SUMMARY_WORD_BITS;
+        const size_t first_bit = first_block % SUMMARY_WORD_BITS;
+        const size_t last_bit = last_block % SUMMARY_WORD_BITS;
+        const uint64_t first_mask = ~uint64_t {0} << first_bit;
+        const uint64_t last_mask = last_bit == SUMMARY_WORD_BITS - 1
+                                           ? ~uint64_t {0}
+                                           : (uint64_t {1} << (last_bit + 1)) - uint64_t {1};
+        if (first_word == last_word) {
+            return (_nonempty_blocks[first_word] & first_mask & last_mask) != 0;
+        }
+        if ((_nonempty_blocks[first_word] & first_mask) != 0) {
+            return true;
+        }
+        for (size_t word = first_word + 1; word < last_word; ++word) {
+            if (_nonempty_blocks[word] != 0) {
+                return true;
+            }
+        }
+        return (_nonempty_blocks[last_word] & last_mask) != 0;
+    }
+
+    bool _has_value_in_raw_index_range(size_t first, size_t last) const {
+        const size_t first_block = first / VALUES_PER_BLOCK;
+        const size_t last_block = last / VALUES_PER_BLOCK;
+        if (first_block == last_block) {
+            return _has_value_in_boundary_range(first, last);
+        }
+
+        const size_t first_block_end = (first_block + 1) * VALUES_PER_BLOCK;
+        if (_has_value_in_boundary_range(first, first_block_end - 1)) {
+            return true;
+        }
+        const size_t last_block_start = last_block * VALUES_PER_BLOCK;
+        if (_has_value_in_boundary_range(last_block_start, last)) {
+            return true;
+        }
+        return first_block + 1 < last_block &&
+               _has_nonempty_block_in_range(first_block + 1, last_block - 1);
+    }
+
+    bool _has_value_in_boundary_range(size_t first, size_t last) const {
+        for (size_t index = first; index <= last; ++index) {
+            if (_data[index]) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     ALWAYS_INLINE constexpr size_t _to_index(T value) const {
         if constexpr (std::is_same_v<T, int8_t>) {
             return static_cast<uint8_t>(value);
@@ -149,8 +240,11 @@ class BitSetContainer {
         }
     }
 
-    std::array<bool, RANGE> _data;
+    std::array<bool, RANGE> _data {};
+    std::array<uint64_t, SUMMARY_WORD_COUNT> _nonempty_blocks {};
     size_t _size = 0;
+    T _min_value {};
+    T _max_value {};
 };
 
-} // namespace doris
\ No newline at end of file
+} // namespace doris
diff --git a/be/src/exprs/expr_zonemap_filter.cpp b/be/src/exprs/expr_zonemap_filter.cpp
index 8cb999d08bafd4..c0cff7ae3eb91b 100644
--- a/be/src/exprs/expr_zonemap_filter.cpp
+++ b/be/src/exprs/expr_zonemap_filter.cpp
@@ -49,10 +49,6 @@ std::optional<std::pair<Field, DataTypePtr>> field_from_literal_expr(const VExpr
     return std::make_pair(std::move(field), literal->get_data_type());
 }
 
-bool value_in_range(const Field& value, const Field& min_value, const Field& max_value) {
-    return value >= min_value && value <= max_value;
-}
-
 bool dictionary_contains(const DictionaryEvalContext::SlotDictionary& dictionary,
                          const Field& value) {
     return std::ranges::any_of(dictionary.values, [&](const Field& dictionary_value) {
@@ -143,40 +139,24 @@ TExprNode create_texpr_node_from_hybrid_set_value(const void* data, const Primit
     return create_texpr_node_from(data, type, precision, scale);
 }
 
-Status materialize_hybrid_set_for_zonemap_filter(HybridSetBase& set, const DataTypePtr& data_type,
-                                                 InZonemapMaterializedSet* result) {
-    DORIS_CHECK(result != nullptr);
+void get_hybrid_set_min_max_for_zonemap_filter(const std::shared_ptr<HybridSetBase>& set,
+                                               const DataTypePtr& data_type,
+                                               InZonemapMinMax& result) {
+    DORIS_CHECK(set != nullptr);
     DORIS_CHECK(data_type != nullptr);
     const auto value_type = remove_nullable(data_type);
     DORIS_CHECK(value_type != nullptr);
 
-    result->contains_null = set.contain_null();
-    result->values.clear();
-    result->min_value = Field();
-    result->max_value = Field();
-
-    auto* iterator = set.begin();
-    while (iterator->has_next()) {
-        const void* value = iterator->get_value();
-        if (value != nullptr) {
-            TExprNode literal_node = create_texpr_node_from_hybrid_set_value(
-                    value, value_type->get_primitive_type(), value_type->get_precision(),
-                    value_type->get_scale());
-            auto literal = VLiteral::create_shared(literal_node);
-            Field field;
-            literal->get_column_ptr()->get(0, field);
-            result->values.emplace_back(std::move(field));
-        }
-        iterator->next();
-    }
-
-    if (!result->values.empty()) {
-        auto minmax = std::ranges::minmax_element(
-                result->values, [](const Field& lhs, const Field& rhs) { return lhs < rhs; });
-        result->min_value = *minmax.min;
-        result->max_value = *minmax.max;
+    set->get_min_max(result.min_value, result.max_value);
+    const auto value_count = set->size();
+    if (value_count != 0) {
+        DORIS_CHECK(!result.min_value.is_null());
+        DORIS_CHECK(!result.max_value.is_null());
+        DORIS_CHECK(field_types_compatible(result.min_value.get_type(),
+                                           value_type->get_primitive_type()));
+        DORIS_CHECK(field_types_compatible(result.max_value.get_type(),
+                                           value_type->get_primitive_type()));
     }
-    return Status::OK();
 }
 
 std::optional<SlotLiteral> extract_slot_and_literal(const VExprSPtrs& args) {
@@ -243,23 +223,30 @@ ZoneMapFilterResult eval_null_zonemap(const ZoneMapEvalContext& ctx, const VExpr
 }
 
 ZoneMapFilterResult eval_in_zonemap(const ZoneMapEvalContext& ctx, const VExprSPtr& slot_expr,
-                                    bool is_not_in, const std::vector<Field>& values,
-                                    const Field& min_value, const Field& max_value) {
+                                    bool is_not_in, const InZonemapMinMax& values,
+                                    const HybridSetBase& set) {
     auto slot = std::dynamic_pointer_cast<VSlotRef>(slot_expr);
     DORIS_CHECK(slot != nullptr);
+    // NOT IN with a NULL literal is UNKNOWN for every non-null value. Zone maps do not retain
+    // enough row-level information to recover a match in that case.
+    if (is_not_in && set.contain_null()) {
+        return ZoneMapFilterResult::kNoMatch;
+    }
     // Empty IN has no candidate values, while NOT IN with an empty set cannot filter anything.
-    if (values.empty()) {
+    if (set.size() == 0) { // NOLINT(readability-container-size-empty)
         return is_not_in ? ZoneMapFilterResult::kMayMatch : ZoneMapFilterResult::kNoMatch;
     }
 
-    // The caller has materialized the IN set and precomputed its non-null min/max. They must match
+    // The caller has precomputed the IN set's owning non-null min/max. They must match
     // the expression slot type before being compared with storage zone-map statistics.
-    DORIS_CHECK(!min_value.is_null());
-    DORIS_CHECK(!max_value.is_null());
+    DORIS_CHECK(!values.min_value.is_null());
+    DORIS_CHECK(!values.max_value.is_null());
     auto data_type = remove_nullable(slot->data_type());
     DORIS_CHECK(data_type != nullptr);
-    DORIS_CHECK(field_types_compatible(min_value.get_type(), data_type->get_primitive_type()));
-    DORIS_CHECK(field_types_compatible(max_value.get_type(), data_type->get_primitive_type()));
+    DORIS_CHECK(
+            field_types_compatible(values.min_value.get_type(), data_type->get_primitive_type()));
+    DORIS_CHECK(
+            field_types_compatible(values.max_value.get_type(), data_type->get_primitive_type()));
 
     // Re-check against the reader-schema type and the available zone map. Missing or unsupported
     // metadata must conservatively fall back to may-match.
@@ -285,39 +272,36 @@ ZoneMapFilterResult eval_in_zonemap(const ZoneMapEvalContext& ctx, const VExprSP
         // NOT IN can only prune when the whole zone contains exactly one non-null value and that
         // value is excluded by the set. Wider ranges may contain values that are not filtered.
         if (zone_map.min_value == zone_map.max_value) {
-            const bool only_value_is_filtered = std::ranges::any_of(
-                    values, [&](const Field& value) { return value == zone_map.min_value; });
+            const bool only_value_is_filtered = set.find(zone_map.min_value);
             return only_value_is_filtered ? ZoneMapFilterResult::kNoMatch
                                           : ZoneMapFilterResult::kMayMatch;
         }
         return ZoneMapFilterResult::kMayMatch;
     }
 
-    // First use the materialized IN-set min/max to rule out disjoint zone-map ranges.
-    if (zone_map.max_value < min_value || zone_map.min_value > max_value) {
+    // First use the IN-set min/max to rule out disjoint zone-map ranges.
+    if (zone_map.max_value < values.min_value || zone_map.min_value > values.max_value) {
         return ZoneMapFilterResult::kNoMatch;
     }
 
     // For large IN sets, avoid checking every point on the scan hot path. The range overlap above
     // is only a coarse may-match signal.
-    if (values.size() > kInZoneMapPointCheckThreshold) {
+    if (std::cmp_greater(set.size(), kInZoneMapPointCheckThreshold)) {
         ++ctx.stats.in_zonemap_range_only_count;
         return ZoneMapFilterResult::kMayMatch;
     }
 
-    // For small IN sets, verify whether any candidate value can fall into the zone-map range.
+    // Convert the two zone-map bounds to the HybridSet's native type once, then compare them
+    // directly with the typed set values without retaining a Field copy of every IN candidate.
     ++ctx.stats.in_zonemap_point_check_count;
-    for (const auto& value : values) {
-        if (value_in_range(value, zone_map.min_value, zone_map.max_value)) {
-            return ZoneMapFilterResult::kMayMatch;
-        }
-    }
-    return ZoneMapFilterResult::kNoMatch;
+    return set.contains_any_in_range(zone_map.min_value, zone_map.max_value)
+                   ? ZoneMapFilterResult::kMayMatch
+                   : ZoneMapFilterResult::kNoMatch;
 }
 
 ZoneMapFilterResult eval_eq_dictionary(const DictionaryEvalContext& ctx,
                                        const SlotLiteral& slot_literal) {
-    auto dictionary = ctx.slot(slot_literal.slot_index);
+    const auto* dictionary = ctx.slot(slot_literal.slot_index);
     if (dictionary == nullptr || dictionary->data_type == nullptr) {
         return ZoneMapFilterResult::kUnsupported;
     }
@@ -330,22 +314,24 @@ ZoneMapFilterResult eval_eq_dictionary(const DictionaryEvalContext& ctx,
 }
 
 ZoneMapFilterResult eval_in_dictionary(const DictionaryEvalContext& ctx, const VExprSPtr& slot_expr,
-                                       bool is_not_in, const std::vector<Field>& values) {
+                                       bool is_not_in, const HybridSetBase& values) {
     if (is_not_in) {
         return ZoneMapFilterResult::kUnsupported;
     }
     auto slot = std::dynamic_pointer_cast<VSlotRef>(slot_expr);
     DORIS_CHECK(slot != nullptr);
-    auto dictionary = ctx.slot(slot->column_id());
+    const auto* dictionary = ctx.slot(slot->column_id());
     if (dictionary == nullptr || dictionary->data_type == nullptr) {
         return ZoneMapFilterResult::kUnsupported;
     }
     DORIS_CHECK(data_types_compatible(dictionary->data_type, slot->data_type()));
-    if (values.empty()) {
+    // HybridSetBase::empty() also treats a NULL literal as non-empty, but dictionary pruning needs
+    // to know whether there are any non-NULL candidates.
+    if (values.size() == 0) { // NOLINT(readability-container-size-empty)
         return ZoneMapFilterResult::kNoMatch;
     }
-    for (const auto& value : values) {
-        if (!value.is_null() && dictionary_contains(*dictionary, value)) {
+    for (const auto& value : dictionary->values) {
+        if (!value.is_null() && values.find(value)) {
             return ZoneMapFilterResult::kMayMatch;
         }
     }
@@ -354,7 +340,7 @@ ZoneMapFilterResult eval_in_dictionary(const DictionaryEvalContext& ctx, const V
 
 ZoneMapFilterResult eval_eq_bloom_filter(const BloomFilterEvalContext& ctx,
                                          const SlotLiteral& slot_literal) {
-    auto slot_filter = ctx.slot(slot_literal.slot_index);
+    const auto* slot_filter = ctx.slot(slot_literal.slot_index);
     if (slot_filter == nullptr || slot_filter->data_type == nullptr ||
         slot_filter->bloom_filter == nullptr) {
         return ZoneMapFilterResult::kUnsupported;
@@ -370,27 +356,41 @@ ZoneMapFilterResult eval_eq_bloom_filter(const BloomFilterEvalContext& ctx,
 
 ZoneMapFilterResult eval_in_bloom_filter(const BloomFilterEvalContext& ctx,
                                          const VExprSPtr& slot_expr, bool is_not_in,
-                                         const std::vector<Field>& values) {
+                                         const HybridSetBase& values) {
     if (is_not_in) {
         return ZoneMapFilterResult::kUnsupported;
     }
     auto slot = std::dynamic_pointer_cast<VSlotRef>(slot_expr);
     DORIS_CHECK(slot != nullptr);
-    auto slot_filter = ctx.slot(slot->column_id());
+    const auto* slot_filter = ctx.slot(slot->column_id());
     if (slot_filter == nullptr || slot_filter->data_type == nullptr ||
         slot_filter->bloom_filter == nullptr) {
         return ZoneMapFilterResult::kUnsupported;
     }
     DORIS_CHECK(data_types_compatible(slot_filter->data_type, slot->data_type()));
-    if (values.empty()) {
+    if (values.size() == 0) { // NOLINT(readability-container-size-empty)
         return ZoneMapFilterResult::kNoMatch;
     }
-    for (const auto& value : values) {
-        if (!value.is_null() && bloom_filter_may_contain(*slot_filter, value)) {
-            return ZoneMapFilterResult::kMayMatch;
-        }
+    const auto value_type = remove_nullable(slot_filter->data_type)->get_primitive_type();
+    switch (value_type) {
+    case TYPE_BOOLEAN:
+    case TYPE_INT:
+    case TYPE_BIGINT:
+    case TYPE_FLOAT:
+    case TYPE_DOUBLE:
+    case TYPE_CHAR:
+    case TYPE_VARCHAR:
+    case TYPE_STRING:
+        break;
+    default:
+        return ZoneMapFilterResult::kMayMatch;
     }
-    return ZoneMapFilterResult::kNoMatch;
+    return values.any_match_raw(value_type,
+                                [slot_filter](const char* data, size_t size) {
+                                    return slot_filter->bloom_filter->test_bytes(data, size);
+                                })
+                   ? ZoneMapFilterResult::kMayMatch
+                   : ZoneMapFilterResult::kNoMatch;
 }
 
 // Return the only slot ordinal referenced by a zonemap-evaluable expression. A negative result is
diff --git a/be/src/exprs/expr_zonemap_filter.h b/be/src/exprs/expr_zonemap_filter.h
index 5b0df00e121547..cffcd2c59b8e77 100644
--- a/be/src/exprs/expr_zonemap_filter.h
+++ b/be/src/exprs/expr_zonemap_filter.h
@@ -18,16 +18,18 @@
 #pragma once
 
 #include <compare>
+#include <cstddef>
 #include <map>
+#include <memory>
 #include <optional>
 #include <vector>
 
 #include "common/check.h"
-#include "common/status.h"
 #include "core/data_type/data_type.h"
 #include "core/data_type/data_type_nullable.h"
 #include "core/data_type/define_primitive_type.h"
 #include "core/field.h"
+#include "exprs/hybrid_set_min_max.h"
 #include "exprs/vexpr_fwd.h"
 #include "storage/index/zone_map/zonemap_eval_context.h"
 #include "storage/index/zone_map/zonemap_filter_result.h"
@@ -44,12 +46,7 @@ class BloomFilter;
 
 namespace doris::expr_zonemap {
 
-struct InZonemapMaterializedSet {
-    bool contains_null = false;
-    std::vector<Field> values;
-    Field min_value;
-    Field max_value;
-};
+using InZonemapMinMax = HybridSetMinMax;
 
 // Dictionary pruning evaluates file-level dictionary values, not row-level data. A kNoMatch result
 // means no non-null dictionary entry can satisfy the expression, so the whole row group can be
@@ -104,8 +101,9 @@ std::optional<SlotLiteral> extract_slot_and_literal(const VExprSPtrs& args);
 TExprNode create_texpr_node_from_hybrid_set_value(const void* data, const PrimitiveType& type,
                                                   int precision, int scale);
 
-Status materialize_hybrid_set_for_zonemap_filter(HybridSetBase& set, const DataTypePtr& data_type,
-                                                 InZonemapMaterializedSet* result);
+void get_hybrid_set_min_max_for_zonemap_filter(const std::shared_ptr<HybridSetBase>& set,
+                                               const DataTypePtr& data_type,
+                                               InZonemapMinMax& result);
 
 inline bool field_types_compatible(PrimitiveType lhs, PrimitiveType rhs) {
     return lhs == rhs || (is_string_type(lhs) && is_string_type(rhs));
@@ -142,21 +140,21 @@ ZoneMapFilterResult eval_null_zonemap(const ZoneMapEvalContext& ctx, const VExpr
                                       bool is_null);
 
 ZoneMapFilterResult eval_in_zonemap(const ZoneMapEvalContext& ctx, const VExprSPtr& slot_expr,
-                                    bool is_not_in, const std::vector<Field>& values,
-                                    const Field& min_value, const Field& max_value);
+                                    bool is_not_in, const InZonemapMinMax& values,
+                                    const HybridSetBase& set);
 
 ZoneMapFilterResult eval_eq_dictionary(const DictionaryEvalContext& ctx,
                                        const SlotLiteral& slot_literal);
 
 ZoneMapFilterResult eval_in_dictionary(const DictionaryEvalContext& ctx, const VExprSPtr& slot_expr,
-                                       bool is_not_in, const std::vector<Field>& values);
+                                       bool is_not_in, const HybridSetBase& values);
 
 ZoneMapFilterResult eval_eq_bloom_filter(const BloomFilterEvalContext& ctx,
                                          const SlotLiteral& slot_literal);
 
 ZoneMapFilterResult eval_in_bloom_filter(const BloomFilterEvalContext& ctx,
                                          const VExprSPtr& slot_expr, bool is_not_in,
-                                         const std::vector<Field>& values);
+                                         const HybridSetBase& values);
 
 // Return the only slot ordinal referenced by a zonemap-evaluable expression in its current
 // binding. Expressions that are unsupported by zonemap pruning, reference multiple slots, or use an
diff --git a/be/src/exprs/hybrid_set.h b/be/src/exprs/hybrid_set.h
index 0717e71fc2029a..878bce11b70277 100644
--- a/be/src/exprs/hybrid_set.h
+++ b/be/src/exprs/hybrid_set.h
@@ -20,13 +20,22 @@
 #include <gen_cpp/internal_service.pb.h>
 #include <pdqsort.h>
 
+#include <algorithm>
+#include <array>
 #include <cstring>
+#include <memory>
+#include <string_view>
+#include <vector>
 
+#include "common/check.h"
+#include "common/compare.h"
 #include "common/object_pool.h"
 #include "core/column/column_nullable.h"
 #include "core/column/column_string.h"
 #include "core/column/column_vector.h"
 #include "core/data_type/primitive_type.h"
+#include "core/field.h"
+#include "core/string_ref.h"
 #include "exec/common/hash_table/phmap_fwd_decl.h"
 #include "exec/runtime_filter/utils.h"
 #include "exprs/bitset_container.h"
@@ -35,6 +44,60 @@
 namespace doris {
 constexpr int FIXED_CONTAINER_MAX_SIZE = 8;
 
+namespace detail {
+
+template <typename Container, typename Less, typename FieldFactory>
+void get_hybrid_set_min_max(const Container& container, Less&& less, FieldFactory&& field_factory,
+                            Field& min_value, Field& max_value) {
+    min_value = Field {};
+    max_value = Field {};
+
+    if constexpr (requires {
+                      container.min_value();
+                      container.max_value();
+                  }) {
+        if (container.size() != 0) {
+            min_value = field_factory(container.min_value());
+            max_value = field_factory(container.max_value());
+            return;
+        }
+    }
+
+    auto iterator = container.begin();
+    const auto end = container.end();
+    if (iterator == end) {
+        return;
+    }
+
+    auto min_iterator = iterator;
+    auto max_iterator = iterator;
+    for (; iterator != end; ++iterator) {
+        const auto& value = *iterator;
+        if (less(value, *min_iterator)) {
+            min_iterator = iterator;
+        }
+        if (less(*max_iterator, value)) {
+            max_iterator = iterator;
+        }
+    }
+    min_value = field_factory(*min_iterator);
+    max_value = field_factory(*max_iterator);
+}
+
+template <PrimitiveType T>
+bool hybrid_set_value_less(const typename PrimitiveTypeTraits<T>::CppType& lhs,
+                           const typename PrimitiveTypeTraits<T>::CppType& rhs) {
+    if constexpr (T == TYPE_DATEV2 || T == TYPE_DATETIMEV2 || T == TYPE_TIMESTAMPTZ) {
+        return lhs.to_date_int_val() < rhs.to_date_int_val();
+    } else if constexpr (T == TYPE_FLOAT || T == TYPE_DOUBLE) {
+        return Compare::less(lhs, rhs);
+    } else {
+        return lhs < rhs;
+    }
+}
+
+} // namespace detail
+
 /**
  * Fix Container can use simd to improve performance. 1 <= N <= 8 can be improved performance by test. FIXED_CONTAINER_MAX_SIZE = 8.
  * @tparam T Element Type
@@ -47,6 +110,7 @@ class FixedContainer {
     using ElementType = T;
 
     class Iterator;
+    class ConstIterator;
 
     FixedContainer() { static_assert(N >= 0 && N <= FIXED_CONTAINER_MAX_SIZE); }
 
@@ -77,6 +141,14 @@ class FixedContainer {
         return _find_impl(value, std::make_index_sequence<N> {});
     }
 
+    template <typename Less>
+    bool contains_any_in_range(const T& min_value, const T& max_value, Less&& less) const {
+        DCHECK(!less(max_value, min_value));
+        return std::ranges::any_of(_data.begin(), _data.begin() + _size, [&](const T& value) {
+            return !less(value, min_value) && !less(max_value, value);
+        });
+    }
+
 private:
     template <size_t... I>
     ALWAYS_INLINE bool _find_impl(const T& value, std::index_sequence<I...>) const {
@@ -122,6 +194,38 @@ class FixedContainer {
     Iterator begin() { return Iterator(_data, 0); }
     Iterator end() { return Iterator(_data, _size); }
 
+    class ConstIterator {
+    public:
+        ConstIterator() = default;
+        explicit ConstIterator(const std::array<T, N>& data, size_t index)
+                : _data(&data), _index(index) {}
+        ConstIterator& operator++() {
+            ++_index;
+            return *this;
+        }
+        ConstIterator operator++(int) {
+            ConstIterator ret_val = *this;
+            ++(*this);
+            return ret_val;
+        }
+        bool operator==(ConstIterator other) const { return _index == other._index; }
+        bool operator!=(ConstIterator other) const { return !(*this == other); }
+        const T& operator*() const { return (*_data)[_index]; }
+        const T* operator->() const { return &operator*(); }
+
+        using iterator_category = std::forward_iterator_tag;
+        using difference_type = std::ptrdiff_t;
+        using value_type = T;
+        using pointer = const T*;
+        using reference = const T&;
+
+    private:
+        const std::array<T, N>* _data = nullptr;
+        size_t _index = 0;
+    };
+    ConstIterator begin() const { return ConstIterator(_data, 0); }
+    ConstIterator end() const { return ConstIterator(_data, _size); }
+
     void clear() {
         std::array<T, N> {}.swap(_data);
         _size = 0;
@@ -148,11 +252,13 @@ struct IsBitSetContainer<BitSetContainer<T>> : std::true_type {};
  * Dynamic Container uses phmap::flat_hash_set.
  * @tparam T Element Type
  */
-template <typename T>
+template <typename T, typename Hash = phmap::Hash<T>, typename Eq = doris::EqualTo<T>>
 class DynamicContainer {
 public:
+    using Set = flat_hash_set<T, Hash, Eq>;
     using Self = DynamicContainer;
-    using Iterator = typename flat_hash_set<T>::iterator;
+    using Iterator = typename Set::iterator;
+    using ConstIterator = typename Set::const_iterator;
     using ElementType = T;
 
     DynamicContainer() = default;
@@ -162,7 +268,19 @@ class DynamicContainer {
 
     void insert(Iterator begin, Iterator end) { _set.insert(begin, end); }
 
-    bool find(const T& value) const { return _set.contains(value); }
+    template <typename Key>
+        requires requires(const Set& set, const Key& key) { set.find(key); }
+    bool find(const Key& value) const {
+        return _set.find(value) != _set.end();
+    }
+
+    template <typename Less>
+    bool contains_any_in_range(const T& min_value, const T& max_value, Less&& less) const {
+        DCHECK(!less(max_value, min_value));
+        return std::ranges::any_of(_set, [&](const T& value) {
+            return !less(value, min_value) && !less(max_value, value);
+        });
+    }
 
     void clear() { _set.clear(); }
 
@@ -170,14 +288,47 @@ class DynamicContainer {
 
     Iterator end() { return _set.end(); }
 
+    ConstIterator begin() const { return _set.begin(); }
+
+    ConstIterator end() const { return _set.end(); }
+
     size_t size() const { return _set.size(); }
 
 private:
-    flat_hash_set<T> _set;
+    Set _set;
+};
+
+struct HybridSetStringHash {
+    using is_transparent = void;
+
+    size_t operator()(const StringRef& value) const { return StringRefHash {}(value); }
+    size_t operator()(const std::string& value) const {
+        return operator()(StringRef(value.data(), value.size()));
+    }
+};
+
+struct HybridSetStringEqual {
+    using is_transparent = void;
+
+    static std::string_view to_view(const StringRef& value) {
+        return {value.size == 0 ? "" : value.data, value.size};
+    }
+    static std::string_view to_view(const std::string& value) { return value; }
+
+    template <typename Lhs, typename Rhs>
+    bool operator()(const Lhs& lhs, const Rhs& rhs) const {
+        return to_view(lhs) == to_view(rhs);
+    }
 };
 
 // TODO Maybe change void* parameter to template parameter better.
 class HybridSetBase : public FilterBase {
+protected:
+    using RawValuePredicateInvoker = bool (*)(const char* data, size_t size, const void* predicate);
+
+    virtual bool _any_match_raw(PrimitiveType value_type, RawValuePredicateInvoker invoke,
+                                const void* predicate) const = 0;
+
 public:
     HybridSetBase(bool null_aware) : FilterBase(null_aware) {}
     virtual ~HybridSetBase() = default;
@@ -200,11 +351,30 @@ class HybridSetBase : public FilterBase {
     }
 
     virtual void clear() = 0;
-    bool empty() { return !_contain_null && size() == 0; }
-    virtual int size() = 0;
+    bool empty() const { return !_contain_null && size() == 0; }
+    virtual int size() const = 0;
     virtual bool find(const void* data) const = 0;
     // use in vectorize execute engine
     virtual bool find(const void* data, size_t) const = 0;
+    virtual bool find(const Field& value) const = 0;
+
+    // Return owning non-null bounds, or two null Fields when the set has no non-null values.
+    // Exact range checks continue to read the typed set directly.
+    virtual void get_min_max(Field& min_value, Field& max_value) const = 0;
+    virtual bool contains_any_in_range(const Field& min_value, const Field& max_value) const = 0;
+
+    // Apply a byte-oriented predicate directly to each non-null native value and stop at the first
+    // match. The predicate and raw bytes are consumed synchronously because string sets may borrow
+    // their storage.
+    template <typename Predicate>
+    bool any_match_raw(PrimitiveType value_type, const Predicate& predicate) const {
+        return _any_match_raw(
+                value_type,
+                [](const char* data, size_t size, const void* erased_predicate) {
+                    return (*static_cast<const Predicate*>(erased_predicate))(data, size);
+                },
+                std::addressof(predicate));
+    }
 
     virtual void find_batch_raw_fixed(const uint8_t* values, size_t rows, size_t value_width,
                                       uint8_t* matches) const {
@@ -319,7 +489,7 @@ class HybridSet : public HybridSetBase {
         }
     }
 
-    int size() override { return (int)_set.size(); }
+    int size() const override { return (int)_set.size(); }
 
     bool find(const void* data) const override {
         return _set.find(*reinterpret_cast<const ElementType*>(data));
@@ -327,8 +497,51 @@ class HybridSet : public HybridSetBase {
 
     bool find(const void* data, size_t /*unused*/) const override { return find(data); }
 
-    void find_batch_raw_fixed(const uint8_t* values, size_t rows, size_t value_width,
-                              uint8_t* matches) const override {
+    bool find(const Field& value) const override {
+        DORIS_CHECK_EQ(value.get_type(), T);
+        return _set.find(value.template get<T>());
+    }
+
+    void get_min_max(Field& min_value, Field& max_value) const override {
+        detail::get_hybrid_set_min_max(
+                _set,
+                [](const ElementType& lhs, const ElementType& rhs) {
+                    return detail::hybrid_set_value_less<T>(lhs, rhs);
+                },
+                [](const ElementType& value) { return Field::create_field<T>(value); }, min_value,
+                max_value);
+    }
+
+    bool contains_any_in_range(const Field& min_value, const Field& max_value) const override {
+        DORIS_CHECK_EQ(min_value.get_type(), T);
+        DORIS_CHECK_EQ(max_value.get_type(), T);
+        const auto& typed_min = min_value.template get<T>();
+        const auto& typed_max = max_value.template get<T>();
+        const auto less = [](const ElementType& lhs, const ElementType& rhs) {
+            return detail::hybrid_set_value_less<T>(lhs, rhs);
+        };
+        DORIS_CHECK(!less(typed_max, typed_min));
+        return _set.contains_any_in_range(typed_min, typed_max, less);
+    }
+
+protected:
+    bool _any_match_raw(PrimitiveType value_type, RawValuePredicateInvoker invoke,
+                        const void* predicate) const override {
+        DORIS_CHECK_EQ(value_type, T);
+        for (const auto& value : _set) {
+            if (invoke(reinterpret_cast<const char*>(&value), sizeof(value), predicate)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+public:
+    // matches is an in/out selection mask updated in place; the checker cannot see that contract
+    // through this templated override.
+    void find_batch_raw_fixed(
+            const uint8_t* values, size_t rows, size_t value_width,
+            uint8_t* matches) const override { // NOLINT(readability-non-const-parameter)
         DORIS_CHECK_EQ(value_width, sizeof(ElementType));
         for (size_t row = 0; row < rows; ++row) {
             ElementType value;
@@ -337,8 +550,11 @@ class HybridSet : public HybridSetBase {
         }
     }
 
-    void find_batch_raw_fixed_negative(const uint8_t* values, size_t rows, size_t value_width,
-                                       uint8_t* matches) const override {
+    // matches is an in/out selection mask updated in place; the checker cannot see that contract
+    // through this templated override.
+    void find_batch_raw_fixed_negative(
+            const uint8_t* values, size_t rows, size_t value_width,
+            uint8_t* matches) const override { // NOLINT(readability-non-const-parameter)
         DORIS_CHECK_EQ(value_width, sizeof(ElementType));
         for (size_t row = 0; row < rows; ++row) {
             ElementType value;
@@ -441,7 +657,7 @@ class HybridSet : public HybridSetBase {
     uint64_t get_digest(uint64_t seed) override {
         std::vector<ElementType> elems(_set.begin(), _set.end());
         pdqsort(elems.begin(), elems.end());
-        if constexpr (std::is_same<ElementType, bool>::value) {
+        if constexpr (std::is_same_v<ElementType, bool>) {
             for (bool v : elems) {
                 seed = HashUtil::crc_hash64(&v, sizeof(v), seed);
             }
@@ -458,7 +674,8 @@ class HybridSet : public HybridSetBase {
     ObjectPool _pool;
 };
 
-template <typename _ContainerType = DynamicContainer<std::string>>
+template <typename _ContainerType =
+                  DynamicContainer<std::string, HybridSetStringHash, HybridSetStringEqual>>
 class StringSet : public HybridSetBase {
 public:
     using ContainerType = _ContainerType;
@@ -533,19 +750,67 @@ class StringSet : public HybridSetBase {
         }
     }
 
-    int size() override { return (int)_set.size(); }
+    int size() const override { return (int)_set.size(); }
 
     bool find(const void* data) const override {
         const auto* value = reinterpret_cast<const StringRef*>(data);
-        std::string str_value(value->data, value->size);
-        return _set.find(str_value);
+        if constexpr (requires { _set.find(*value); }) {
+            return _set.find(*value);
+        } else {
+            return _set.find(std::string(value->data, value->size));
+        }
     }
 
     bool find(const void* data, size_t size) const override {
-        std::string str_value(reinterpret_cast<const char*>(data), size);
-        return _set.find(str_value);
+        const StringRef value(reinterpret_cast<const char*>(data), size);
+        if constexpr (requires { _set.find(value); }) {
+            return _set.find(value);
+        } else {
+            return _set.find(std::string(value.data, value.size));
+        }
     }
 
+    bool find(const Field& value) const override {
+        DORIS_CHECK(is_string_type(value.get_type()));
+        const auto string_value = value.as_string_view();
+        return find(string_value.data(), string_value.size());
+    }
+
+    void get_min_max(Field& min_value, Field& max_value) const override {
+        detail::get_hybrid_set_min_max(
+                _set, [](const std::string& lhs, const std::string& rhs) { return lhs < rhs; },
+                [](const std::string& value) {
+                    return Field::create_field<TYPE_STRING>(String(value.data(), value.size()));
+                },
+                min_value, max_value);
+    }
+
+    bool contains_any_in_range(const Field& min_value, const Field& max_value) const override {
+        DORIS_CHECK(is_string_type(min_value.get_type()));
+        DORIS_CHECK(is_string_type(max_value.get_type()));
+        const auto typed_min = min_value.as_string_view();
+        const auto typed_max = max_value.as_string_view();
+        DORIS_CHECK(typed_min <= typed_max);
+        return std::ranges::any_of(_set, [&](const std::string& value) {
+            const std::string_view typed_value(value.data(), value.size());
+            return typed_value >= typed_min && typed_value <= typed_max;
+        });
+    }
+
+protected:
+    bool _any_match_raw(PrimitiveType value_type, RawValuePredicateInvoker invoke,
+                        const void* predicate) const override {
+        DORIS_CHECK(is_string_type(value_type));
+        for (const auto& value : _set) {
+            const char* data = value.empty() ? "" : value.data();
+            if (invoke(data, value.size(), predicate)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+public:
     void find_batch(const doris::IColumn& column, size_t rows,
                     doris::ColumnUInt8::Container& results,
                     const uint8_t* __restrict filter = nullptr) override {
@@ -587,15 +852,15 @@ class StringSet : public HybridSetBase {
         auto* __restrict result_data = results.data();
 
         auto update_value = [&](size_t i) {
-            const auto& string_data = col.get_data_at(i).to_string();
+            const auto string_data = col.get_data_at(i);
             if constexpr (!is_nullable && !is_negative) {
-                result_data[i] = _set.find(string_data);
+                result_data[i] = find(string_data.data, string_data.size);
             } else if constexpr (!is_nullable && is_negative) {
-                result_data[i] = !_set.find(string_data);
+                result_data[i] = !find(string_data.data, string_data.size);
             } else if constexpr (is_nullable && !is_negative) {
-                result_data[i] = _set.find(string_data) & (!null_map_data[i]);
+                result_data[i] = find(string_data.data, string_data.size) & (!null_map_data[i]);
             } else { // (is_nullable && is_negative)
-                result_data[i] = !(_set.find(string_data) & (!null_map_data[i]));
+                result_data[i] = !(find(string_data.data, string_data.size) & (!null_map_data[i]));
             }
         };
 
@@ -736,7 +1001,7 @@ class StringValueSet : public HybridSetBase {
         }
     }
 
-    int size() override { return (int)_set.size(); }
+    int size() const override { return (int)_set.size(); }
 
     bool find(const void* data) const override {
         const auto* value = reinterpret_cast<const StringRef*>(data);
@@ -748,6 +1013,48 @@ class StringValueSet : public HybridSetBase {
         return _set.find(sv);
     }
 
+    bool find(const Field& value) const override {
+        DORIS_CHECK(is_string_type(value.get_type()));
+        const auto string_value = value.as_string_view();
+        return find(string_value.data(), string_value.size());
+    }
+
+    void get_min_max(Field& min_value, Field& max_value) const override {
+        detail::get_hybrid_set_min_max(
+                _set, [](const StringRef& lhs, const StringRef& rhs) { return lhs < rhs; },
+                [](const StringRef& value) {
+                    return Field::create_field<TYPE_STRING>(
+                            String(value.size == 0 ? "" : value.data, value.size));
+                },
+                min_value, max_value);
+    }
+
+    bool contains_any_in_range(const Field& min_value, const Field& max_value) const override {
+        DORIS_CHECK(is_string_type(min_value.get_type()));
+        DORIS_CHECK(is_string_type(max_value.get_type()));
+        const auto typed_min = min_value.as_string_view();
+        const auto typed_max = max_value.as_string_view();
+        DORIS_CHECK(typed_min <= typed_max);
+        return std::ranges::any_of(_set, [&](const StringRef& value) {
+            const std::string_view typed_value(value.size == 0 ? "" : value.data, value.size);
+            return typed_value >= typed_min && typed_value <= typed_max;
+        });
+    }
+
+protected:
+    bool _any_match_raw(PrimitiveType value_type, RawValuePredicateInvoker invoke,
+                        const void* predicate) const override {
+        DORIS_CHECK(is_string_type(value_type));
+        for (const auto& value : _set) {
+            const char* data = value.size == 0 ? "" : value.data;
+            if (invoke(data, value.size, predicate)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+public:
     void find_batch(const doris::IColumn& column, size_t rows,
                     doris::ColumnUInt8::Container& results,
                     const uint8_t* __restrict filter) override {
diff --git a/be/src/exprs/hybrid_set_min_max.h b/be/src/exprs/hybrid_set_min_max.h
new file mode 100644
index 00000000000000..1cfc5cb5ed476e
--- /dev/null
+++ b/be/src/exprs/hybrid_set_min_max.h
@@ -0,0 +1,30 @@
+// 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.
+
+#pragma once
+
+#include "core/field.h"
+
+namespace doris {
+
+// Owning non-null bounds used by zonemap pruning. Empty and null-only sets leave both Fields null.
+struct HybridSetMinMax {
+    Field min_value;
+    Field max_value;
+};
+
+} // namespace doris
diff --git a/be/src/exprs/vdirect_in_predicate.h b/be/src/exprs/vdirect_in_predicate.h
index 693571d25bf777..62f9a7c61e835c 100644
--- a/be/src/exprs/vdirect_in_predicate.h
+++ b/be/src/exprs/vdirect_in_predicate.h
@@ -17,9 +17,8 @@
 
 #pragma once
 
-#include <mutex>
+#include <memory>
 #include <utility>
-#include <vector>
 
 #include "common/logging.h"
 #include "common/status.h"
@@ -27,6 +26,7 @@
 #include "core/types.h"
 #include "exprs/expr_zonemap_filter.h"
 #include "exprs/hybrid_set.h"
+#include "exprs/hybrid_set_min_max.h"
 #include "exprs/vexpr.h"
 #include "exprs/vin_predicate.h"
 #include "exprs/vliteral.h"
@@ -37,15 +37,6 @@ namespace doris {
 class VDirectInPredicate final : public VExpr {
     ENABLE_FACTORY_CREATOR(VDirectInPredicate);
 
-    struct PruningState {
-        std::once_flag materialize_once;
-        Status materialization_status;
-        bool zonemap_materialized = false;
-        std::vector<Field> seg_filter_values;
-        Field seg_filter_min;
-        Field seg_filter_max;
-    };
-
 public:
     // `hybrid_set_values_match_child_type` tells whether values in `filter` can be interpreted with
     // the child expression type. Parquet/ORC dictionary-filter rewrites evaluate the original
@@ -53,7 +44,7 @@ class VDirectInPredicate final : public VExpr {
     // dictionary codes, for example `col IN ('a', 'b')` becomes `dict_code IN (0, 1)`. In that
     // shape the HybridSet stores TYPE_INT dictionary codes while the child slot still has the
     // original logical type such as STRING. Callers must pass false to disable zonemap
-    // materialization and slot-IN rewrite that would otherwise rebuild child-typed literals from
+    // min/max preparation and slot-IN rewrite that would otherwise rebuild child-typed literals from
     // dictionary codes.
     VDirectInPredicate(const TExprNode& node, const std::shared_ptr<HybridSetBase>& filter,
                        bool hybrid_set_values_match_child_type = true)
@@ -70,7 +61,7 @@ class VDirectInPredicate final : public VExpr {
     Status prepare(RuntimeState* state, const RowDescriptor& row_desc,
                    VExprContext* context) override {
         RETURN_IF_ERROR_OR_PREPARED(VExpr::prepare(state, row_desc, context));
-        RETURN_IF_ERROR(_materialize_for_zonemap_filter());
+        _prepare_zonemap_min_max();
         _prepare_finished = true;
         return Status::OK();
     }
@@ -99,24 +90,24 @@ class VDirectInPredicate final : public VExpr {
     std::shared_ptr<HybridSetBase> get_set_func() const override { return _filter; }
 
     ZoneMapFilterResult evaluate_zonemap_filter(const ZoneMapEvalContext& ctx) const override {
-        return expr_zonemap::eval_in_zonemap(
-                ctx, get_child(0), false, _pruning_state->seg_filter_values,
-                _pruning_state->seg_filter_min, _pruning_state->seg_filter_max);
+        DORIS_CHECK(_zonemap_min_max != nullptr);
+        DORIS_CHECK(_filter != nullptr);
+        return expr_zonemap::eval_in_zonemap(ctx, get_child(0), false, *_zonemap_min_max, *_filter);
     }
 
     bool can_evaluate_zonemap_filter() const override {
-        return _pruning_state->zonemap_materialized &&
+        return _zonemap_min_max != nullptr &&
                std::dynamic_pointer_cast<VSlotRef>(get_child(0)) != nullptr;
     }
 
     ZoneMapFilterResult evaluate_dictionary_filter(
             const DictionaryEvalContext& ctx) const override {
-        return expr_zonemap::eval_in_dictionary(ctx, get_child(0), false,
-                                                _pruning_state->seg_filter_values);
+        DORIS_CHECK(_filter != nullptr);
+        return expr_zonemap::eval_in_dictionary(ctx, get_child(0), false, *_filter);
     }
 
     bool can_evaluate_dictionary_filter() const override {
-        return _pruning_state->zonemap_materialized &&
+        return _zonemap_min_max != nullptr &&
                std::dynamic_pointer_cast<VSlotRef>(get_child(0)) != nullptr;
     }
 
@@ -137,9 +128,11 @@ class VDirectInPredicate final : public VExpr {
         return _raw_fixed_value_size(raw_type->get_primitive_type()) != 0;
     }
 
-    Status execute_on_raw_fixed_values(const uint8_t* values, size_t num_values, size_t value_width,
-                                       const DataTypePtr& data_type, int column_id,
-                                       uint8_t* matches) const override {
+    // matches is an in/out selection mask forwarded to the HybridSet batch probe.
+    Status execute_on_raw_fixed_values(
+            const uint8_t* values, size_t num_values, size_t value_width,
+            const DataTypePtr& data_type, int column_id,
+            uint8_t* matches) const override { // NOLINT(readability-non-const-parameter)
         if (!can_execute_on_raw_fixed_values(data_type, column_id)) {
             return Status::NotSupported(
                     "Direct IN predicate cannot evaluate raw fixed-width values");
@@ -172,9 +165,10 @@ class VDirectInPredicate final : public VExpr {
                is_string_type(remove_nullable(slot->data_type())->get_primitive_type());
     }
 
-    Status execute_on_raw_binary_values(const StringRef* values, size_t num_values,
-                                        const DataTypePtr& data_type, int column_id,
-                                        uint8_t* matches) const override {
+    // matches is an in/out selection mask forwarded to the HybridSet batch probe.
+    Status execute_on_raw_binary_values(
+            const StringRef* values, size_t num_values, const DataTypePtr& data_type, int column_id,
+            uint8_t* matches) const override { // NOLINT(readability-non-const-parameter)
         if (!can_execute_on_raw_binary_values(data_type, column_id)) {
             return Status::NotSupported("Direct IN predicate cannot evaluate raw binary values");
         }
@@ -190,9 +184,9 @@ class VDirectInPredicate final : public VExpr {
         DORIS_CHECK(cloned_expr != nullptr);
         auto cloned = VDirectInPredicate::create_shared(clone_texpr_node(), _filter,
                                                         _hybrid_set_values_match_child_type);
-        // Runtime-filter sets are immutable after publication, and file-local rewrites preserve
-        // the predicate's logical child type, so every split clone must reuse this materialization.
-        cloned->_pruning_state = _pruning_state;
+        // clone_node is never concurrent with prepare. Copy already-prepared bounds so file-local
+        // split clones can prune without traversing the set again.
+        cloned->_zonemap_min_max = _zonemap_min_max;
         *cloned_expr = std::move(cloned);
         return Status::OK();
     }
@@ -279,9 +273,10 @@ class VDirectInPredicate final : public VExpr {
         }
     }
 
+    // arg_column optionally returns the fully materialized argument column to the caller.
     Status _do_execute(VExprContext* context, const Block* block, const uint8_t* __restrict filter,
                        const Selector* selector, size_t count, ColumnPtr& result_column,
-                       ColumnPtr* arg_column) const {
+                       ColumnPtr* arg_column) const { // NOLINT(readability-non-const-parameter)
         DCHECK(_open_finished || block == nullptr);
         DCHECK(!(filter != nullptr && selector != nullptr))
                 << "filter and selector can not be both set";
@@ -312,37 +307,29 @@ class VDirectInPredicate final : public VExpr {
         return Status::OK();
     }
 
-    Status _materialize_for_zonemap_filter() {
-        const auto pruning_state = _pruning_state;
-        std::call_once(pruning_state->materialize_once, [&] {
-            if (!_hybrid_set_values_match_child_type) {
-                return;
-            }
-            DORIS_CHECK(_filter != nullptr);
-            auto& filter = *_filter;
-            const auto& data_type = remove_nullable(get_child(0)->data_type());
-            expr_zonemap::InZonemapMaterializedSet materialized;
-            pruning_state->materialization_status =
-                    expr_zonemap::materialize_hybrid_set_for_zonemap_filter(filter, data_type,
-                                                                            &materialized);
-            if (!pruning_state->materialization_status.ok()) {
-                return;
-            }
-            pruning_state->seg_filter_values = std::move(materialized.values);
-            pruning_state->seg_filter_min = std::move(materialized.min_value);
-            pruning_state->seg_filter_max = std::move(materialized.max_value);
-            pruning_state->zonemap_materialized = true;
-        });
-        return pruning_state->materialization_status;
+    void _prepare_zonemap_min_max() {
+        if (!_hybrid_set_values_match_child_type) {
+            _zonemap_min_max.reset();
+            return;
+        }
+        if (_zonemap_min_max != nullptr) {
+            return;
+        }
+        DORIS_CHECK(_filter != nullptr);
+        const auto& data_type = remove_nullable(get_child(0)->data_type());
+        auto zonemap_min_max = std::make_shared<HybridSetMinMax>();
+        expr_zonemap::get_hybrid_set_min_max_for_zonemap_filter(_filter, data_type,
+                                                                *zonemap_min_max);
+        _zonemap_min_max = std::move(zonemap_min_max);
     }
 
     std::shared_ptr<HybridSetBase> _filter;
     // Dictionary-filter rewrites may store physical dictionary codes in the HybridSet while the
-    // child slot keeps the original logical type. Such values must not be materialized as child-type
-    // literals for zonemap pruning or slot-IN rewrite.
+    // child slot keeps the original logical type. Such values must not be interpreted as child-type
+    // bounds for zonemap pruning or literals for slot-IN rewrite.
     bool _hybrid_set_values_match_child_type = true;
     std::string _expr_name;
-    std::shared_ptr<PruningState> _pruning_state = std::make_shared<PruningState>();
+    std::shared_ptr<const HybridSetMinMax> _zonemap_min_max;
 };
 
 } // namespace doris
diff --git a/be/src/exprs/vin_predicate.cpp b/be/src/exprs/vin_predicate.cpp
index 0dbfb3f7e9a289..1134dac2b84024 100644
--- a/be/src/exprs/vin_predicate.cpp
+++ b/be/src/exprs/vin_predicate.cpp
@@ -34,6 +34,8 @@
 #include "exprs/expr_zonemap_filter.h"
 #include "exprs/function/in.h"
 #include "exprs/function/simple_function_factory.h"
+#include "exprs/hybrid_set.h"
+#include "exprs/hybrid_set_min_max.h"
 #include "exprs/vexpr_context.h"
 #include "exprs/vslot_ref.h"
 #include "runtime/runtime_state.h"
@@ -84,6 +86,10 @@ size_t raw_in_value_size(PrimitiveType primitive_type) {
 VInPredicate::VInPredicate(const TExprNode& node)
         : VExpr(node), _is_not_in(node.in_predicate.is_not_in) {}
 
+#ifdef BE_TEST
+VInPredicate::VInPredicate() = default;
+#endif
+
 Status VInPredicate::prepare(RuntimeState* state, const RowDescriptor& desc,
                              VExprContext* context) {
     RETURN_IF_ERROR_OR_PREPARED(VExpr::prepare(state, desc, context));
@@ -137,8 +143,8 @@ Status VInPredicate::open(RuntimeState* state, VExprContext* context,
     _is_args_all_constant = std::all_of(_children.begin() + 1, _children.end(),
                                         [](const VExprSPtr& expr) { return expr->is_constant(); });
     if (scope == FunctionContext::FRAGMENT_LOCAL && _is_args_all_constant &&
-        !_zonemap_materialized) {
-        RETURN_IF_ERROR(_materialize_for_zonemap_filter(context));
+        _zonemap_min_max == nullptr) {
+        _prepare_zonemap_min_max(context);
     }
     _open_finished = true;
     return Status::OK();
@@ -154,19 +160,17 @@ Status VInPredicate::evaluate_inverted_index(VExprContext* context, uint32_t seg
     return _evaluate_inverted_index(context, _function, segment_num_rows);
 }
 
-Status VInPredicate::_materialize_for_zonemap_filter(VExprContext* context) {
-    _seg_filter_values.clear();
-    _seg_filter_contains_null = false;
-    _zonemap_materialized = false;
+void VInPredicate::_prepare_zonemap_min_max(VExprContext* context) {
+    _zonemap_min_max.reset();
     _direct_filter_set.reset();
     if (_children.size() < 2 || !_children[0]->is_slot_ref()) {
-        return Status::OK();
+        return;
     }
 
     const auto data_type = remove_nullable(_children[0]->data_type());
     DORIS_CHECK(data_type != nullptr);
     if (is_complex_type(data_type->get_primitive_type())) {
-        return Status::OK();
+        return;
     }
 
     DORIS_CHECK(context != nullptr);
@@ -179,51 +183,48 @@ Status VInPredicate::_materialize_for_zonemap_filter(VExprContext* context) {
     DORIS_CHECK(in_state->hybrid_set != nullptr);
     _direct_filter_set = in_state->hybrid_set;
 
-    expr_zonemap::InZonemapMaterializedSet materialized;
-    RETURN_IF_ERROR(expr_zonemap::materialize_hybrid_set_for_zonemap_filter(
-            *in_state->hybrid_set, data_type, &materialized));
-    _seg_filter_contains_null = materialized.contains_null;
-    _seg_filter_values = std::move(materialized.values);
-    _seg_filter_min = std::move(materialized.min_value);
-    _seg_filter_max = std::move(materialized.max_value);
-    _zonemap_materialized = true;
-    return Status::OK();
+    auto zonemap_min_max = std::make_shared<HybridSetMinMax>();
+    expr_zonemap::get_hybrid_set_min_max_for_zonemap_filter(in_state->hybrid_set, data_type,
+                                                            *zonemap_min_max);
+    _zonemap_min_max = std::move(zonemap_min_max);
 }
 
 ZoneMapFilterResult VInPredicate::evaluate_zonemap_filter(const ZoneMapEvalContext& ctx) const {
-    if (_is_not_in && _seg_filter_contains_null) {
-        return ZoneMapFilterResult::kNoMatch;
-    }
-    return expr_zonemap::eval_in_zonemap(ctx, get_child(0), _is_not_in, _seg_filter_values,
-                                         _seg_filter_min, _seg_filter_max);
+    DORIS_CHECK(_zonemap_min_max != nullptr);
+    DORIS_CHECK(_direct_filter_set != nullptr);
+    return expr_zonemap::eval_in_zonemap(ctx, get_child(0), _is_not_in, *_zonemap_min_max,
+                                         *_direct_filter_set);
 }
 
 bool VInPredicate::can_evaluate_zonemap_filter() const {
-    return _zonemap_materialized && std::dynamic_pointer_cast<VSlotRef>(get_child(0)) != nullptr;
+    return _zonemap_min_max != nullptr &&
+           std::dynamic_pointer_cast<VSlotRef>(get_child(0)) != nullptr;
 }
 
 ZoneMapFilterResult VInPredicate::evaluate_dictionary_filter(
         const DictionaryEvalContext& ctx) const {
-    return expr_zonemap::eval_in_dictionary(ctx, get_child(0), _is_not_in, _seg_filter_values);
+    DORIS_CHECK(_direct_filter_set != nullptr);
+    return expr_zonemap::eval_in_dictionary(ctx, get_child(0), _is_not_in, *_direct_filter_set);
 }
 
 bool VInPredicate::can_evaluate_dictionary_filter() const {
-    return _zonemap_materialized && !_is_not_in &&
+    return _zonemap_min_max != nullptr && !_is_not_in &&
            std::dynamic_pointer_cast<VSlotRef>(get_child(0)) != nullptr;
 }
 
 ZoneMapFilterResult VInPredicate::evaluate_bloom_filter(const BloomFilterEvalContext& ctx) const {
-    return expr_zonemap::eval_in_bloom_filter(ctx, get_child(0), _is_not_in, _seg_filter_values);
+    DORIS_CHECK(_direct_filter_set != nullptr);
+    return expr_zonemap::eval_in_bloom_filter(ctx, get_child(0), _is_not_in, *_direct_filter_set);
 }
 
 bool VInPredicate::can_evaluate_bloom_filter() const {
-    return _zonemap_materialized && !_is_not_in &&
+    return _zonemap_min_max != nullptr && !_is_not_in &&
            std::dynamic_pointer_cast<VSlotRef>(get_child(0)) != nullptr;
 }
 
 bool VInPredicate::can_execute_on_raw_fixed_values(const DataTypePtr& data_type,
                                                    int column_id) const {
-    if (!_zonemap_materialized || _direct_filter_set == nullptr || data_type == nullptr ||
+    if (_zonemap_min_max == nullptr || _direct_filter_set == nullptr || data_type == nullptr ||
         !_is_args_all_constant) {
         return false;
     }
@@ -252,7 +253,7 @@ Status VInPredicate::execute_on_raw_fixed_values(const uint8_t* values, size_t n
     }
     // NOT IN with a NULL literal is UNKNOWN for every non-null physical value. Definition levels
     // handle input NULLs, while this guard preserves the remaining three-valued SQL invariant.
-    if (_is_not_in && _seg_filter_contains_null) {
+    if (_is_not_in && _direct_filter_set->contain_null()) {
         std::fill(matches, matches + num_values, uint8_t {0});
     } else if (_is_not_in) {
         _direct_filter_set->find_batch_raw_fixed_negative(values, num_values, value_width, matches);
@@ -264,7 +265,7 @@ Status VInPredicate::execute_on_raw_fixed_values(const uint8_t* values, size_t n
 
 bool VInPredicate::can_execute_on_raw_binary_values(const DataTypePtr& data_type,
                                                     int column_id) const {
-    if (!_zonemap_materialized || _direct_filter_set == nullptr || data_type == nullptr ||
+    if (_zonemap_min_max == nullptr || _direct_filter_set == nullptr || data_type == nullptr ||
         !_is_args_all_constant ||
         !is_string_type(remove_nullable(data_type)->get_primitive_type())) {
         return false;
@@ -282,7 +283,7 @@ Status VInPredicate::execute_on_raw_binary_values(const StringRef* values, size_
     }
     DORIS_CHECK(values != nullptr || num_values == 0);
     DORIS_CHECK(matches != nullptr || num_values == 0);
-    if (_is_not_in && _seg_filter_contains_null) {
+    if (_is_not_in && _direct_filter_set->contain_null()) {
         std::fill(matches, matches + num_values, uint8_t {0});
     } else if (_is_not_in) {
         _direct_filter_set->find_batch_raw_binary_negative(values, num_values, matches);
diff --git a/be/src/exprs/vin_predicate.h b/be/src/exprs/vin_predicate.h
index 299a69f46a49b9..78a733708c2c82 100644
--- a/be/src/exprs/vin_predicate.h
+++ b/be/src/exprs/vin_predicate.h
@@ -17,12 +17,12 @@
 
 #pragma once
 
+#include <memory>
 #include <string>
-#include <vector>
+#include <utility>
 
 #include "common/object_pool.h"
 #include "common/status.h"
-#include "core/field.h"
 #include "exprs/function/function.h"
 #include "exprs/function_context.h"
 #include "exprs/vexpr.h"
@@ -34,6 +34,7 @@ class TExprNode;
 class Block;
 class VExprContext;
 class HybridSetBase;
+struct HybridSetMinMax;
 } // namespace doris
 
 namespace doris {
@@ -43,7 +44,7 @@ class VInPredicate MOCK_REMOVE(final) : public VExpr {
 public:
     VInPredicate(const TExprNode& node);
 #ifdef BE_TEST
-    VInPredicate() = default;
+    VInPredicate();
 #endif
     ~VInPredicate() override = default;
     Status execute_column_impl(VExprContext* context, const Block* block, const Selector* selector,
@@ -57,7 +58,7 @@ class VInPredicate MOCK_REMOVE(final) : public VExpr {
 
     std::string debug_string() const override;
 
-    const FunctionBasePtr function() { return _function; }
+    FunctionBasePtr function() const { return _function; }
 
     bool is_not_in() const { return _is_not_in; };
     Status evaluate_inverted_index(VExprContext* context, uint32_t segment_num_rows) override;
@@ -91,7 +92,7 @@ class VInPredicate MOCK_REMOVE(final) : public VExpr {
     }
 
 private:
-    Status _materialize_for_zonemap_filter(VExprContext* context);
+    void _prepare_zonemap_min_max(VExprContext* context);
 
     FunctionBasePtr _function;
     std::string _expr_name;
@@ -100,11 +101,7 @@ class VInPredicate MOCK_REMOVE(final) : public VExpr {
     static const constexpr char* function_name = "in";
     uint32_t _in_list_value_count_threshold = 10;
     bool _is_args_all_constant = false;
-    bool _zonemap_materialized = false;
-    bool _seg_filter_contains_null = false;
     std::shared_ptr<HybridSetBase> _direct_filter_set;
-    std::vector<Field> _seg_filter_values;
-    Field _seg_filter_min;
-    Field _seg_filter_max;
+    std::shared_ptr<const HybridSetMinMax> _zonemap_min_max;
 };
 } // namespace doris
diff --git a/be/src/storage/predicate/in_list_predicate.h b/be/src/storage/predicate/in_list_predicate.h
index 47e1a9e9cd111f..a0de781415bca5 100644
--- a/be/src/storage/predicate/in_list_predicate.h
+++ b/be/src/storage/predicate/in_list_predicate.h
@@ -78,7 +78,7 @@ class InListPredicateBase final : public ColumnPredicate {
         CHECK(hybrid_set != nullptr);
 
         // String types need a copy because:
-        // The caller's set is StringSet<DynamicContainer<std::string>>, but here we want
+        // The caller's set is an owning dynamic StringSet, but here we want
         // StringSet<FixedContainer<std::string, N>> for small-set optimization — different
         // C++ types, cannot share the pointer.
         //
diff --git a/be/test/exprs/expr_zonemap_filter_test.cpp b/be/test/exprs/expr_zonemap_filter_test.cpp
index 72981c01275c25..cda94fcccbcaa4 100644
--- a/be/test/exprs/expr_zonemap_filter_test.cpp
+++ b/be/test/exprs/expr_zonemap_filter_test.cpp
@@ -20,9 +20,9 @@
 #include <gtest/gtest.h>
 
 #include <array>
+#include <limits>
 #include <map>
 #include <memory>
-#include <mutex>
 #include <set>
 #include <string>
 #include <vector>
@@ -34,6 +34,7 @@
 #include "core/data_type/data_type_nullable.h"
 #include "core/data_type/data_type_number.h"
 #include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_time.h"
 #include "core/field.h"
 #include "core/string_ref.h"
 #include "core/value/vdatetime_value.h"
@@ -42,6 +43,7 @@
 #include "exprs/function/functions_comparison.h"
 #include "exprs/function/simple_function_factory.h"
 #include "exprs/hybrid_set.h"
+#include "exprs/hybrid_set_min_max.h"
 #include "exprs/runtime_filter_expr.h"
 #include "exprs/vbloom_predicate.h"
 #include "exprs/vcompound_pred.h"
@@ -154,6 +156,16 @@ std::unique_ptr<segment_v2::BlockSplitBloomFilter> make_int_bloom_filter(
     return bloom_filter;
 }
 
+std::unique_ptr<segment_v2::BlockSplitBloomFilter> make_string_bloom_filter(
+        const std::vector<std::string>& values) {
+    auto bloom_filter = std::make_unique<segment_v2::BlockSplitBloomFilter>();
+    EXPECT_TRUE(bloom_filter->init(segment_v2::BloomFilter::MINIMUM_BYTES).ok());
+    for (const auto& value : values) {
+        bloom_filter->add_bytes(value.data(), value.size());
+    }
+    return bloom_filter;
+}
+
 BloomFilterEvalContext make_bloom_filter_context(const segment_v2::BloomFilter* bloom_filter,
                                                  const DataTypePtr& data_type) {
     BloomFilterEvalContext ctx;
@@ -164,6 +176,53 @@ BloomFilterEvalContext make_bloom_filter_context(const segment_v2::BloomFilter*
     return ctx;
 }
 
+struct MinMaxTestSet {
+    std::shared_ptr<HybridSetBase> set;
+    HybridSetMinMax min_max;
+};
+
+MinMaxTestSet make_int_set_with_min_max(const std::vector<int32_t>& values, bool null_aware = false,
+                                        bool contains_null = false) {
+    MinMaxTestSet result;
+    result.set.reset(create_set(TYPE_INT, null_aware));
+    for (const auto value : values) {
+        result.set->insert(&value);
+    }
+    if (contains_null) {
+        result.set->insert(static_cast<const void*>(nullptr));
+    }
+    expr_zonemap::get_hybrid_set_min_max_for_zonemap_filter(result.set, int_type(), result.min_max);
+    return result;
+}
+
+template <PrimitiveType T>
+MinMaxTestSet make_typed_set_with_min_max(
+        const std::vector<typename PrimitiveTypeTraits<T>::CppType>& values,
+        const DataTypePtr& data_type, bool null_aware = false, bool contains_null = false) {
+    MinMaxTestSet result;
+    result.set.reset(create_set(T, null_aware));
+    for (const auto& value : values) {
+        result.set->insert(&value);
+    }
+    if (contains_null) {
+        result.set->insert(static_cast<const void*>(nullptr));
+    }
+    expr_zonemap::get_hybrid_set_min_max_for_zonemap_filter(result.set, data_type, result.min_max);
+    return result;
+}
+
+MinMaxTestSet make_string_set_with_min_max(const std::vector<std::string>& values,
+                                           const DataTypePtr& data_type) {
+    MinMaxTestSet result;
+    result.set.reset(create_set(TYPE_STRING, false));
+    for (const auto& value : values) {
+        StringRef string_value(value);
+        result.set->insert(&string_value);
+    }
+    expr_zonemap::get_hybrid_set_min_max_for_zonemap_filter(result.set, data_type, result.min_max);
+    return result;
+}
+
 segment_v2::ZoneMap make_int_zonemap(int32_t min_value, int32_t max_value) {
     segment_v2::ZoneMap zone_map;
     zone_map.min_value = int_field(min_value);
@@ -180,11 +239,11 @@ segment_v2::ZoneMap make_string_zonemap(std::string min_value, std::string max_v
     return zone_map;
 }
 
-TDescriptorTable make_k2_scan_desc_tbl() {
+TDescriptorTable make_k2_scan_desc_tbl(PrimitiveType primitive_type = TYPE_INT) {
     TDescriptorTableBuilder desc_tbl_builder;
     TTupleDescriptorBuilder tuple_builder;
     auto k2_slot = TSlotDescriptorBuilder()
-                           .type(TYPE_INT)
+                           .type(primitive_type)
                            .column_name("k2")
                            .column_pos(0)
                            .nullable(false)
@@ -475,11 +534,10 @@ TEST(ExprZonemapFilterTest, MissingSlotTypeCountsUnsupportedZonemapEvalOnce) {
                                                    {string_slot, make_string_literal("ab")}));
     EXPECT_EQ(1, starts_with_ctx.stats.unusable_zonemap_eval_count);
 
-    std::vector<Field> values {int_field(10)};
+    auto values = make_int_set_with_min_max({10});
     ZoneMapEvalContext in_ctx;
     EXPECT_EQ(ZoneMapFilterResult::kUnsupported,
-              expr_zonemap::eval_in_zonemap(in_ctx, slot, false, values, int_field(10),
-                                            int_field(10)));
+              expr_zonemap::eval_in_zonemap(in_ctx, slot, false, values.min_max, *values.set));
     EXPECT_EQ(1, in_ctx.stats.unusable_zonemap_eval_count);
 }
 
@@ -536,26 +594,24 @@ TEST(ExprZonemapFilterTest, RangeStatsUnusableFlagsFallback) {
 TEST(ExprZonemapFilterTest, InZonemapSkipsZonesWithoutNonNullValues) {
     auto type = int_type();
     auto slot = make_slot(0, type);
-    std::vector<Field> values {int_field(10)};
+    auto values = make_int_set_with_min_max({10});
 
     segment_v2::ZoneMap empty_zone;
     auto empty_ctx = make_context(empty_zone, type);
     EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
-              expr_zonemap::eval_in_zonemap(empty_ctx, slot, false, values, int_field(10),
-                                            int_field(10)));
+              expr_zonemap::eval_in_zonemap(empty_ctx, slot, false, values.min_max, *values.set));
     EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
-              expr_zonemap::eval_in_zonemap(empty_ctx, slot, true, values, int_field(10),
-                                            int_field(10)));
+              expr_zonemap::eval_in_zonemap(empty_ctx, slot, true, values.min_max, *values.set));
 
     segment_v2::ZoneMap only_null_zone;
     only_null_zone.has_null = true;
     auto only_null_ctx = make_context(only_null_zone, type);
-    EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
-              expr_zonemap::eval_in_zonemap(only_null_ctx, slot, false, values, int_field(10),
-                                            int_field(10)));
-    EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
-              expr_zonemap::eval_in_zonemap(only_null_ctx, slot, true, values, int_field(10),
-                                            int_field(10)));
+    EXPECT_EQ(
+            ZoneMapFilterResult::kNoMatch,
+            expr_zonemap::eval_in_zonemap(only_null_ctx, slot, false, values.min_max, *values.set));
+    EXPECT_EQ(
+            ZoneMapFilterResult::kNoMatch,
+            expr_zonemap::eval_in_zonemap(only_null_ctx, slot, true, values.min_max, *values.set));
 }
 
 TEST(ExprZonemapFilterTest, FunctionStringStartsWithZonemapUsesPrefixRange) {
@@ -641,11 +697,10 @@ TEST(ExprZonemapFilterTest, CharZonemapUsesTrimmedLogicalBounds) {
               starts_with->evaluate_zonemap_filter(starts_with_ctx,
                                                    {slot, make_string_literal("ga")}));
 
-    auto in_value = Field::create_field<TYPE_STRING>("gamma");
-    std::vector<Field> values {in_value};
+    auto values = make_string_set_with_min_max({"gamma"}, char_type);
     auto in_ctx = make_context(zone_map, char_type);
     EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
-              expr_zonemap::eval_in_zonemap(in_ctx, slot, false, values, in_value, in_value));
+              expr_zonemap::eval_in_zonemap(in_ctx, slot, false, values.min_max, *values.set));
 }
 
 TEST(ExprZonemapFilterTest, InZonemapFallsBackToRangeWhenPointListIsLarge) {
@@ -653,12 +708,26 @@ TEST(ExprZonemapFilterTest, InZonemapFallsBackToRangeWhenPointListIsLarge) {
     auto slot = make_slot(0, type);
     auto ctx = make_context(make_int_zonemap(10, 20), type);
 
-    std::vector<Field> values;
+    std::vector<int32_t> values;
     for (int value = 1; value <= 65; ++value) {
-        values.emplace_back(int_field(value));
+        values.emplace_back(value);
     }
+    auto values_with_min_max = make_int_set_with_min_max(values);
+    EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
+              expr_zonemap::eval_in_zonemap(ctx, slot, false, values_with_min_max.min_max,
+                                            *values_with_min_max.set));
+    EXPECT_EQ(65, values_with_min_max.set->size());
     EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
-              expr_zonemap::eval_in_zonemap(ctx, slot, false, values, int_field(1), int_field(65)));
+              expr_zonemap::eval_in_dictionary(make_dictionary_context({int_field(65)}, type), slot,
+                                               false, *values_with_min_max.set));
+    EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
+              expr_zonemap::eval_in_dictionary(make_dictionary_context({int_field(100)}, type),
+                                               slot, false, *values_with_min_max.set));
+
+    auto singleton_ctx = make_context(make_int_zonemap(10, 10), type);
+    EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
+              expr_zonemap::eval_in_zonemap(singleton_ctx, slot, true, values_with_min_max.min_max,
+                                            *values_with_min_max.set));
 
     EXPECT_EQ(0, ctx.stats.in_zonemap_point_check_count);
     EXPECT_EQ(1, ctx.stats.in_zonemap_range_only_count);
@@ -669,9 +738,9 @@ TEST(ExprZonemapFilterTest, InZonemapUsesPointChecksUnderThreshold) {
     auto slot = make_slot(0, type);
     auto ctx = make_context(make_int_zonemap(10, 20), type);
 
-    std::vector<Field> values {int_field(1), int_field(30)};
+    auto values = make_int_set_with_min_max({1, 30});
     EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
-              expr_zonemap::eval_in_zonemap(ctx, slot, false, values, int_field(1), int_field(30)));
+              expr_zonemap::eval_in_zonemap(ctx, slot, false, values.min_max, *values.set));
     EXPECT_EQ(1, ctx.stats.in_zonemap_point_check_count);
 }
 
@@ -680,22 +749,205 @@ TEST(ExprZonemapFilterTest, InZonemapHandlesEmptyListAndNotInSingleValueRange) {
     auto slot = make_slot(0, type);
     auto ctx = make_context(make_int_zonemap(10, 20), type);
 
-    std::vector<Field> empty_values;
+    auto empty_values = make_int_set_with_min_max({});
     EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
-              expr_zonemap::eval_in_zonemap(ctx, slot, false, empty_values, {}, {}));
+              expr_zonemap::eval_in_zonemap(ctx, slot, false, empty_values.min_max,
+                                            *empty_values.set));
     EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
-              expr_zonemap::eval_in_zonemap(ctx, slot, true, empty_values, {}, {}));
+              expr_zonemap::eval_in_zonemap(ctx, slot, true, empty_values.min_max,
+                                            *empty_values.set));
 
     auto single_value_ctx = make_context(make_int_zonemap(10, 10), type);
-    std::vector<Field> values {int_field(10)};
+    auto values = make_int_set_with_min_max({10});
+    EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
+              expr_zonemap::eval_in_zonemap(single_value_ctx, slot, true, values.min_max,
+                                            *values.set));
+
+    auto other_values = make_int_set_with_min_max({11});
+    EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
+              expr_zonemap::eval_in_zonemap(single_value_ctx, slot, true, other_values.min_max,
+                                            *other_values.set));
+}
+
+// GTest assertion macros dominate the reported cognitive complexity of this linear scenario.
+// NOLINTNEXTLINE(readability-function-cognitive-complexity)
+TEST(ExprZonemapFilterTest, GetMinMaxHandlesEmptyNullOnlyAndSignedBitSetBounds) {
+    auto empty = make_int_set_with_min_max({});
+    EXPECT_EQ(0, empty.set->size());
+    EXPECT_FALSE(empty.set->contain_null());
+    EXPECT_TRUE(empty.min_max.min_value.is_null());
+    EXPECT_TRUE(empty.min_max.max_value.is_null());
+
+    auto null_only = make_int_set_with_min_max({}, true, true);
+    EXPECT_EQ(0, null_only.set->size());
+    EXPECT_TRUE(null_only.set->contain_null());
+    EXPECT_TRUE(null_only.min_max.min_value.is_null());
+    EXPECT_TRUE(null_only.min_max.max_value.is_null());
+
+    const std::vector<int8_t> values {127, -128, 0, -1};
+    auto tinyint =
+            make_typed_set_with_min_max<TYPE_TINYINT>(values, std::make_shared<DataTypeInt8>());
+    EXPECT_EQ(values.size(), tinyint.set->size());
+    EXPECT_EQ(Field::create_field<TYPE_TINYINT>(-128), tinyint.min_max.min_value);
+    EXPECT_EQ(Field::create_field<TYPE_TINYINT>(127), tinyint.min_max.max_value);
+
+    std::shared_ptr<HybridSetBase> smallint_set(create_set(TYPE_SMALLINT, false));
+    const int16_t smallint_min = std::numeric_limits<int16_t>::min();
+    const int16_t smallint_max = std::numeric_limits<int16_t>::max();
+    smallint_set->insert(&smallint_min);
+    smallint_set->insert(&smallint_max);
+    for (int16_t value = -31; value <= 31; ++value) {
+        smallint_set->insert(&value);
+    }
+    HybridSetMinMax smallint_min_max;
+    expr_zonemap::get_hybrid_set_min_max_for_zonemap_filter(
+            smallint_set, std::make_shared<DataTypeInt16>(), smallint_min_max);
+    EXPECT_EQ(65, smallint_set->size());
+    EXPECT_EQ(Field::create_field<TYPE_SMALLINT>(smallint_min), smallint_min_max.min_value);
+    EXPECT_EQ(Field::create_field<TYPE_SMALLINT>(smallint_max), smallint_min_max.max_value);
+
+    smallint_set->clear();
+    for (int16_t value = 100; value <= 164; ++value) {
+        smallint_set->insert(&value);
+    }
+    expr_zonemap::get_hybrid_set_min_max_for_zonemap_filter(
+            smallint_set, std::make_shared<DataTypeInt16>(), smallint_min_max);
+    EXPECT_EQ(65, smallint_set->size());
+    EXPECT_EQ(Field::create_field<TYPE_SMALLINT>(100), smallint_min_max.min_value);
+    EXPECT_EQ(Field::create_field<TYPE_SMALLINT>(164), smallint_min_max.max_value);
+}
+
+TEST(ExprZonemapFilterTest, StringSetHeterogeneousLookupPreservesBinaryValues) {
+    StringSet<> set(false);
+    const std::string empty;
+    const std::string binary("a\0b", 3);
+    StringRef empty_ref(empty);
+    StringRef binary_ref(binary);
+    set.insert(&empty_ref);
+    set.insert(&binary_ref);
+
+    EXPECT_TRUE(set.find(&empty_ref));
+    EXPECT_TRUE(set.find(empty.data(), empty.size()));
+    EXPECT_TRUE(set.find(Field::create_field<TYPE_STRING>(empty)));
+    EXPECT_TRUE(set.find(&binary_ref));
+    EXPECT_TRUE(set.find(binary.data(), binary.size()));
+    EXPECT_TRUE(set.find(Field::create_field<TYPE_STRING>(binary)));
+
+    const std::string prefix("a\0", 2);
+    EXPECT_FALSE(set.find(prefix.data(), prefix.size()));
+}
+
+TEST(ExprZonemapFilterTest, GetMinMaxPreservesDecimalAndDatetimeV2Values) {
+    const Decimal64 decimal_low(-1234);
+    const Decimal64 decimal_high(5678);
+    auto decimals = make_typed_set_with_min_max<TYPE_DECIMAL64>(
+            {decimal_high, decimal_low}, std::make_shared<DataTypeDecimal64>(18, 2));
+    EXPECT_EQ(Field::create_field<TYPE_DECIMAL64>(decimal_low), decimals.min_max.min_value);
+    EXPECT_EQ(Field::create_field<TYPE_DECIMAL64>(decimal_high), decimals.min_max.max_value);
+    EXPECT_TRUE(
+            decimals.set->contains_any_in_range(Field::create_field<TYPE_DECIMAL64>(decimal_low),
+                                                Field::create_field<TYPE_DECIMAL64>(decimal_low)));
+    EXPECT_FALSE(decimals.set->contains_any_in_range(
+            Field::create_field<TYPE_DECIMAL64>(Decimal64(-1000)),
+            Field::create_field<TYPE_DECIMAL64>(Decimal64(5000))));
+
+    DateV2Value<DateTimeV2ValueType> datetime_low;
+    datetime_low.unchecked_set_time(2024, 1, 2, 3, 4, 5, 123456);
+    DateV2Value<DateTimeV2ValueType> datetime_high;
+    datetime_high.unchecked_set_time(2025, 6, 7, 8, 9, 10, 654321);
+    auto datetimes = make_typed_set_with_min_max<TYPE_DATETIMEV2>(
+            {datetime_high, datetime_low}, std::make_shared<DataTypeDateTimeV2>(6));
+    EXPECT_EQ(Field::create_field<TYPE_DATETIMEV2>(datetime_low), datetimes.min_max.min_value);
+    EXPECT_EQ(Field::create_field<TYPE_DATETIMEV2>(datetime_high), datetimes.min_max.max_value);
+    EXPECT_TRUE(datetimes.set->contains_any_in_range(
+            Field::create_field<TYPE_DATETIMEV2>(datetime_high),
+            Field::create_field<TYPE_DATETIMEV2>(datetime_high)));
+    DateV2Value<DateTimeV2ValueType> datetime_hole;
+    datetime_hole.unchecked_set_time(2024, 6, 7, 8, 9, 10, 654321);
+    EXPECT_FALSE(datetimes.set->contains_any_in_range(
+            Field::create_field<TYPE_DATETIMEV2>(datetime_hole),
+            Field::create_field<TYPE_DATETIMEV2>(datetime_hole)));
+}
+
+TEST(ExprZonemapFilterTest, InBloomFilterHandlesEmptyAndNullOnlySets) {
+    auto type = int_type();
+    auto bloom_filter = make_int_bloom_filter({7});
+    auto bloom_ctx = make_bloom_filter_context(bloom_filter.get(), type);
+
+    std::shared_ptr<HybridSetBase> empty_values(create_set(TYPE_INT, false));
+    ASSERT_EQ(0, empty_values->size());
+    EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
+              expr_zonemap::eval_in_bloom_filter(bloom_ctx, make_slot(0, type), false,
+                                                 *empty_values));
+
+    std::shared_ptr<HybridSetBase> null_only_values(create_set(TYPE_INT, true));
+    null_only_values->insert(static_cast<const void*>(nullptr));
+    ASSERT_EQ(0, null_only_values->size());
+    ASSERT_TRUE(null_only_values->contain_null());
+    EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
+              expr_zonemap::eval_in_bloom_filter(bloom_ctx, make_slot(0, type), false,
+                                                 *null_only_values));
+}
+
+TEST(ExprZonemapFilterTest, InBloomFilterProbesInteriorNativeValues) {
+    auto type = int_type();
+    auto values = make_int_set_with_min_max({2, 4, 6});
+
+    auto missing_bloom_filter = make_int_bloom_filter({1, 3, 5});
+    auto missing_bloom_ctx = make_bloom_filter_context(missing_bloom_filter.get(), type);
     EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
-              expr_zonemap::eval_in_zonemap(single_value_ctx, slot, true, values, int_field(10),
-                                            int_field(10)));
+              expr_zonemap::eval_in_bloom_filter(missing_bloom_ctx, make_slot(0, type), false,
+                                                 *values.set));
 
-    std::vector<Field> other_values {int_field(11)};
+    // 4 is neither the IN-set minimum nor maximum, so this requires probing native set values.
+    auto matching_bloom_filter = make_int_bloom_filter({4});
+    auto matching_bloom_ctx = make_bloom_filter_context(matching_bloom_filter.get(), type);
     EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
-              expr_zonemap::eval_in_zonemap(single_value_ctx, slot, true, other_values,
-                                            int_field(11), int_field(11)));
+              expr_zonemap::eval_in_bloom_filter(matching_bloom_ctx, make_slot(0, type), false,
+                                                 *values.set));
+}
+
+TEST(ExprZonemapFilterTest, InBloomFilterPreservesEmptyAndEmbeddedNullStrings) {
+    auto type = std::make_shared<DataTypeString>();
+    const std::string empty;
+    const std::string binary("a\0b", 3);
+    for (const bool borrowed_values : {false, true}) {
+        SCOPED_TRACE(borrowed_values ? "StringValueSet" : "StringSet");
+        std::shared_ptr<HybridSetBase> values(borrowed_values ? create_string_value_set(false)
+                                                              : create_set(TYPE_STRING, false));
+        StringRef empty_ref(empty);
+        StringRef binary_ref(binary);
+        values->insert(&empty_ref);
+        values->insert(&binary_ref);
+
+        auto missing_bloom_filter = make_string_bloom_filter({});
+        auto missing_bloom_ctx = make_bloom_filter_context(missing_bloom_filter.get(), type);
+        EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
+                  expr_zonemap::eval_in_bloom_filter(missing_bloom_ctx, make_slot(0, type), false,
+                                                     *values));
+
+        auto empty_bloom_filter = make_string_bloom_filter({empty});
+        auto empty_bloom_ctx = make_bloom_filter_context(empty_bloom_filter.get(), type);
+        EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
+                  expr_zonemap::eval_in_bloom_filter(empty_bloom_ctx, make_slot(0, type), false,
+                                                     *values));
+
+        auto binary_bloom_filter = make_string_bloom_filter({binary});
+        auto binary_bloom_ctx = make_bloom_filter_context(binary_bloom_filter.get(), type);
+        EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
+                  expr_zonemap::eval_in_bloom_filter(binary_bloom_ctx, make_slot(0, type), false,
+                                                     *values));
+    }
+}
+
+TEST(ExprZonemapFilterTest, InBloomFilterKeepsUnsupportedTypeConservative) {
+    auto type = std::make_shared<DataTypeInt8>();
+    auto values = make_typed_set_with_min_max<TYPE_TINYINT>({int8_t {7}}, type);
+    auto bloom_filter = make_int_bloom_filter({});
+    auto bloom_ctx = make_bloom_filter_context(bloom_filter.get(), type);
+    EXPECT_EQ(
+            ZoneMapFilterResult::kMayMatch,
+            expr_zonemap::eval_in_bloom_filter(bloom_ctx, make_slot(0, type), false, *values.set));
 }
 
 TEST(ExprZonemapFilterTest, UnsupportedSingleSlotExprDoesNotAdvertiseZonemapCapability) {
@@ -713,7 +965,7 @@ TEST(ExprZonemapFilterTest, UnsupportedSingleSlotExprDoesNotAdvertiseZonemapCapa
     EXPECT_FALSE(equals.can_evaluate_zonemap_filter({unsupported_expr, make_int_literal(10)}));
 }
 
-TEST(ExprZonemapFilterTest, VInPredicateMaterializesZonemapValues) {
+TEST(ExprZonemapFilterTest, VInPredicatePreparesZonemapMinMax) {
     auto type = int_type();
     ObjectPool obj_pool;
     DescriptorTbl* desc_tbl = nullptr;
@@ -736,9 +988,9 @@ TEST(ExprZonemapFilterTest, VInPredicateMaterializesZonemapValues) {
 
     auto ctx = make_context(make_int_zonemap(10, 20), type);
     EXPECT_EQ(ZoneMapFilterResult::kNoMatch, in_predicate->evaluate_zonemap_filter(ctx));
-    EXPECT_TRUE(in_predicate->_zonemap_materialized);
-    EXPECT_EQ(int_field(1), in_predicate->_seg_filter_min);
-    EXPECT_EQ(int_field(30), in_predicate->_seg_filter_max);
+    ASSERT_NE(nullptr, in_predicate->_zonemap_min_max);
+    EXPECT_EQ(int_field(1), in_predicate->_zonemap_min_max->min_value);
+    EXPECT_EQ(int_field(30), in_predicate->_zonemap_min_max->max_value);
 
     auto not_in_with_null = std::make_shared<VInPredicate>(make_in_predicate_node(true, 3));
     auto not_in_slot = make_slot(0, type);
@@ -753,10 +1005,11 @@ TEST(ExprZonemapFilterTest, VInPredicateMaterializesZonemapValues) {
     auto may_match_ctx = make_context(make_int_zonemap(11, 11), type);
     EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
               not_in_with_null->evaluate_zonemap_filter(may_match_ctx));
-    EXPECT_TRUE(not_in_with_null->_seg_filter_contains_null);
+    ASSERT_NE(nullptr, not_in_with_null->_direct_filter_set);
+    EXPECT_TRUE(not_in_with_null->_direct_filter_set->contain_null());
 }
 
-TEST(ExprZonemapFilterTest, VInPredicateDictionaryAndBloomUseMaterializedValues) {
+TEST(ExprZonemapFilterTest, VInPredicateDictionaryAndBloomProbePreparedSet) {
     auto type = int_type();
     ObjectPool obj_pool;
     DescriptorTbl* desc_tbl = nullptr;
@@ -767,12 +1020,13 @@ TEST(ExprZonemapFilterTest, VInPredicateDictionaryAndBloomUseMaterializedValues)
     runtime_state.set_desc_tbl(desc_tbl);
     RowDescriptor row_desc(runtime_state.desc_tbl(), {0});
 
-    auto in_predicate = std::make_shared<VInPredicate>(make_in_predicate_node(false, 3));
+    auto in_predicate = std::make_shared<VInPredicate>(make_in_predicate_node(false, 4));
     auto in_slot = make_slot(0, type);
     std::static_pointer_cast<VSlotRef>(in_slot)->set_slot_id(0);
     in_predicate->add_child(in_slot);
     in_predicate->add_child(make_int_literal(2));
     in_predicate->add_child(make_int_literal(4));
+    in_predicate->add_child(make_int_literal(6));
     VExprContext in_context(in_predicate);
     ASSERT_TRUE(in_context.prepare(&runtime_state, row_desc).ok());
     ASSERT_TRUE(in_context.open(&runtime_state).ok());
@@ -790,13 +1044,77 @@ TEST(ExprZonemapFilterTest, VInPredicateDictionaryAndBloomUseMaterializedValues)
     auto missing_bloom_ctx = make_bloom_filter_context(missing_bloom_filter.get(), type);
     EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
               in_predicate->evaluate_bloom_filter(missing_bloom_ctx));
+    // 4 is neither the IN-set minimum nor maximum. Bloom pruning must probe the full native set.
     auto matching_bloom_filter = make_int_bloom_filter({4});
     auto matching_bloom_ctx = make_bloom_filter_context(matching_bloom_filter.get(), type);
     EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
               in_predicate->evaluate_bloom_filter(matching_bloom_ctx));
 }
 
-TEST(ExprZonemapFilterTest, DirectInPredicateMaterializesStringSetForZonemap) {
+// GoogleTest assertions inflate this linear ownership-lifetime test's complexity metric.
+TEST(ExprZonemapFilterTest, // NOLINT(readability-function-cognitive-complexity)
+     VInPredicatePreparesOwningStringZonemapMinMax) {
+    std::shared_ptr<const HybridSetMinMax> snapshot;
+    std::weak_ptr<HybridSetBase> borrowed_set;
+    {
+        auto type = std::make_shared<DataTypeString>();
+        ObjectPool obj_pool;
+        DescriptorTbl* desc_tbl = nullptr;
+        auto thrift_desc_tbl = make_k2_scan_desc_tbl(TYPE_STRING);
+        ASSERT_TRUE(DescriptorTbl::create(&obj_pool, thrift_desc_tbl, &desc_tbl).ok());
+
+        RuntimeState runtime_state;
+        runtime_state.set_desc_tbl(desc_tbl);
+        RowDescriptor row_desc(runtime_state.desc_tbl(), {0});
+
+        auto in_predicate = std::make_shared<VInPredicate>(make_in_predicate_node(false, 5));
+        auto in_slot = make_slot(0, type);
+        std::static_pointer_cast<VSlotRef>(in_slot)->set_slot_id(0);
+        in_predicate->add_child(in_slot);
+        in_predicate->add_child(make_string_literal("zzz"));
+        in_predicate->add_child(make_string_literal("aaa"));
+        in_predicate->add_child(make_string_literal("aaa"));
+        in_predicate->add_child(make_null_string_literal());
+        VExprContext in_context(in_predicate);
+        ASSERT_TRUE(in_context.prepare(&runtime_state, row_desc).ok());
+        ASSERT_TRUE(in_context.open(&runtime_state).ok());
+
+        ASSERT_NE(nullptr, in_predicate->_zonemap_min_max);
+        ASSERT_NE(nullptr, dynamic_cast<StringValueSet<>*>(in_predicate->_direct_filter_set.get()));
+        borrowed_set = in_predicate->_direct_filter_set;
+        EXPECT_EQ(2, in_predicate->_direct_filter_set->size());
+        EXPECT_TRUE(in_predicate->_direct_filter_set->contain_null());
+        EXPECT_EQ(Field::create_field<TYPE_STRING>("aaa"),
+                  in_predicate->_zonemap_min_max->min_value);
+        EXPECT_EQ(Field::create_field<TYPE_STRING>("zzz"),
+                  in_predicate->_zonemap_min_max->max_value);
+
+        EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
+                  in_predicate->evaluate_dictionary_filter(make_dictionary_context(
+                          {Field::create_field<TYPE_STRING>("aaa")}, type)));
+        EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
+                  in_predicate->evaluate_dictionary_filter(make_dictionary_context(
+                          {Field::create_field<TYPE_STRING>("mmm")}, type)));
+
+        auto missing_bloom_filter = make_string_bloom_filter({});
+        auto missing_bloom_ctx = make_bloom_filter_context(missing_bloom_filter.get(), type);
+        EXPECT_EQ(ZoneMapFilterResult::kNoMatch,
+                  in_predicate->evaluate_bloom_filter(missing_bloom_ctx));
+        auto matching_bloom_filter = make_string_bloom_filter({"aaa"});
+        auto matching_bloom_ctx = make_bloom_filter_context(matching_bloom_filter.get(), type);
+        EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
+                  in_predicate->evaluate_bloom_filter(matching_bloom_ctx));
+
+        snapshot = in_predicate->_zonemap_min_max;
+    }
+
+    ASSERT_NE(nullptr, snapshot);
+    EXPECT_TRUE(borrowed_set.expired());
+    EXPECT_EQ(Field::create_field<TYPE_STRING>("aaa"), snapshot->min_value);
+    EXPECT_EQ(Field::create_field<TYPE_STRING>("zzz"), snapshot->max_value);
+}
+
+TEST(ExprZonemapFilterTest, DirectInPredicatePreparesStringMinMaxForZonemap) {
     auto type = std::make_shared<DataTypeString>();
     std::shared_ptr<HybridSetBase> filter(create_set(PrimitiveType::TYPE_STRING, false));
     StringRef aaa("aaa");
@@ -807,17 +1125,14 @@ TEST(ExprZonemapFilterTest, DirectInPredicateMaterializesStringSetForZonemap) {
     auto slot = make_slot(0, type);
     VDirectInPredicate direct_in_expr(make_in_predicate_node(false, 2), filter, true);
     direct_in_expr.add_child(slot);
-    ASSERT_TRUE(direct_in_expr._materialize_for_zonemap_filter().ok());
-
-    EXPECT_TRUE(direct_in_expr._pruning_state->zonemap_materialized);
-    EXPECT_EQ(2, direct_in_expr._pruning_state->seg_filter_values.size());
-    EXPECT_EQ(Field::create_field<TYPE_STRING>("aaa"),
-              direct_in_expr._pruning_state->seg_filter_min);
-    EXPECT_EQ(Field::create_field<TYPE_STRING>("zzz"),
-              direct_in_expr._pruning_state->seg_filter_max);
+    direct_in_expr._prepare_zonemap_min_max();
+
+    ASSERT_NE(nullptr, direct_in_expr._zonemap_min_max);
+    EXPECT_EQ(Field::create_field<TYPE_STRING>("aaa"), direct_in_expr._zonemap_min_max->min_value);
+    EXPECT_EQ(Field::create_field<TYPE_STRING>("zzz"), direct_in_expr._zonemap_min_max->max_value);
 }
 
-TEST(ExprZonemapFilterTest, DirectInPredicateMaterializesZonemapValuesDuringPrepare) {
+TEST(ExprZonemapFilterTest, DirectInPredicatePreparesZonemapMinMax) {
     auto type = int_type();
     ObjectPool obj_pool;
     DescriptorTbl* desc_tbl = nullptr;
@@ -843,14 +1158,13 @@ TEST(ExprZonemapFilterTest, DirectInPredicateMaterializesZonemapValuesDuringPrep
     VExprContext context(direct_in_expr);
     ASSERT_TRUE(context.prepare(&runtime_state, row_desc).ok());
 
-    EXPECT_TRUE(direct_in_expr->_pruning_state->zonemap_materialized);
     EXPECT_TRUE(direct_in_expr->can_evaluate_zonemap_filter());
-    EXPECT_EQ(2, direct_in_expr->_pruning_state->seg_filter_values.size());
-    EXPECT_EQ(int_field(1), direct_in_expr->_pruning_state->seg_filter_min);
-    EXPECT_EQ(int_field(30), direct_in_expr->_pruning_state->seg_filter_max);
+    ASSERT_NE(nullptr, direct_in_expr->_zonemap_min_max);
+    EXPECT_EQ(int_field(1), direct_in_expr->_zonemap_min_max->min_value);
+    EXPECT_EQ(int_field(30), direct_in_expr->_zonemap_min_max->max_value);
 }
 
-TEST(ExprZonemapFilterTest, DirectInPredicateDeepCloneReusesMaterializedPruningState) {
+TEST(ExprZonemapFilterTest, DirectInDeepCloneAfterMinMaxPreparationReusesSnapshot) {
     auto type = int_type();
     std::shared_ptr<HybridSetBase> filter(create_set(PrimitiveType::TYPE_INT, false));
     int32_t low_value = 1;
@@ -861,18 +1175,54 @@ TEST(ExprZonemapFilterTest, DirectInPredicateDeepCloneReusesMaterializedPruningS
     auto direct_in_expr =
             std::make_shared<VDirectInPredicate>(make_in_predicate_node(false, 1), filter, true);
     direct_in_expr->add_child(make_slot(0, type));
-    ASSERT_TRUE(direct_in_expr->_materialize_for_zonemap_filter().ok());
+    direct_in_expr->_prepare_zonemap_min_max();
 
     VExprSPtr cloned_expr;
     ASSERT_TRUE(direct_in_expr->deep_clone(&cloned_expr).ok());
     auto cloned_direct_in = std::dynamic_pointer_cast<VDirectInPredicate>(cloned_expr);
     ASSERT_NE(cloned_direct_in, nullptr);
-    EXPECT_EQ(direct_in_expr->_pruning_state, cloned_direct_in->_pruning_state);
+    EXPECT_EQ(direct_in_expr->_zonemap_min_max.get(), cloned_direct_in->_zonemap_min_max.get());
     EXPECT_TRUE(cloned_direct_in->can_evaluate_zonemap_filter());
     EXPECT_EQ(ZoneMapFilterResult::kNoMatch, cloned_direct_in->evaluate_zonemap_filter(
                                                      make_context(make_int_zonemap(10, 20), type)));
 }
 
+TEST(ExprZonemapFilterTest, DirectInDeepCloneBeforeMinMaxPreparationBuildsIndependentSnapshots) {
+    auto type = int_type();
+    std::shared_ptr<HybridSetBase> filter(create_set(TYPE_INT, false));
+    int32_t low_value = 1;
+    int32_t high_value = 30;
+    filter->insert(&low_value);
+    filter->insert(&high_value);
+
+    auto direct_in_expr =
+            std::make_shared<VDirectInPredicate>(make_in_predicate_node(false, 1), filter, true);
+    direct_in_expr->add_child(make_slot(0, type));
+
+    VExprSPtr cloned_expr;
+    ASSERT_TRUE(direct_in_expr->deep_clone(&cloned_expr).ok());
+    auto cloned_direct_in = std::dynamic_pointer_cast<VDirectInPredicate>(cloned_expr);
+    ASSERT_NE(nullptr, cloned_direct_in);
+    EXPECT_EQ(nullptr, direct_in_expr->_zonemap_min_max);
+    EXPECT_EQ(nullptr, cloned_direct_in->_zonemap_min_max);
+
+    direct_in_expr->_prepare_zonemap_min_max();
+    cloned_direct_in->_prepare_zonemap_min_max();
+    ASSERT_NE(nullptr, direct_in_expr->_zonemap_min_max);
+    ASSERT_NE(nullptr, cloned_direct_in->_zonemap_min_max);
+    EXPECT_NE(direct_in_expr->_zonemap_min_max.get(), cloned_direct_in->_zonemap_min_max.get());
+    EXPECT_EQ(direct_in_expr->_zonemap_min_max->min_value,
+              cloned_direct_in->_zonemap_min_max->min_value);
+    EXPECT_EQ(direct_in_expr->_zonemap_min_max->max_value,
+              cloned_direct_in->_zonemap_min_max->max_value);
+
+    auto ctx = make_context(make_int_zonemap(10, 20), type);
+    EXPECT_EQ(ZoneMapFilterResult::kNoMatch, cloned_direct_in->evaluate_zonemap_filter(ctx));
+    EXPECT_EQ(ZoneMapFilterResult::kMayMatch,
+              cloned_direct_in->evaluate_dictionary_filter(
+                      make_dictionary_context({int_field(30)}, type)));
+}
+
 TEST(ExprZonemapFilterTest, DirectInPredicateRewritesStringSetToInPredicate) {
     auto type = std::make_shared<DataTypeString>();
     auto slot = make_slot(0, type);
@@ -888,7 +1238,28 @@ TEST(ExprZonemapFilterTest, DirectInPredicateRewritesStringSetToInPredicate) {
     EXPECT_NE(std::string::npos, in_expr->debug_string().find("iceberg"));
 }
 
-TEST(ExprZonemapFilterTest, DirectInPredicateSkipsMaterializationWhenSetTypeDiffersFromChild) {
+TEST(ExprZonemapFilterTest, DirectInPredicateRewritePreservesEmbeddedNullString) {
+    auto type = std::make_shared<DataTypeString>();
+    auto slot = make_slot(0, type);
+    std::shared_ptr<HybridSetBase> filter(create_set(PrimitiveType::TYPE_STRING, false));
+    const std::string binary_value("a\0b", 3);
+    StringRef value(binary_value);
+    filter->insert(&value);
+
+    VDirectInPredicate direct_in_expr(make_in_predicate_node(false, 1), filter, true);
+    direct_in_expr.add_child(slot);
+
+    VExprSPtr in_expr;
+    ASSERT_TRUE(direct_in_expr.get_slot_in_expr(in_expr));
+    ASSERT_EQ(2, in_expr->get_num_children());
+    auto literal = std::dynamic_pointer_cast<VLiteral>(in_expr->get_child(1));
+    ASSERT_NE(nullptr, literal);
+    Field materialized_value;
+    literal->get_column_ptr()->get(0, materialized_value);
+    EXPECT_EQ(binary_value, std::string(materialized_value.as_string_view()));
+}
+
+TEST(ExprZonemapFilterTest, DirectInPredicateSkipsMinMaxWhenSetTypeDiffersFromChild) {
     auto string_type = std::make_shared<DataTypeString>();
     auto slot = make_slot(0, string_type);
     std::shared_ptr<HybridSetBase> filter(create_set(PrimitiveType::TYPE_INT, false));
@@ -898,8 +1269,8 @@ TEST(ExprZonemapFilterTest, DirectInPredicateSkipsMaterializationWhenSetTypeDiff
     VDirectInPredicate direct_in_expr(make_in_predicate_node(false, 1), filter, false);
     direct_in_expr.add_child(slot);
 
-    ASSERT_TRUE(direct_in_expr._materialize_for_zonemap_filter().ok());
-    EXPECT_FALSE(direct_in_expr._pruning_state->zonemap_materialized);
+    direct_in_expr._prepare_zonemap_min_max();
+    EXPECT_EQ(nullptr, direct_in_expr._zonemap_min_max);
     VExprSPtr in_expr;
     EXPECT_FALSE(direct_in_expr.get_slot_in_expr(in_expr));
 }
@@ -916,7 +1287,7 @@ TEST(ExprZonemapFilterTest, RuntimeFilterExprNullAwareZonemapKeepsZonesWithNull)
     auto direct_in_expr =
             std::make_shared<VDirectInPredicate>(make_in_predicate_node(false, 1), filter, true);
     direct_in_expr->add_child(slot);
-    ASSERT_TRUE(direct_in_expr->_materialize_for_zonemap_filter().ok());
+    direct_in_expr->_prepare_zonemap_min_max();
 
     auto runtime_filter = RuntimeFilterExpr::create_shared(make_in_predicate_node(false, 1),
                                                            direct_in_expr, 0.0, true, 7);
@@ -950,7 +1321,7 @@ TEST(ExprZonemapFilterTest, RuntimeFilterExprDelegatesDirectInDictionaryAndRawEv
     auto direct_in_expr =
             std::make_shared<VDirectInPredicate>(make_in_predicate_node(false, 1), filter, true);
     direct_in_expr->add_child(slot);
-    ASSERT_TRUE(direct_in_expr->_materialize_for_zonemap_filter().ok());
+    direct_in_expr->_prepare_zonemap_min_max();
 
     auto runtime_filter = RuntimeFilterExpr::create_shared(make_in_predicate_node(false, 1),
                                                            direct_in_expr, 0.0, false, 7);
diff --git a/be/test/exprs/hybrid_set_test.cpp b/be/test/exprs/hybrid_set_test.cpp
index aed2103d66f34b..bbf044bbe57687 100644
--- a/be/test/exprs/hybrid_set_test.cpp
+++ b/be/test/exprs/hybrid_set_test.cpp
@@ -19,9 +19,13 @@
 
 #include <gtest/gtest.h>
 
+#include <array>
+#include <cstdint>
+#include <limits>
 #include <memory>
 #include <string>
 
+#include "core/field.h"
 #include "exprs/create_predicate_function.h"
 #include "gtest/internal/gtest-internal.h"
 #include "testutil/column_helper.h"
@@ -127,6 +131,148 @@ TEST_F(HybridSetTest, Numeric) {
     TEST_NUMERIC(PrimitiveType::TYPE_DECIMAL128I);
 }
 
+TEST_F(HybridSetTest, IntegerMinMaxAndRangeLookup) {
+    const auto field = [](int32_t value) { return Field::create_field<TYPE_INT>(value); };
+    const auto verify = [&](HybridSetBase& set) {
+        for (int32_t value : {1, 5, 9}) {
+            set.insert(&value);
+        }
+
+        Field min_value;
+        Field max_value;
+        set.get_min_max(min_value, max_value);
+        EXPECT_EQ(min_value.get<TYPE_INT>(), 1);
+        EXPECT_EQ(max_value.get<TYPE_INT>(), 9);
+
+        EXPECT_TRUE(set.contains_any_in_range(field(1), field(1)));
+        EXPECT_TRUE(set.contains_any_in_range(field(4), field(5)));
+        EXPECT_TRUE(set.contains_any_in_range(field(9), field(10)));
+        EXPECT_FALSE(set.contains_any_in_range(field(2), field(4)));
+        EXPECT_FALSE(set.contains_any_in_range(field(6), field(8)));
+
+        set.clear();
+        set.get_min_max(min_value, max_value);
+        EXPECT_TRUE(min_value.is_null());
+        EXPECT_TRUE(max_value.is_null());
+    };
+
+    HybridSet<TYPE_INT> dynamic_set(false);
+    verify(dynamic_set);
+
+    HybridSet<TYPE_INT, FixedContainer<int32_t, 3>> fixed_set(false);
+    verify(fixed_set);
+}
+
+TEST_F(HybridSetTest, SignedBitSetRangeLookup) {
+    const auto tinyint_field = [](int8_t value) {
+        return Field::create_field<TYPE_TINYINT>(value);
+    };
+    HybridSet<TYPE_TINYINT, BitSetContainer<int8_t>> tinyint_set(false);
+    for (int8_t value : {int8_t {-100}, int8_t {-1}, int8_t {0}, int8_t {100}}) {
+        tinyint_set.insert(&value);
+    }
+    EXPECT_TRUE(tinyint_set.contains_any_in_range(tinyint_field(-2), tinyint_field(1)));
+    EXPECT_TRUE(tinyint_set.contains_any_in_range(tinyint_field(-100), tinyint_field(-100)));
+    EXPECT_FALSE(tinyint_set.contains_any_in_range(tinyint_field(-99), tinyint_field(-2)));
+    EXPECT_FALSE(tinyint_set.contains_any_in_range(tinyint_field(1), tinyint_field(99)));
+
+    const auto smallint_field = [](int16_t value) {
+        return Field::create_field<TYPE_SMALLINT>(value);
+    };
+    HybridSet<TYPE_SMALLINT, BitSetContainer<int16_t>> edge_set(false);
+    int16_t min_value = std::numeric_limits<int16_t>::min();
+    int16_t max_value = std::numeric_limits<int16_t>::max();
+    edge_set.insert(&min_value);
+    edge_set.insert(&max_value);
+
+    Field min_field;
+    Field max_field;
+    edge_set.get_min_max(min_field, max_field);
+    EXPECT_EQ(min_field.get<TYPE_SMALLINT>(), min_value);
+    EXPECT_EQ(max_field.get<TYPE_SMALLINT>(), max_value);
+    EXPECT_TRUE(
+            edge_set.contains_any_in_range(smallint_field(min_value), smallint_field(min_value)));
+    EXPECT_TRUE(
+            edge_set.contains_any_in_range(smallint_field(max_value), smallint_field(max_value)));
+    EXPECT_FALSE(
+            edge_set.contains_any_in_range(smallint_field(static_cast<int16_t>(min_value + 1)),
+                                           smallint_field(static_cast<int16_t>(max_value - 1))));
+
+    HybridSet<TYPE_SMALLINT, BitSetContainer<int16_t>> crossing_set(false);
+    for (int16_t value : {int16_t {-30000}, int16_t {-1}, int16_t {0}, int16_t {30000}}) {
+        crossing_set.insert(&value);
+    }
+    EXPECT_TRUE(crossing_set.contains_any_in_range(smallint_field(-2), smallint_field(1)));
+    EXPECT_TRUE(crossing_set.contains_any_in_range(smallint_field(-1), smallint_field(-1)));
+    EXPECT_TRUE(crossing_set.contains_any_in_range(smallint_field(0), smallint_field(0)));
+    EXPECT_FALSE(crossing_set.contains_any_in_range(smallint_field(-29999), smallint_field(-2)));
+    EXPECT_FALSE(crossing_set.contains_any_in_range(smallint_field(1), smallint_field(29999)));
+
+    // 63/64 straddles a 64-value block. 4095/4096 straddles a summary-word boundary;
+    // -28673/-28672 maps to the equivalent boundary in the negative raw-index half.
+    HybridSet<TYPE_SMALLINT, BitSetContainer<int16_t>> boundary_set(false);
+    for (int16_t value : {int16_t {63}, int16_t {64}, int16_t {4095}, int16_t {4096},
+                          int16_t {-28673}, int16_t {-28672}}) {
+        boundary_set.insert(&value);
+    }
+    EXPECT_TRUE(boundary_set.contains_any_in_range(smallint_field(63), smallint_field(63)));
+    EXPECT_TRUE(boundary_set.contains_any_in_range(smallint_field(64), smallint_field(64)));
+    EXPECT_TRUE(boundary_set.contains_any_in_range(smallint_field(4095), smallint_field(4095)));
+    EXPECT_TRUE(boundary_set.contains_any_in_range(smallint_field(4096), smallint_field(4096)));
+    EXPECT_TRUE(boundary_set.contains_any_in_range(smallint_field(-28673), smallint_field(-28673)));
+    EXPECT_TRUE(boundary_set.contains_any_in_range(smallint_field(-28672), smallint_field(-28672)));
+    EXPECT_FALSE(boundary_set.contains_any_in_range(smallint_field(65), smallint_field(4094)));
+
+    boundary_set.clear();
+    int16_t value_after_clear = 4096;
+    boundary_set.insert(&value_after_clear);
+    EXPECT_FALSE(boundary_set.contains_any_in_range(smallint_field(4095), smallint_field(4095)));
+    EXPECT_TRUE(boundary_set.contains_any_in_range(smallint_field(4096), smallint_field(4096)));
+}
+
+TEST_F(HybridSetTest, StringRangeLookupPreservesEmbeddedNull) {
+    const std::array<std::string, 3> values = {std::string("a\0a", 3), std::string("a\0c", 3),
+                                               std::string("b\0b", 3)};
+    const std::string missing("a\0b", 3);
+    const std::string upper_hole("b\0a", 3);
+    const auto field = [](const std::string& value) {
+        return Field::create_field<TYPE_STRING>(String(value.data(), value.size()));
+    };
+    const auto verify = [&](HybridSetBase& set) {
+        Field min_value;
+        Field max_value;
+        set.get_min_max(min_value, max_value);
+        EXPECT_EQ(min_value.get<TYPE_STRING>(), values.front());
+        EXPECT_EQ(max_value.get<TYPE_STRING>(), values.back());
+
+        EXPECT_TRUE(set.contains_any_in_range(field(values.front()), field(values.front())));
+        EXPECT_TRUE(set.contains_any_in_range(field(missing), field(values[1])));
+        EXPECT_FALSE(set.contains_any_in_range(field(missing), field(missing)));
+        EXPECT_FALSE(set.contains_any_in_range(field(std::string("a\0d", 3)), field(upper_hole)));
+    };
+
+    StringSet<> owning_set(false);
+    for (const auto& value : values) {
+        StringRef ref(value);
+        owning_set.insert(&ref);
+    }
+    verify(owning_set);
+
+    StringSet<FixedContainer<std::string, 3>> fixed_owning_set(false);
+    for (const auto& value : values) {
+        StringRef ref(value);
+        fixed_owning_set.insert(&ref);
+    }
+    verify(fixed_owning_set);
+
+    StringValueSet<> borrowed_set(false);
+    for (const auto& value : values) {
+        StringRef ref(value);
+        borrowed_set.insert(&ref);
+    }
+    verify(borrowed_set);
+}
+
 #define TEST_DATE(primitive_type)                                                  \
     do {                                                                           \
         using NumericType = PrimitiveTypeTraits<primitive_type>::CppType;          \
diff --git a/be/test/format_v2/orc/orc_reader_test.cpp b/be/test/format_v2/orc/orc_reader_test.cpp
index 63a09b04b1cced..90a3b296d38f4e 100644
--- a/be/test/format_v2/orc/orc_reader_test.cpp
+++ b/be/test/format_v2/orc/orc_reader_test.cpp
@@ -4495,7 +4495,7 @@ class NewOrcReaderTest : public testing::Test {
 
     struct DirectInScanResult {
         std::vector<int32_t> ids;
-        size_t materialize_calls = 0;
+        size_t begin_calls = 0;
         size_t prepare_literals_calls = 0;
         int64_t filtered_row_groups = 0;
         int64_t filtered_row_groups_by_min_max = 0;
@@ -4630,7 +4630,7 @@ class NewOrcReaderTest : public testing::Test {
             }
         }
 
-        result->materialize_calls = filter->begin_calls();
+        result->begin_calls = filter->begin_calls();
         result->filtered_row_groups = reader->reader_statistics().filtered_row_groups;
         result->filtered_row_groups_by_min_max =
                 reader->reader_statistics().filtered_row_groups_by_min_max;
@@ -7019,11 +7019,11 @@ TEST_F(NewOrcReaderTest, SargDirectInCompoundMaterializesLiteralsOnce) {
                                           shape, &disabled)
                             .ok());
 
-        EXPECT_EQ(enabled.materialize_calls, 1);
+        EXPECT_EQ(enabled.begin_calls, 1);
         EXPECT_EQ(enabled.prepare_literals_calls, 1);
         EXPECT_EQ(enabled.filtered_row_groups, 1);
         EXPECT_EQ(enabled.filtered_row_groups_by_min_max, 1);
-        EXPECT_EQ(disabled.materialize_calls, 0);
+        EXPECT_EQ(disabled.begin_calls, 0);
         EXPECT_EQ(disabled.prepare_literals_calls, 0);
         EXPECT_EQ(disabled.filtered_row_groups, 0);
         EXPECT_EQ(disabled.filtered_row_groups_by_min_max, 0);
@@ -7049,11 +7049,11 @@ TEST_F(NewOrcReaderTest, SargDirectInNullSafeOrFallsBackBeforeLiteralConversion)
                                       DirectInPredicateShape::OR_WITH_NULL_SAFE_EQUAL, &disabled)
                         .ok());
 
-    EXPECT_EQ(enabled.materialize_calls, 1);
+    EXPECT_EQ(enabled.begin_calls, 1);
     EXPECT_EQ(enabled.prepare_literals_calls, 0);
     EXPECT_EQ(enabled.filtered_row_groups, 0);
     EXPECT_EQ(enabled.filtered_row_groups_by_min_max, 0);
-    EXPECT_EQ(disabled.materialize_calls, 0);
+    EXPECT_EQ(disabled.begin_calls, 0);
     EXPECT_EQ(disabled.prepare_literals_calls, 0);
     EXPECT_EQ(disabled.filtered_row_groups, 0);
     EXPECT_EQ(disabled.filtered_row_groups_by_min_max, 0);
@@ -7081,11 +7081,11 @@ TEST_F(NewOrcReaderTest, SargDirectInOverLimitFallsBackBeforeMaterialization) {
                                       DirectInPredicateShape::ROOT, &disabled)
                         .ok());
 
-    EXPECT_EQ(enabled.materialize_calls, 0);
+    EXPECT_EQ(enabled.begin_calls, 0);
     EXPECT_EQ(enabled.prepare_literals_calls, 0);
     EXPECT_EQ(enabled.filtered_row_groups, 0);
     EXPECT_EQ(enabled.filtered_row_groups_by_min_max, 0);
-    EXPECT_EQ(disabled.materialize_calls, 0);
+    EXPECT_EQ(disabled.begin_calls, 0);
     EXPECT_EQ(disabled.prepare_literals_calls, 0);
     EXPECT_EQ(disabled.filtered_row_groups, 0);
     EXPECT_EQ(disabled.filtered_row_groups_by_min_max, 0);
diff --git a/be/test/format_v2/parquet/parquet_statistics_test.cpp b/be/test/format_v2/parquet/parquet_statistics_test.cpp
index 740892367bb04e..baace3296be2a1 100644
--- a/be/test/format_v2/parquet/parquet_statistics_test.cpp
+++ b/be/test/format_v2/parquet/parquet_statistics_test.cpp
@@ -38,7 +38,9 @@
 #include "core/data_type/data_type_string.h"
 #include "core/data_type/data_type_time.h"
 #include "core/field.h"
+#include "exprs/create_predicate_function.h"
 #include "exprs/expr_zonemap_filter.h"
+#include "exprs/hybrid_set.h"
 #include "exprs/vexpr.h"
 #include "exprs/vexpr_context.h"
 #include "exprs/vslot_ref.h"
@@ -84,7 +86,7 @@ class StatisticsMemoryFileReader final : public io::FileReader {
 };
 class BloomInExpr final : public VExpr {
 public:
-    BloomInExpr(int column_id, DataTypePtr data_type, std::vector<Field> values)
+    BloomInExpr(int column_id, DataTypePtr data_type, std::shared_ptr<HybridSetBase> values)
             : VExpr(std::make_shared<DataTypeUInt8>(), false),
               _slot(VSlotRef::create_shared(0, column_id, -1, std::move(data_type), "c0")),
               _values(std::move(values)) {}
@@ -99,7 +101,7 @@ class BloomInExpr final : public VExpr {
     bool can_evaluate_bloom_filter() const override { return true; }
 
     ZoneMapFilterResult evaluate_bloom_filter(const BloomFilterEvalContext& ctx) const override {
-        return expr_zonemap::eval_in_bloom_filter(ctx, _slot, false, _values);
+        return expr_zonemap::eval_in_bloom_filter(ctx, _slot, false, *_values);
     }
 
     void collect_slot_column_ids(std::set<int>& column_ids) const override {
@@ -108,7 +110,7 @@ class BloomInExpr final : public VExpr {
 
 private:
     VExprSPtr _slot;
-    std::vector<Field> _values;
+    std::shared_ptr<HybridSetBase> _values;
     const std::string _expr_name = "BloomInExpr";
 };
 
@@ -196,16 +198,20 @@ class MetadataBoundsProbeExpr final : public VExpr {
     bool _require_false_boolean;
     const std::string _expr_name = "MetadataBoundsProbeExpr";
 };
-VExprContextSPtrs bloom_conjuncts(DataTypePtr data_type, std::vector<Field> values) {
+VExprContextSPtrs bloom_conjuncts(DataTypePtr data_type, const std::vector<int64_t>& values) {
+    std::shared_ptr<HybridSetBase> set(create_set(PrimitiveType::TYPE_BIGINT, false));
+    for (const auto value : values) {
+        set->insert(&value);
+    }
     return {VExprContext::create_shared(
-            std::make_shared<BloomInExpr>(0, std::move(data_type), std::move(values)))};
+            std::make_shared<BloomInExpr>(0, std::move(data_type), std::move(set)))};
 }
 
 format::FileScanRequest request_with_bloom_conjunct(DataTypePtr data_type,
-                                                    std::vector<Field> values) {
+                                                    const std::vector<int64_t>& values) {
     format::FileScanRequest request;
     request.local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0));
-    request.conjuncts = bloom_conjuncts(std::move(data_type), std::move(values));
+    request.conjuncts = bloom_conjuncts(std::move(data_type), values);
     return request;
 }
 format::parquet::ParquetColumnSchema uint32_parquet_bloom_schema() {
@@ -376,14 +382,9 @@ TEST(ParquetBloomFilterPruningTest, NativeUint32BloomUsesPhysicalInt32Hash) {
     bloom_filter.add_bytes(reinterpret_cast<const char*>(&physical_value), sizeof(physical_value));
 
     EXPECT_FALSE(format::parquet::ParquetStatisticsUtils::NativeBloomFilterExcludes(
-            column_schema, 0,
-            bloom_conjuncts(column_schema.type, {Field::create_field<TYPE_BIGINT>(
-                                                        static_cast<int64_t>(present_value))}),
-            bloom_filter));
+            column_schema, 0, bloom_conjuncts(column_schema.type, {present_value}), bloom_filter));
     EXPECT_TRUE(format::parquet::ParquetStatisticsUtils::NativeBloomFilterExcludes(
-            column_schema, 0,
-            bloom_conjuncts(column_schema.type, {Field::create_field<TYPE_BIGINT>(-1)}),
-            bloom_filter));
+            column_schema, 0, bloom_conjuncts(column_schema.type, {-1}), bloom_filter));
 }
 
 TEST(ParquetBloomFilterPruningTest, NativeRowGroupKeepsPresentUint32AboveInt32Max) {
@@ -440,9 +441,7 @@ TEST(ParquetBloomFilterPruningTest, NativeRowGroupKeepsPresentUint32AboveInt32Ma
 
     format::parquet::ParquetFileContext file_context;
     file_context.native_file = std::make_shared<StatisticsMemoryFileReader>(std::move(bloom_bytes));
-    auto request = request_with_bloom_conjunct(
-            column_schema->type,
-            {Field::create_field<TYPE_BIGINT>(static_cast<int64_t>(present_value))});
+    auto request = request_with_bloom_conjunct(column_schema->type, {present_value});
     std::vector<std::unique_ptr<format::parquet::ParquetColumnSchema>> schema;
     schema.push_back(std::move(column_schema));
     std::vector<int> selected_row_groups;