diff --git a/be/src/exprs/hybrid_set.h b/be/src/exprs/hybrid_set.h
index 6536ec2430fe08..dbf31a54772ea3 100644
--- a/be/src/exprs/hybrid_set.h
+++ b/be/src/exprs/hybrid_set.h
@@ -17,21 +17,8 @@
 
 #pragma once
 
-#include <glog/logging.h>
-
-#include <type_traits>
-
-#include "common/exception.h"
-#include "common/object_pool.h"
-#include "common/status.h"
 #include "exprs/runtime_filter.h"
-#include "runtime/decimalv2_value.h"
-#include "runtime/define_primitive_type.h"
-#include "runtime/primitive_type.h"
-#include "vec/columns/column_nullable.h"
-#include "vec/columns/column_string.h"
-#include "vec/common/hash_table/phmap_fwd_decl.h"
-#include "vec/common/string_ref.h"
+#include "exprs/runtime_filter_convertor.h"
 
 namespace doris {
 
@@ -221,30 +208,19 @@ class HybridSetBase : public RuntimeFilterFuncBase {
     virtual bool find(const void* data, size_t) const = 0;
 
     virtual void find_batch(const doris::vectorized::IColumn& column, size_t rows,
-                            doris::vectorized::ColumnUInt8::Container& results) {
-        LOG(FATAL) << "HybridSetBase not support find_batch";
-        __builtin_unreachable();
-    }
-
+                            doris::vectorized::ColumnUInt8::Container& results) = 0;
     virtual void find_batch_negative(const doris::vectorized::IColumn& column, size_t rows,
-                                     doris::vectorized::ColumnUInt8::Container& results) {
-        LOG(FATAL) << "HybridSetBase not support find_batch_negative";
-        __builtin_unreachable();
-    }
-
+                                     doris::vectorized::ColumnUInt8::Container& results) = 0;
     virtual void find_batch_nullable(const doris::vectorized::IColumn& column, size_t rows,
                                      const doris::vectorized::NullMap& null_map,
-                                     doris::vectorized::ColumnUInt8::Container& results) {
-        LOG(FATAL) << "HybridSetBase not support find_batch_nullable";
-        __builtin_unreachable();
-    }
+                                     doris::vectorized::ColumnUInt8::Container& results) = 0;
 
-    virtual void find_batch_nullable_negative(const doris::vectorized::IColumn& column, size_t rows,
-                                              const doris::vectorized::NullMap& null_map,
-                                              doris::vectorized::ColumnUInt8::Container& results) {
-        LOG(FATAL) << "HybridSetBase not support find_batch_nullable_negative";
-        __builtin_unreachable();
-    }
+    virtual void find_batch_nullable_negative(
+            const doris::vectorized::IColumn& column, size_t rows,
+            const doris::vectorized::NullMap& null_map,
+            doris::vectorized::ColumnUInt8::Container& results) = 0;
+
+    virtual void to_pb(PInFilter* filter) = 0;
 
     class IteratorBase {
     public:
@@ -261,26 +237,6 @@ class HybridSetBase : public RuntimeFilterFuncBase {
     bool _contains_null = false;
 };
 
-template <typename Type>
-const Type* check_and_get_hybrid_set(const HybridSetBase& column) {
-    return typeid_cast<const Type*>(&column);
-}
-
-template <typename Type>
-const Type* check_and_get_hybrid_set(const HybridSetBase* column) {
-    return typeid_cast<const Type*>(column);
-}
-
-template <typename Type>
-bool check_hybrid_set(const HybridSetBase& column) {
-    return check_and_get_hybrid_set<Type>(&column);
-}
-
-template <typename Type>
-bool check_hybrid_set(const HybridSetBase* column) {
-    return check_and_get_hybrid_set<Type>(column);
-}
-
 template <PrimitiveType T,
           typename _ContainerType = DynamicContainer<typename PrimitiveTypeTraits<T>::CppType>,
           typename _ColumnType = typename PrimitiveTypeTraits<T>::ColumnType>
@@ -409,6 +365,14 @@ class HybridSet : public HybridSetBase {
 
     ContainerType* get_inner_set() { return &_set; }
 
+    void set_pb(PInFilter* filter, auto f) {
+        for (auto v : _set) {
+            f(filter->add_values(), v);
+        }
+    }
+
+    void to_pb(PInFilter* filter) override { set_pb(filter, get_convertor<ElementType>()); }
+
 private:
     ContainerType _set;
     ObjectPool _pool;
@@ -569,6 +533,14 @@ class StringSet : public HybridSetBase {
 
     ContainerType* get_inner_set() { return &_set; }
 
+    void set_pb(PInFilter* filter, auto f) {
+        for (const auto& v : _set) {
+            f(filter->add_values(), v);
+        }
+    }
+
+    void to_pb(PInFilter* filter) override { set_pb(filter, get_convertor<std::string>()); }
+
 private:
     ContainerType _set;
     ObjectPool _pool;
@@ -735,6 +707,10 @@ class StringValueSet : public HybridSetBase {
 
     ContainerType* get_inner_set() { return &_set; }
 
+    void to_pb(PInFilter* filter) override {
+        throw Exception(ErrorCode::INTERNAL_ERROR, "StringValueSet do not support to_pb");
+    }
+
 private:
     ContainerType _set;
     ObjectPool _pool;
diff --git a/be/src/exprs/minmax_predicate.h b/be/src/exprs/minmax_predicate.h
index 377b33696c82b9..6c5d8a2d3c4bd5 100644
--- a/be/src/exprs/minmax_predicate.h
+++ b/be/src/exprs/minmax_predicate.h
@@ -17,16 +17,8 @@
 
 #pragma once
 
-#include <type_traits>
-
-#include "common/object_pool.h"
 #include "exprs/runtime_filter.h"
-#include "runtime/type_limit.h"
-#include "vec/columns/column.h"
-#include "vec/columns/column_nullable.h"
-#include "vec/columns/column_string.h"
-#include "vec/common/assert_cast.h"
-#include "vec/common/string_ref.h"
+#include "exprs/runtime_filter_convertor.h"
 
 namespace doris {
 // only used in Runtime Filter
@@ -45,6 +37,8 @@ class MinMaxFuncBase : public RuntimeFilterFuncBase {
 
     void set_contain_null() { _contain_null = true; }
 
+    virtual void to_pb(PMinMaxFilter* filter) = 0;
+
 protected:
     bool _contain_null = false;
 };
@@ -165,6 +159,17 @@ class MinMaxNumFunc : public MinMaxFuncBase {
         return Status::OK();
     }
 
+    void set_pb(PMinMaxFilter* filter, auto f) {
+        if constexpr (NeedMin) {
+            f(filter->mutable_min_val(), _min);
+        }
+        if constexpr (NeedMax) {
+            f(filter->mutable_max_val(), _max);
+        }
+    }
+
+    void to_pb(PMinMaxFilter* filter) override { set_pb(filter, get_convertor<T>()); }
+
 protected:
     T _max = type_limit<T>::min();
     T _min = type_limit<T>::max();
diff --git a/be/src/exprs/runtime_filter.cpp b/be/src/exprs/runtime_filter.cpp
index bac14b616b2ce6..6eed49e8567e1c 100644
--- a/be/src/exprs/runtime_filter.cpp
+++ b/be/src/exprs/runtime_filter.cpp
@@ -903,17 +903,10 @@ class RuntimePredicateWrapper {
         return Status::InternalError("not support!");
     }
 
-    HybridSetBase::IteratorBase* get_in_filter_iterator() { return _context->hybrid_set->begin(); }
-
     void get_bloom_filter_desc(char** data, int* filter_length) {
         _context->bloom_filter_func->get_data(data, filter_length);
     }
 
-    void get_minmax_filter_desc(void** min_data, void** max_data) {
-        *min_data = _context->minmax_func->get_min();
-        *max_data = _context->minmax_func->get_max();
-    }
-
     PrimitiveType column_type() { return _column_return_type; }
 
     bool is_bloomfilter() const { return get_real_type() == RuntimeFilterType::BLOOM_FILTER; }
@@ -1560,17 +1553,6 @@ Status IRuntimeFilter::merge_from(const RuntimePredicateWrapper* wrapper) {
     return Status::OK();
 }
 
-template <typename T>
-void batch_copy(PInFilter* filter, HybridSetBase::IteratorBase* it,
-                void (*set_func)(PColumnValue*, const T*)) {
-    while (it->has_next()) {
-        const void* void_value = it->get_value();
-        auto origin_value = reinterpret_cast<const T*>(void_value);
-        set_func(filter->add_values(), origin_value);
-        it->next();
-    }
-}
-
 template <class T>
 Status IRuntimeFilter::serialize_impl(T* request, void** data, int* len) {
     auto real_runtime_filter_type = _wrapper->get_real_type();
@@ -1598,273 +1580,13 @@ Status IRuntimeFilter::serialize_impl(T* request, void** data, int* len) {
 }
 
 void IRuntimeFilter::to_protobuf(PInFilter* filter) {
-    auto column_type = _wrapper->column_type();
-    filter->set_column_type(to_proto(column_type));
-
-    auto* it = _wrapper->get_in_filter_iterator();
-    DCHECK(it != nullptr);
-
-    switch (column_type) {
-    case TYPE_BOOLEAN: {
-        batch_copy<bool>(filter, it, [](PColumnValue* column, const bool* value) {
-            column->set_boolval(*value);
-        });
-        return;
-    }
-    case TYPE_TINYINT: {
-        batch_copy<int8_t>(filter, it, [](PColumnValue* column, const int8_t* value) {
-            column->set_intval(*value);
-        });
-        return;
-    }
-    case TYPE_SMALLINT: {
-        batch_copy<int16_t>(filter, it, [](PColumnValue* column, const int16_t* value) {
-            column->set_intval(*value);
-        });
-        return;
-    }
-    case TYPE_INT: {
-        batch_copy<int32_t>(filter, it, [](PColumnValue* column, const int32_t* value) {
-            column->set_intval(*value);
-        });
-        return;
-    }
-    case TYPE_BIGINT: {
-        batch_copy<int64_t>(filter, it, [](PColumnValue* column, const int64_t* value) {
-            column->set_longval(*value);
-        });
-        return;
-    }
-    case TYPE_LARGEINT: {
-        batch_copy<int128_t>(filter, it, [](PColumnValue* column, const int128_t* value) {
-            column->set_stringval(LargeIntValue::to_string(*value));
-        });
-        return;
-    }
-    case TYPE_FLOAT: {
-        batch_copy<float>(filter, it, [](PColumnValue* column, const float* value) {
-            column->set_doubleval(*value);
-        });
-        return;
-    }
-    case TYPE_DOUBLE: {
-        batch_copy<double>(filter, it, [](PColumnValue* column, const double* value) {
-            column->set_doubleval(*value);
-        });
-        return;
-    }
-    case TYPE_DATEV2: {
-        batch_copy<DateV2Value<DateV2ValueType>>(
-                filter, it, [](PColumnValue* column, const DateV2Value<DateV2ValueType>* value) {
-                    column->set_intval(*reinterpret_cast<const int32_t*>(value));
-                });
-        return;
-    }
-    case TYPE_DATETIMEV2: {
-        batch_copy<DateV2Value<DateTimeV2ValueType>>(
-                filter, it,
-                [](PColumnValue* column, const DateV2Value<DateTimeV2ValueType>* value) {
-                    column->set_longval(*reinterpret_cast<const int64_t*>(value));
-                });
-        return;
-    }
-    case TYPE_DATE:
-    case TYPE_DATETIME: {
-        batch_copy<VecDateTimeValue>(filter, it,
-                                     [](PColumnValue* column, const VecDateTimeValue* value) {
-                                         char convert_buffer[30];
-                                         value->to_string(convert_buffer);
-                                         column->set_stringval(convert_buffer);
-                                     });
-        return;
-    }
-    case TYPE_DECIMALV2: {
-        batch_copy<DecimalV2Value>(filter, it,
-                                   [](PColumnValue* column, const DecimalV2Value* value) {
-                                       column->set_stringval(value->to_string());
-                                   });
-        return;
-    }
-    case TYPE_DECIMAL32: {
-        batch_copy<int32_t>(filter, it, [](PColumnValue* column, const int32_t* value) {
-            column->set_intval(*value);
-        });
-        return;
-    }
-    case TYPE_DECIMAL64: {
-        batch_copy<int64_t>(filter, it, [](PColumnValue* column, const int64_t* value) {
-            column->set_longval(*value);
-        });
-        return;
-    }
-    case TYPE_DECIMAL128I: {
-        batch_copy<int128_t>(filter, it, [](PColumnValue* column, const int128_t* value) {
-            column->set_stringval(LargeIntValue::to_string(*value));
-        });
-        return;
-    }
-    case TYPE_DECIMAL256: {
-        batch_copy<wide::Int256>(filter, it, [](PColumnValue* column, const wide::Int256* value) {
-            column->set_stringval(wide::to_string(*value));
-        });
-        return;
-    }
-    case TYPE_CHAR:
-    case TYPE_VARCHAR:
-    case TYPE_STRING: {
-        //const void* void_value = it->get_value();
-        //Now the get_value return void* is StringRef
-        batch_copy<StringRef>(filter, it, [](PColumnValue* column, const StringRef* value) {
-            column->set_stringval(value->to_string());
-        });
-        return;
-    }
-    case TYPE_IPV4: {
-        batch_copy<IPv4>(filter, it, [](PColumnValue* column, const IPv4* value) {
-            column->set_intval(*reinterpret_cast<const int32_t*>(value));
-        });
-        return;
-    }
-    case TYPE_IPV6: {
-        batch_copy<IPv6>(filter, it, [](PColumnValue* column, const IPv6* value) {
-            column->set_stringval(LargeIntValue::to_string(*value));
-        });
-        return;
-    }
-    default: {
-        throw Exception(ErrorCode::INTERNAL_ERROR,
-                        "runtime filter meet invalid PrimitiveType type {}", int(column_type));
-    }
-    }
+    filter->set_column_type(to_proto(_wrapper->column_type()));
+    _wrapper->_context->hybrid_set->to_pb(filter);
 }
 
 void IRuntimeFilter::to_protobuf(PMinMaxFilter* filter) {
-    void* min_data = nullptr;
-    void* max_data = nullptr;
-    _wrapper->get_minmax_filter_desc(&min_data, &max_data);
-    DCHECK(min_data != nullptr && max_data != nullptr);
     filter->set_column_type(to_proto(_wrapper->column_type()));
-
-    switch (_wrapper->column_type()) {
-    case TYPE_BOOLEAN: {
-        filter->mutable_min_val()->set_boolval(*reinterpret_cast<const bool*>(min_data));
-        filter->mutable_max_val()->set_boolval(*reinterpret_cast<const bool*>(max_data));
-        return;
-    }
-    case TYPE_TINYINT: {
-        filter->mutable_min_val()->set_intval(*reinterpret_cast<const int8_t*>(min_data));
-        filter->mutable_max_val()->set_intval(*reinterpret_cast<const int8_t*>(max_data));
-        return;
-    }
-    case TYPE_SMALLINT: {
-        filter->mutable_min_val()->set_intval(*reinterpret_cast<const int16_t*>(min_data));
-        filter->mutable_max_val()->set_intval(*reinterpret_cast<const int16_t*>(max_data));
-        return;
-    }
-    case TYPE_INT: {
-        filter->mutable_min_val()->set_intval(*reinterpret_cast<const int32_t*>(min_data));
-        filter->mutable_max_val()->set_intval(*reinterpret_cast<const int32_t*>(max_data));
-        return;
-    }
-    case TYPE_BIGINT: {
-        filter->mutable_min_val()->set_longval(*reinterpret_cast<const int64_t*>(min_data));
-        filter->mutable_max_val()->set_longval(*reinterpret_cast<const int64_t*>(max_data));
-        return;
-    }
-    case TYPE_LARGEINT: {
-        filter->mutable_min_val()->set_stringval(
-                LargeIntValue::to_string(*reinterpret_cast<const int128_t*>(min_data)));
-        filter->mutable_max_val()->set_stringval(
-                LargeIntValue::to_string(*reinterpret_cast<const int128_t*>(max_data)));
-        return;
-    }
-    case TYPE_FLOAT: {
-        filter->mutable_min_val()->set_doubleval(*reinterpret_cast<const float*>(min_data));
-        filter->mutable_max_val()->set_doubleval(*reinterpret_cast<const float*>(max_data));
-        return;
-    }
-    case TYPE_DOUBLE: {
-        filter->mutable_min_val()->set_doubleval(*reinterpret_cast<const double*>(min_data));
-        filter->mutable_max_val()->set_doubleval(*reinterpret_cast<const double*>(max_data));
-        return;
-    }
-    case TYPE_DATEV2: {
-        filter->mutable_min_val()->set_intval(*reinterpret_cast<const int32_t*>(min_data));
-        filter->mutable_max_val()->set_intval(*reinterpret_cast<const int32_t*>(max_data));
-        return;
-    }
-    case TYPE_DATETIMEV2: {
-        filter->mutable_min_val()->set_longval(*reinterpret_cast<const int64_t*>(min_data));
-        filter->mutable_max_val()->set_longval(*reinterpret_cast<const int64_t*>(max_data));
-        return;
-    }
-    case TYPE_DATE:
-    case TYPE_DATETIME: {
-        char convert_buffer[30];
-        reinterpret_cast<const VecDateTimeValue*>(min_data)->to_string(convert_buffer);
-        filter->mutable_min_val()->set_stringval(convert_buffer);
-        reinterpret_cast<const VecDateTimeValue*>(max_data)->to_string(convert_buffer);
-        filter->mutable_max_val()->set_stringval(convert_buffer);
-        return;
-    }
-    case TYPE_DECIMALV2: {
-        filter->mutable_min_val()->set_stringval(
-                reinterpret_cast<const DecimalV2Value*>(min_data)->to_string());
-        filter->mutable_max_val()->set_stringval(
-                reinterpret_cast<const DecimalV2Value*>(max_data)->to_string());
-        return;
-    }
-    case TYPE_DECIMAL32: {
-        filter->mutable_min_val()->set_intval(*reinterpret_cast<const int32_t*>(min_data));
-        filter->mutable_max_val()->set_intval(*reinterpret_cast<const int32_t*>(max_data));
-        return;
-    }
-    case TYPE_DECIMAL64: {
-        filter->mutable_min_val()->set_longval(*reinterpret_cast<const int64_t*>(min_data));
-        filter->mutable_max_val()->set_longval(*reinterpret_cast<const int64_t*>(max_data));
-        return;
-    }
-    case TYPE_DECIMAL128I: {
-        filter->mutable_min_val()->set_stringval(
-                LargeIntValue::to_string(*reinterpret_cast<const int128_t*>(min_data)));
-        filter->mutable_max_val()->set_stringval(
-                LargeIntValue::to_string(*reinterpret_cast<const int128_t*>(max_data)));
-        return;
-    }
-    case TYPE_DECIMAL256: {
-        filter->mutable_min_val()->set_stringval(
-                wide::to_string(*reinterpret_cast<const wide::Int256*>(min_data)));
-        filter->mutable_max_val()->set_stringval(
-                wide::to_string(*reinterpret_cast<const wide::Int256*>(max_data)));
-        return;
-    }
-    case TYPE_CHAR:
-    case TYPE_VARCHAR:
-    case TYPE_STRING: {
-        const auto* min_string_value = reinterpret_cast<const std::string*>(min_data);
-        filter->mutable_min_val()->set_stringval(*min_string_value);
-        const auto* max_string_value = reinterpret_cast<const std::string*>(max_data);
-        filter->mutable_max_val()->set_stringval(*max_string_value);
-        break;
-    }
-    case TYPE_IPV4: {
-        filter->mutable_min_val()->set_intval(*reinterpret_cast<const int32_t*>(min_data));
-        filter->mutable_max_val()->set_intval(*reinterpret_cast<const int32_t*>(max_data));
-        return;
-    }
-    case TYPE_IPV6: {
-        filter->mutable_min_val()->set_stringval(
-                LargeIntValue::to_string(*reinterpret_cast<const uint128_t*>(min_data)));
-        filter->mutable_max_val()->set_stringval(
-                LargeIntValue::to_string(*reinterpret_cast<const uint128_t*>(max_data)));
-        return;
-    }
-    default: {
-        throw Exception(ErrorCode::INTERNAL_ERROR,
-                        "runtime filter meet invalid PrimitiveType type {}",
-                        int(_wrapper->column_type()));
-    }
-    }
+    _wrapper->_context->minmax_func->to_pb(filter);
 }
 
 RuntimeFilterType IRuntimeFilter::get_real_type() {
diff --git a/be/src/exprs/runtime_filter_convertor.h b/be/src/exprs/runtime_filter_convertor.h
new file mode 100644
index 00000000000000..82df75e4abf329
--- /dev/null
+++ b/be/src/exprs/runtime_filter_convertor.h
@@ -0,0 +1,76 @@
+// 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 <gen_cpp/internal_service.pb.h>
+
+#include "runtime/large_int_value.h"
+#include "vec/common/string_ref.h"
+#include "vec/core/wide_integer.h"
+
+namespace doris {
+
+template <typename T>
+auto get_convertor() {
+    if constexpr (std::is_same_v<T, bool>) {
+        return [](PColumnValue* value, const T& data) { value->set_boolval(data); };
+    } else if constexpr (std::is_same_v<T, int8_t> || std::is_same_v<T, int16_t> ||
+                         std::is_same_v<T, int32_t> || std::is_same_v<T, uint32_t> ||
+                         std::is_same_v<T, vectorized::Decimal32>) {
+        return [](PColumnValue* value, const T& data) { value->set_intval(data); };
+    } else if constexpr (std::is_same_v<T, int64_t> || std::is_same_v<T, vectorized::Decimal64>) {
+        return [](PColumnValue* value, const T& data) { value->set_longval(data); };
+    } else if constexpr (std::is_same_v<T, float> || std::is_same_v<T, double>) {
+        return [](PColumnValue* value, const T& data) { value->set_doubleval(data); };
+    } else if constexpr (std::is_same_v<T, int128_t> || std::is_same_v<T, uint128_t> ||
+                         std::is_same_v<T, vectorized::Decimal128V3>) {
+        return [](PColumnValue* value, const T& data) {
+            value->set_stringval(LargeIntValue::to_string(data));
+        };
+    } else if constexpr (std::is_same_v<T, vectorized::Decimal256>) {
+        return [](PColumnValue* value, const T& data) {
+            value->set_stringval(wide::to_string(wide::Int256(data)));
+        };
+    } else if constexpr (std::is_same_v<T, std::string>) {
+        return [](PColumnValue* value, const T& data) { value->set_stringval(data); };
+    } else if constexpr (std::is_same_v<T, StringRef> ||
+                         std::is_same_v<T, vectorized::Decimal128V2> ||
+                         std::is_same_v<T, DecimalV2Value>) {
+        return [](PColumnValue* value, const T& data) { value->set_stringval(data.to_string()); };
+    } else if constexpr (std::is_same_v<T, VecDateTimeValue>) {
+        return [](PColumnValue* value, const T& data) {
+            char convert_buffer[30];
+            data.to_string(convert_buffer);
+            value->set_stringval(convert_buffer);
+        };
+    } else if constexpr (std::is_same_v<T, DateV2Value<DateV2ValueType>>) {
+        return [](PColumnValue* value, const T& data) {
+            value->set_intval(data.to_date_int_val());
+        };
+    } else if constexpr (std::is_same_v<T, DateV2Value<DateTimeV2ValueType>>) {
+        return [](PColumnValue* value, const T& data) {
+            value->set_longval(data.to_date_int_val());
+        };
+    } else {
+        throw Exception(ErrorCode::INTERNAL_ERROR,
+                        "runtime filter data convertor meet invalid type {}", typeid(T).name());
+        return [](PColumnValue* value, const T& data) {};
+    }
+}
+
+} // namespace doris
diff --git a/be/src/pipeline/exec/hashjoin_build_sink.cpp b/be/src/pipeline/exec/hashjoin_build_sink.cpp
index 7d3f4da935099e..5c1fa9daa0d837 100644
--- a/be/src/pipeline/exec/hashjoin_build_sink.cpp
+++ b/be/src/pipeline/exec/hashjoin_build_sink.cpp
@@ -157,7 +157,8 @@ Status HashJoinBuildSinkLocalState::close(RuntimeState* state, Status exec_statu
         }
     }
     SCOPED_TIMER(_publish_runtime_filter_timer);
-    RETURN_IF_ERROR(_runtime_filter_slots->publish(state, !_should_build_hash_table));
+    RETURN_IF_ERROR_OR_CATCH_EXCEPTION(
+            _runtime_filter_slots->publish(state, !_should_build_hash_table));
     return Base::close(state, exec_status);
 }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index e15133246190f9..2c09ee73f57168 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -2433,6 +2433,8 @@ public void initFuzzyModeVariables() {
 
             this.runtimeFilterType = 1 << randomInt;
             this.enableParallelScan = Config.pull_request_id % 2 == 0 ? randomInt % 2 == 0 : randomInt % 1 == 0;
+            this.enableRuntimeFilterPrune = (randomInt % 2) == 0;
+
             switch (randomInt) {
                 case 0:
                     this.parallelScanMaxScannersCount = 32;
diff --git a/regression-test/data/datatype_p0/ip/ip_rf/test_ip_rf.out b/regression-test/data/datatype_p0/ip/ip_rf/test_ip_rf.out
index 7e8a59791f636e..ae05f362144d0a 100644
--- a/regression-test/data/datatype_p0/ip/ip_rf/test_ip_rf.out
+++ b/regression-test/data/datatype_p0/ip/ip_rf/test_ip_rf.out
@@ -29,3 +29,33 @@
 -- !sql --
 2
 
+-- !sql --
+2
+
+-- !sql --
+2
+
+-- !sql --
+2
+
+-- !sql --
+2
+
+-- !sql --
+2
+
+-- !sql --
+2
+
+-- !sql --
+2
+
+-- !sql --
+2
+
+-- !sql --
+2
+
+-- !sql --
+2
+
diff --git a/regression-test/suites/datatype_p0/ip/ip_rf/test_ip_rf.groovy b/regression-test/suites/datatype_p0/ip/ip_rf/test_ip_rf.groovy
index c7f6c30481eb0b..8ca5e4a42c5486 100644
--- a/regression-test/suites/datatype_p0/ip/ip_rf/test_ip_rf.groovy
+++ b/regression-test/suites/datatype_p0/ip/ip_rf/test_ip_rf.groovy
@@ -16,6 +16,8 @@
 // specific language governing permissions and limitations
 // under the License.
 suite("test_ip_rf") {
+    sql "set enable_runtime_filter_prune=false;"
+
     sql """ DROP TABLE IF EXISTS ip_test """
     sql """ DROP TABLE IF EXISTS ip_test2 """
     sql """
@@ -57,17 +59,26 @@ suite("test_ip_rf") {
     sql "set runtime_filter_type=0;"
     qt_sql "select count(*) from ip_test a, ip_test2 b where a.ip_v4=b.ip_v4;"
     qt_sql "select count(*) from ip_test a, ip_test2 b where a.ip_v6=b.ip_v6;"
+    qt_sql "select count(*) from ip_test a, ip_test2 b, ip_test c where a.ip_v4=b.ip_v4 and c.ip_v4=b.ip_v4;"
+    qt_sql "select count(*) from ip_test a, ip_test2 b, ip_test c where a.ip_v6=b.ip_v6 and c.ip_v6=b.ip_v6;"
     sql "set runtime_filter_type=1;"
     qt_sql "select count(*) from ip_test a, ip_test2 b where a.ip_v4=b.ip_v4;"
     qt_sql "select count(*) from ip_test a, ip_test2 b where a.ip_v6=b.ip_v6;"
+    qt_sql "select count(*) from ip_test a, ip_test2 b, ip_test c where a.ip_v4=b.ip_v4 and c.ip_v4=b.ip_v4;"
+    qt_sql "select count(*) from ip_test a, ip_test2 b, ip_test c where a.ip_v6=b.ip_v6 and c.ip_v6=b.ip_v6;"
     sql "set runtime_filter_type=2;"
     qt_sql "select count(*) from ip_test a, ip_test2 b where a.ip_v4=b.ip_v4;"
     qt_sql "select count(*) from ip_test a, ip_test2 b where a.ip_v6=b.ip_v6;"
+    qt_sql "select count(*) from ip_test a, ip_test2 b, ip_test c where a.ip_v4=b.ip_v4 and c.ip_v4=b.ip_v4;"
+    qt_sql "select count(*) from ip_test a, ip_test2 b, ip_test c where a.ip_v6=b.ip_v6 and c.ip_v6=b.ip_v6;"
     sql "set runtime_filter_type=4;"
     qt_sql "select count(*) from ip_test a, ip_test2 b where a.ip_v4=b.ip_v4;"
     qt_sql "select count(*) from ip_test a, ip_test2 b where a.ip_v6=b.ip_v6;"
+    qt_sql "select count(*) from ip_test a, ip_test2 b, ip_test c where a.ip_v4=b.ip_v4 and c.ip_v4=b.ip_v4;"
+    qt_sql "select count(*) from ip_test a, ip_test2 b, ip_test c where a.ip_v6=b.ip_v6 and c.ip_v6=b.ip_v6;"
     sql "set runtime_filter_type=8;"
     qt_sql "select count(*) from ip_test a, ip_test2 b where a.ip_v4=b.ip_v4;"
     qt_sql "select count(*) from ip_test a, ip_test2 b where a.ip_v6=b.ip_v6;"
-   
+    qt_sql "select count(*) from ip_test a, ip_test2 b, ip_test c where a.ip_v4=b.ip_v4 and c.ip_v4=b.ip_v4;"
+    qt_sql "select count(*) from ip_test a, ip_test2 b, ip_test c where a.ip_v6=b.ip_v6 and c.ip_v6=b.ip_v6;"
 }
diff --git a/regression-test/suites/nereids_p0/join/test_join_on.groovy b/regression-test/suites/nereids_p0/join/test_join_on.groovy
index 2415a3496cf718..ebd43a544cf9eb 100644
--- a/regression-test/suites/nereids_p0/join/test_join_on.groovy
+++ b/regression-test/suites/nereids_p0/join/test_join_on.groovy
@@ -39,16 +39,16 @@ suite("test_join_on", "nereids_p0") {
     qt_sql """ select * from join_on order by k1; """
     test {
         sql """ select * from join_on as j1 inner join join_on as j2 on j1.d_array = j2.d_array; """
-        exception "meet invalid type, type=Array(Nullable(Int32))"
+        exception "errCode = 2"
     }
     test {
         sql """ select * from join_on as j1 inner join join_on as j2 on j1.hll_col = j2.hll_col; """
-        exception "data type HLL could not used in ComparisonPredicate"
+        exception "errCode = 2"
     }
 
     test {
         sql """ select * from join_on as j1 inner join join_on as j2 on j1.k3 = j2.k3; """
-        exception "data type BITMAP could not used in ComparisonPredicate"
+        exception "errCode = 2"
     }
 
     test {