From 143cdaa0aa7156b39587d074b46d46a267ed97a5 Mon Sep 17 00:00:00 2001
From: BiteTheDDDDt <xl@selectdb.com>
Date: Thu, 14 Nov 2024 15:56:34 +0800
Subject: [PATCH 1/6] refactor of runtime filter void type data usage

---
 be/src/exprs/minmax_predicate.h |  13 ++++
 be/src/exprs/runtime_filter.cpp | 131 +-------------------------------
 be/src/exprs/runtime_filter.h   |  35 +++++++++
 3 files changed, 49 insertions(+), 130 deletions(-)

diff --git a/be/src/exprs/minmax_predicate.h b/be/src/exprs/minmax_predicate.h
index 377b33696c82b9..f057f637f04045 100644
--- a/be/src/exprs/minmax_predicate.h
+++ b/be/src/exprs/minmax_predicate.h
@@ -45,6 +45,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 +167,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 85f1c535c7038b..99bbdff89a28fa 100644
--- a/be/src/exprs/runtime_filter.cpp
+++ b/be/src/exprs/runtime_filter.cpp
@@ -904,11 +904,6 @@ class RuntimePredicateWrapper {
         _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; }
@@ -1703,132 +1698,8 @@ void IRuntimeFilter::to_protobuf(PInFilter* 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.h b/be/src/exprs/runtime_filter.h
index f5a069d9e55f85..204454d0de54ad 100644
--- a/be/src/exprs/runtime_filter.h
+++ b/be/src/exprs/runtime_filter.h
@@ -439,4 +439,39 @@ class RuntimeFilterWrapperHolder {
     WrapperPtr _wrapper;
 };
 
+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>) {
+        return [](PColumnValue* value, const T& data) { value->set_intval(data); };
+    } else if constexpr (std::is_same_v<T, int64_t>) {
+        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>) {
+        return [](PColumnValue* value, const T& data) {
+            value->set_stringval(LargeIntValue::to_string(data));
+        };
+    } else if constexpr (std::is_same_v<T, wide::Int256>) {
+        return [](PColumnValue* value, const T& data) {
+            value->set_stringval(wide::to_string(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, 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, DecimalV2Value>) {
+        return [](PColumnValue* value, const T& data) { value->set_stringval(data.to_string()); };
+    } else {
+        throw Exception(ErrorCode::INTERNAL_ERROR, "minmax meet invalid type {}", typeid(T).name());
+        return [](PColumnValue* value, const T& data) {};
+    }
+}
+
 } // namespace doris

From 68e2567bb8d0c42a2fc406c3d5893fe2f3fafcaf Mon Sep 17 00:00:00 2001
From: BiteTheDDDDt <xl@selectdb.com>
Date: Thu, 14 Nov 2024 16:43:16 +0800
Subject: [PATCH 2/6] update

---
 be/src/exprs/hybrid_set.h               |  64 +++++-----
 be/src/exprs/minmax_predicate.h         |  10 +-
 be/src/exprs/runtime_filter.cpp         | 153 +-----------------------
 be/src/exprs/runtime_filter.h           |  35 ------
 be/src/exprs/runtime_filter_convertor.h |  65 ++++++++++
 5 files changed, 98 insertions(+), 229 deletions(-)
 create mode 100644 be/src/exprs/runtime_filter_convertor.h

diff --git a/be/src/exprs/hybrid_set.h b/be/src/exprs/hybrid_set.h
index 6536ec2430fe08..635f6fc05edc4f 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:
@@ -409,6 +385,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 +553,10 @@ class StringSet : public HybridSetBase {
 
     ContainerType* get_inner_set() { return &_set; }
 
+    void to_pb(PInFilter* filter) override {
+        throw Exception(ErrorCode::INTERNAL_ERROR, "StringSet do not support to_pb");
+    }
+
 private:
     ContainerType _set;
     ObjectPool _pool;
@@ -735,6 +723,14 @@ class StringValueSet : 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<StringRef>()); }
+
 private:
     ContainerType _set;
     ObjectPool _pool;
diff --git a/be/src/exprs/minmax_predicate.h b/be/src/exprs/minmax_predicate.h
index f057f637f04045..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
diff --git a/be/src/exprs/runtime_filter.cpp b/be/src/exprs/runtime_filter.cpp
index 99bbdff89a28fa..dfe59d1f37b1a2 100644
--- a/be/src/exprs/runtime_filter.cpp
+++ b/be/src/exprs/runtime_filter.cpp
@@ -898,8 +898,6 @@ 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);
     }
@@ -1519,17 +1517,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();
@@ -1557,144 +1544,8 @@ 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) {
diff --git a/be/src/exprs/runtime_filter.h b/be/src/exprs/runtime_filter.h
index 204454d0de54ad..f5a069d9e55f85 100644
--- a/be/src/exprs/runtime_filter.h
+++ b/be/src/exprs/runtime_filter.h
@@ -439,39 +439,4 @@ class RuntimeFilterWrapperHolder {
     WrapperPtr _wrapper;
 };
 
-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>) {
-        return [](PColumnValue* value, const T& data) { value->set_intval(data); };
-    } else if constexpr (std::is_same_v<T, int64_t>) {
-        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>) {
-        return [](PColumnValue* value, const T& data) {
-            value->set_stringval(LargeIntValue::to_string(data));
-        };
-    } else if constexpr (std::is_same_v<T, wide::Int256>) {
-        return [](PColumnValue* value, const T& data) {
-            value->set_stringval(wide::to_string(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, 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, DecimalV2Value>) {
-        return [](PColumnValue* value, const T& data) { value->set_stringval(data.to_string()); };
-    } else {
-        throw Exception(ErrorCode::INTERNAL_ERROR, "minmax meet invalid type {}", typeid(T).name());
-        return [](PColumnValue* value, const T& data) {};
-    }
-}
-
 } // namespace doris
diff --git a/be/src/exprs/runtime_filter_convertor.h b/be/src/exprs/runtime_filter_convertor.h
new file mode 100644
index 00000000000000..52e06ed914941c
--- /dev/null
+++ b/be/src/exprs/runtime_filter_convertor.h
@@ -0,0 +1,65 @@
+// 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>) {
+        return [](PColumnValue* value, const T& data) { value->set_intval(data); };
+    } else if constexpr (std::is_same_v<T, int64_t>) {
+        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>) {
+        return [](PColumnValue* value, const T& data) {
+            value->set_stringval(LargeIntValue::to_string(data));
+        };
+    } else if constexpr (std::is_same_v<T, wide::Int256>) {
+        return [](PColumnValue* value, const T& data) {
+            value->set_stringval(wide::to_string(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>) {
+        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, DecimalV2Value>) {
+        return [](PColumnValue* value, const T& data) { value->set_stringval(data.to_string()); };
+    } else {
+        throw Exception(ErrorCode::INTERNAL_ERROR, "minmax meet invalid type {}", typeid(T).name());
+        return [](PColumnValue* value, const T& data) {};
+    }
+}
+
+} // namespace doris

From 75e956b7650560843fe9c124a5e1a2e80ced3a4e Mon Sep 17 00:00:00 2001
From: BiteTheDDDDt <xl@selectdb.com>
Date: Thu, 14 Nov 2024 18:30:59 +0800
Subject: [PATCH 3/6] fix

---
 be/src/exprs/runtime_filter_convertor.h      | 3 ++-
 be/src/pipeline/exec/hashjoin_build_sink.cpp | 2 +-
 2 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/be/src/exprs/runtime_filter_convertor.h b/be/src/exprs/runtime_filter_convertor.h
index 52e06ed914941c..f9e3809eded7cf 100644
--- a/be/src/exprs/runtime_filter_convertor.h
+++ b/be/src/exprs/runtime_filter_convertor.h
@@ -36,7 +36,8 @@ auto get_convertor() {
         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>) {
+    } 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));
         };
diff --git a/be/src/pipeline/exec/hashjoin_build_sink.cpp b/be/src/pipeline/exec/hashjoin_build_sink.cpp
index 37de9ac93d839f..5d720c755bf89e 100644
--- a/be/src/pipeline/exec/hashjoin_build_sink.cpp
+++ b/be/src/pipeline/exec/hashjoin_build_sink.cpp
@@ -157,7 +157,7 @@ Status HashJoinBuildSinkLocalState::close(RuntimeState* state, Status exec_statu
         }
     }
     SCOPED_TIMER(_publish_runtime_filter_timer);
-    RETURN_IF_ERROR(_runtime_filter_slots->publish(!_should_build_hash_table));
+    RETURN_IF_ERROR_OR_CATCH_EXCEPTION(_runtime_filter_slots->publish(!_should_build_hash_table));
     return Base::close(state, exec_status);
 }
 

From 2d1e1d7c54de01a3a3d25205f0ca32a89af5ddc6 Mon Sep 17 00:00:00 2001
From: BiteTheDDDDt <xl@selectdb.com>
Date: Fri, 15 Nov 2024 00:53:55 +0800
Subject: [PATCH 4/6] update

---
 be/src/exprs/runtime_filter_convertor.h | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/be/src/exprs/runtime_filter_convertor.h b/be/src/exprs/runtime_filter_convertor.h
index f9e3809eded7cf..7d8249dbd5f400 100644
--- a/be/src/exprs/runtime_filter_convertor.h
+++ b/be/src/exprs/runtime_filter_convertor.h
@@ -30,9 +30,11 @@ 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, int32_t> || std::is_same_v<T, vectorized::Decimal32> ||
+                         std::is_same_v<T, DateV2Value<DateV2ValueType>>) {
         return [](PColumnValue* value, const T& data) { value->set_intval(data); };
-    } else if constexpr (std::is_same_v<T, int64_t>) {
+    } else if constexpr (std::is_same_v<T, int64_t> || std::is_same_v<T, vectorized::Decimal64> ||
+                         std::is_same_v<T, DateV2Value<DateTimeV2ValueType>>) {
         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); };
@@ -47,7 +49,8 @@ auto get_convertor() {
         };
     } 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>) {
+    } else if constexpr (std::is_same_v<T, StringRef> ||
+                         std::is_same_v<T, vectorized::Decimal128V2>) {
         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) {

From 88b015a220ffc7d3b60446552e4c6d11d8120b78 Mon Sep 17 00:00:00 2001
From: BiteTheDDDDt <xl@selectdb.com>
Date: Fri, 15 Nov 2024 10:07:14 +0800
Subject: [PATCH 5/6] update

---
 be/src/exprs/hybrid_set.h | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

diff --git a/be/src/exprs/hybrid_set.h b/be/src/exprs/hybrid_set.h
index 635f6fc05edc4f..2679873ebfbb36 100644
--- a/be/src/exprs/hybrid_set.h
+++ b/be/src/exprs/hybrid_set.h
@@ -553,10 +553,14 @@ class StringSet : public HybridSetBase {
 
     ContainerType* get_inner_set() { return &_set; }
 
-    void to_pb(PInFilter* filter) override {
-        throw Exception(ErrorCode::INTERNAL_ERROR, "StringSet do not support to_pb");
+    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<StringRef>()); }
+
 private:
     ContainerType _set;
     ObjectPool _pool;
@@ -723,14 +727,10 @@ class StringValueSet : 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 {
+        throw Exception(ErrorCode::INTERNAL_ERROR, "StringValueSet do not support to_pb");
     }
 
-    void to_pb(PInFilter* filter) override { set_pb(filter, get_convertor<StringRef>()); }
-
 private:
     ContainerType _set;
     ObjectPool _pool;

From a26679dfa37d7431e5685ff6ee76805edd1bebbd Mon Sep 17 00:00:00 2001
From: BiteTheDDDDt <xl@selectdb.com>
Date: Fri, 15 Nov 2024 11:41:32 +0800
Subject: [PATCH 6/6] update

---
 be/src/exprs/hybrid_set.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/be/src/exprs/hybrid_set.h b/be/src/exprs/hybrid_set.h
index 2679873ebfbb36..4aacfb678a1712 100644
--- a/be/src/exprs/hybrid_set.h
+++ b/be/src/exprs/hybrid_set.h
@@ -559,7 +559,7 @@ class StringSet : public HybridSetBase {
         }
     }
 
-    void to_pb(PInFilter* filter) override { set_pb(filter, get_convertor<StringRef>()); }
+    void to_pb(PInFilter* filter) override { set_pb(filter, get_convertor<std::string>()); }
 
 private:
     ContainerType _set;