[Bug](runtime-filter) fix core dump on rf between varchar and char (a…

…pache#43758)

### What problem does this PR solve?
fix core dump on rf between varchar and char
Problem Summary:
F20241112 15:33:12.916148 3455401 assert_cast.h:48] Bad cast from
type:doris::BloomFilterFunc<(doris::PrimitiveType)15>* to
doris::BloomFilterFunc<(doris::PrimitiveType)10>*

```c++
 0# doris::signal::(anonymous namespace)::FailureSignalHandler(int, siginfo_t*, void*) at /mnt/disk1/xiaolei/incubator-doris/be/src/common/signal_handler.h:421
 1# 0x00007F73ACB75B50 in /lib64/libc.so.6
 2# gsignal in /lib64/libc.so.6
 3# __GI_abort in /lib64/libc.so.6
 4# 0x00005645EDABC138 in /mnt/disk1/xiaolei/incubator-doris/output/be/lib/doris_be
 5# 0x00005645EDAAD89A in /mnt/disk1/xiaolei/incubator-doris/output/be/lib/doris_be
 6# google::LogMessage::SendToLog() in /mnt/disk1/xiaolei/incubator-doris/output/be/lib/doris_be
 7# google::LogMessage::Flush() in /mnt/disk1/xiaolei/incubator-doris/output/be/lib/doris_be
 8# google::LogMessageFatal::~LogMessageFatal() in /mnt/disk1/xiaolei/incubator-doris/output/be/lib/doris_be
 9# doris::BloomFilterFunc<(doris::PrimitiveType)10>* assert_cast<doris::BloomFilterFunc<(doris::PrimitiveType)10>*, (TypeCheckOnRelease)1, doris::BloomFilterFuncBase*>(doris::BloomFilterFuncBase*&&)::{lambda(auto:1&&)#1}::operator()<doris::BloomFilterFuncBase*>(doris::BloomFilterFuncBase*&&) const at /mnt/disk1/xiaolei/incubator-doris/be/src/vec/common/assert_cast.h:48
10# doris::BloomFilterFunc<(doris::PrimitiveType)10>* assert_cast<doris::BloomFilterFunc<(doris::PrimitiveType)10>*, (TypeCheckOnRelease)1, doris::BloomFilterFuncBase*>(doris::BloomFilterFuncBase*&&) at /mnt/disk1/xiaolei/incubator-doris/be/src/vec/common/assert_cast.h:64
11# doris::BloomFilterColumnPredicate<(doris::PrimitiveType)10>::BloomFilterColumnPredicate(unsigned int, std::shared_ptr<doris::BloomFilterFuncBase> const&) at /mnt/disk1/xiaolei/incubator-doris/be/src/olap/bloom_filter_predicate.h:44
12# doris::ColumnPredicate* doris::create_olap_column_predicate<(doris::PrimitiveType)10>(unsigned int, std::shared_ptr<doris::BloomFilterFuncBase> const&, int, doris::TabletColumn const*) at /mnt/disk1/xiaolei/incubator-doris/be/src/exprs/create_predicate_function.h:237
13# doris::ColumnPredicate* doris::create_column_predicate<doris::BloomFilterFuncBase>(unsigned int, std::shared_ptr<doris::BloomFilterFuncBase> const&, doris::FieldType, int, doris::TabletColumn const*) at /mnt/disk1/xiaolei/incubator-doris/be/src/exprs/create_predicate_function.h:290
14# doris::TabletReader::_parse_to_predicate(std::pair<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >, std::shared_ptr<doris::BloomFilterFuncBase> > const&) at /mnt/disk1/xiaolei/incubator-doris/be/src/olap/tablet_reader.cpp:582
15# doris::TabletReader::_init_conditions_param(doris::TabletReader::ReaderParams const&) at /mnt/disk1/xiaolei/incubator-doris/be/src/olap/tablet_reader.cpp:511
16# doris::TabletReader::_init_params(doris::TabletReader::ReaderParams const&) at /mnt/disk1/xiaolei/incubator-doris/be/src/olap/tablet_reader.cpp:294
17# doris::TabletReader::init(doris::TabletReader::ReaderParams const&) at /mnt/disk1/xiaolei/incubator-doris/be/src/olap/tablet_reader.cpp:125
18# doris::vectorized::BlockReader::init(doris::TabletReader::ReaderParams const&) at /mnt/disk1/xiaolei/incubator-doris/be/src/vec/olap/block_reader.cpp:193
19# doris::vectorized::NewOlapScanner::open(doris::RuntimeState*) at /mnt/disk1/xiaolei/incubator-doris/be/src/vec/exec/scan/new_olap_scanner.cpp:231
20# doris::vectorized::ScannerScheduler::_scanner_scan(std::shared_ptr<doris::vectorized::ScannerContext>, std::shared_ptr<doris::vectorized::ScanTask>) at /mnt/disk1/xiaolei/incubator-doris/be/src/vec/exec/scan/scanner_scheduler.cpp:247
```

be/src/exprs/create_predicate_function.h

@@ -234,7 +234,8 @@ ColumnPredicate* create_olap_column_predicate(uint32_t column_id,
     std::shared_ptr<BloomFilterFuncBase> filter_olap;
     filter_olap.reset(create_bloom_filter(PT));
     filter_olap->light_copy(filter.get());
-    return new BloomFilterColumnPredicate<PT>(column_id, filter);
+    // create a new filter to match the input filter and PT. For example, filter may be varchar, but PT is char
+    return new BloomFilterColumnPredicate<PT>(column_id, filter_olap);
 }
 
 template <PrimitiveType PT>

be/src/olap/bitmap_filter_predicate.h

@@ -30,8 +30,6 @@
 #include "vec/exprs/vruntimefilter_wrapper.h"
 
 namespace doris {
-
-// only use in runtime filter and segment v2
 template <PrimitiveType T>
 class BitmapFilterColumnPredicate : public ColumnPredicate {
 public:
@@ -89,7 +87,7 @@ class BitmapFilterColumnPredicate : public ColumnPredicate {
 
         uint16_t new_size = 0;
         new_size = _specific_filter->find_fixed_len_olap_engine(
-                (char*)reinterpret_cast<
+                (char*)assert_cast<
                         const vectorized::PredicateColumnType<PredicateEvaluateType<T>>*>(&column)
                         ->get_data()
                         .data(),
@@ -113,7 +111,7 @@ uint16_t BitmapFilterColumnPredicate<T>::_evaluate_inner(const vectorized::IColu
                                                          uint16_t* sel, uint16_t size) const {
     uint16_t new_size = 0;
     if (column.is_nullable()) {
-        const auto* nullable_col = reinterpret_cast<const vectorized::ColumnNullable*>(&column);
+        const auto* nullable_col = assert_cast<const vectorized::ColumnNullable*>(&column);
         const auto& null_map_data = nullable_col->get_null_map_column().get_data();
         new_size =
                 evaluate<true>(nullable_col->get_nested_column(), null_map_data.data(), sel, size);

be/src/olap/bloom_filter_predicate.h

@@ -30,8 +30,6 @@
 
 namespace doris {
 
-// only use in runtime filter and segment v2
-
 template <PrimitiveType T>
 class BloomFilterColumnPredicate : public ColumnPredicate {
 public:
@@ -41,7 +39,7 @@ class BloomFilterColumnPredicate : public ColumnPredicate {
                                const std::shared_ptr<BloomFilterFuncBase>& filter)
             : ColumnPredicate(column_id),
               _filter(filter),
-              _specific_filter(reinterpret_cast<SpecificFilter*>(_filter.get())) {}
+              _specific_filter(assert_cast<SpecificFilter*>(_filter.get())) {}
     ~BloomFilterColumnPredicate() override = default;
 
     PredicateType type() const override { return PredicateType::BF; }
@@ -105,7 +103,7 @@ template <PrimitiveType T>
 uint16_t BloomFilterColumnPredicate<T>::_evaluate_inner(const vectorized::IColumn& column,
                                                         uint16_t* sel, uint16_t size) const {
     if (column.is_nullable()) {
-        const auto* nullable_col = reinterpret_cast<const vectorized::ColumnNullable*>(&column);
+        const auto* nullable_col = assert_cast<const vectorized::ColumnNullable*>(&column);
         const auto& null_map_data = nullable_col->get_null_map_column().get_data();
         return evaluate<true>(nullable_col->get_nested_column(), null_map_data.data(), sel, size);
     } else {

be/src/olap/comparison_predicate.h

@@ -274,7 +274,7 @@ class ComparisonPredicateBase : public ColumnPredicate {
             const auto* nullable_column_ptr =
                     vectorized::check_and_get_column<vectorized::ColumnNullable>(column);
             const auto& nested_column = nullable_column_ptr->get_nested_column();
-            const auto& null_map = reinterpret_cast<const vectorized::ColumnUInt8&>(
+            const auto& null_map = assert_cast<const vectorized::ColumnUInt8&>(
                                            nullable_column_ptr->get_null_map_column())
                                            .get_data();
 
@@ -379,7 +379,7 @@ class ComparisonPredicateBase : public ColumnPredicate {
             const auto* nullable_column_ptr =
                     vectorized::check_and_get_column<vectorized::ColumnNullable>(column);
             const auto& nested_column = nullable_column_ptr->get_nested_column();
-            const auto& null_map = reinterpret_cast<const vectorized::ColumnUInt8&>(
+            const auto& null_map = assert_cast<const vectorized::ColumnUInt8&>(
                                            nullable_column_ptr->get_null_map_column())
                                            .get_data();
 
@@ -465,7 +465,7 @@ class ComparisonPredicateBase : public ColumnPredicate {
             const auto* nullable_column_ptr =
                     vectorized::check_and_get_column<vectorized::ColumnNullable>(column);
             const auto& nested_column = nullable_column_ptr->get_nested_column();
-            const auto& null_map = reinterpret_cast<const vectorized::ColumnUInt8&>(
+            const auto& null_map = assert_cast<const vectorized::ColumnUInt8&>(
                                            nullable_column_ptr->get_null_map_column())
                                            .get_data();
 

be/src/olap/in_list_predicate.h

@@ -232,9 +232,9 @@ class InListPredicateBase : public ColumnPredicate {
         if (column.is_nullable()) {
             const auto* nullable_col =
                     vectorized::check_and_get_column<vectorized::ColumnNullable>(column);
-            const auto& null_bitmap = reinterpret_cast<const vectorized::ColumnUInt8&>(
-                                              nullable_col->get_null_map_column())
-                                              .get_data();
+            const auto& null_bitmap =
+                    assert_cast<const vectorized::ColumnUInt8&>(nullable_col->get_null_map_column())
+                            .get_data();
             const auto& nested_col = nullable_col->get_nested_column();
 
             if (_opposite) {
@@ -355,9 +355,9 @@ class InListPredicateBase : public ColumnPredicate {
         if (column.is_nullable()) {
             const auto* nullable_col =
                     vectorized::check_and_get_column<vectorized::ColumnNullable>(column);
-            const auto& null_map = reinterpret_cast<const vectorized::ColumnUInt8&>(
-                                           nullable_col->get_null_map_column())
-                                           .get_data();
+            const auto& null_map =
+                    assert_cast<const vectorized::ColumnUInt8&>(nullable_col->get_null_map_column())
+                            .get_data();
             const auto& nested_col = nullable_col->get_nested_column();
 
             if (_opposite) {

regression-test/data/correctness_p0/test_runtime_filter.out

@@ -0,0 +1,33 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !test --
+a	a	a	a
+b	b	b	b
+
+-- !test --
+a	a	a	a
+b	b	b	b
+
+-- !test --
+a	a	a	a
+b	b	b	b
+
+-- !test --
+a	a	a	a
+b	b	b	b
+
+-- !test --
+a	a	a	a
+b	b	b	b
+
+-- !test --
+a	a	a	a
+b	b	b	b
+
+-- !test --
+a	a	a	a
+b	b	b	b
+
+-- !test --
+a	a	a	a
+b	b	b	b
+

regression-test/suites/correctness_p0/test_runtime_filter.groovy

@@ -20,7 +20,7 @@
 // and modified by Doris.
 
 suite("test_runtime_filter") {
-
+    sql "set enable_runtime_filter_prune=false;"
     sql """ DROP TABLE IF EXISTS rf_tblA """
     sql """
             CREATE TABLE IF NOT EXISTS rf_tblA (
@@ -109,5 +109,45 @@ suite("test_runtime_filter") {
         contains "runtime filters: RF001[max] <- c"
         contains "runtime filters: RF002[max] <- c"
 
-    }   
+    }
+
+    sql """ DROP TABLE IF EXISTS v_table """
+    sql """
+            create table v_table (
+                kc char(100),
+                kv varchar(100)
+            )
+            duplicate key (kc)
+            distributed BY hash(kv) buckets 1
+            properties("replication_num" = "1");
+        """
+
+    sql """ DROP TABLE IF EXISTS c_table """
+    sql """
+            create table c_table (
+                kc char(100),
+                kv varchar(100)
+            )
+            duplicate key (kc)
+            distributed BY hash(kv) buckets 1
+            properties("replication_num" = "1");
+        """
+    sql """
+    insert into c_table values ('a','a'),('b','b');
+    """
+    sql """
+    insert into v_table values ('a','a'),('b','b'),('c','c');
+    """
+    sql "set runtime_filter_type='1';"
+    qt_test "select * from c_table,v_table where c_table.kc=v_table.kv;"
+    qt_test "select * from c_table,v_table where c_table.kv=v_table.kc;"
+    sql "set runtime_filter_type='2';"
+    qt_test "select * from c_table,v_table where c_table.kc=v_table.kv;"
+    qt_test "select * from c_table,v_table where c_table.kv=v_table.kc;"
+    sql "set runtime_filter_type='4';"
+    qt_test "select * from c_table,v_table where c_table.kc=v_table.kv;"
+    qt_test "select * from c_table,v_table where c_table.kv=v_table.kc;"
+    sql "set runtime_filter_type='8';"
+    qt_test "select * from c_table,v_table where c_table.kc=v_table.kv;"
+    qt_test "select * from c_table,v_table where c_table.kv=v_table.kc;"
 }

regression-test/suites/nereids_p0/join/test_runtimefilter_on_datev2.groovy

@@ -16,6 +16,7 @@
 // under the License.
 
 suite("test_runtimefilter_on_datev2", "nereids_p0") {
+    sql "set enable_runtime_filter_prune=false;"
     sql "SET enable_nereids_planner=true"
     sql "SET enable_fallback_to_original_planner=false"
     def dateTable = "dateTable"

regression-test/suites/nereids_p0/join/test_runtimefilter_on_decimal.groovy

@@ -16,6 +16,7 @@
 // under the License.
 
 suite("test_runtimefilter_on_decimal", "nereids_p0") {
+    sql "set enable_runtime_filter_prune=false;"
     sql "SET enable_nereids_planner=true"
     sql "SET enable_fallback_to_original_planner=false"
 

regression-test/suites/query_p0/join/test_runtime_filter_boolean.groovy

@@ -16,6 +16,7 @@
 // under the License.
 
 suite("test_runtime_filter_boolean", "query_p0") {
+    sql "set enable_runtime_filter_prune=false;"
     sql "drop table if exists test_runtime_filter_boolean0;"
     sql """ create table test_runtime_filter_boolean0(k1 int, v1 boolean)
                 DUPLICATE KEY(`k1`)

regression-test/suites/query_p0/join/test_runtime_filter_decimal256.groovy

@@ -16,6 +16,7 @@
 // under the License.
 
 suite("test_runtime_filter_decimal256", "query_p0") {
+    sql "set enable_runtime_filter_prune=false;"
     sql "set enable_nereids_planner = true;"
     sql "set enable_decimal256 = true;"
     sql "set parallel_fragment_exec_instance_num = 4;"

regression-test/suites/query_p0/join/test_runtimefilter_2.groovy

@@ -15,9 +15,10 @@
  // specific language governing permissions and limitations
  // under the License.
 
- suite("test_runtimefilter_2", "query_p0") {
-     sql "drop table if exists t_ods_tpisyncjpa4_2;"
-     sql """ create table t_ods_tpisyncjpa4_2(INTERNAL_CODE varchar(50), USER_ID varchar(50), USER_NAME varchar(50), STATE_ID varchar(50)) distributed by hash(INTERNAL_CODE) properties('replication_num'='1'); """
+suite("test_runtimefilter_2", "query_p0") {
+    sql "set enable_runtime_filter_prune=false;"
+    sql "drop table if exists t_ods_tpisyncjpa4_2;"
+    sql """ create table t_ods_tpisyncjpa4_2(INTERNAL_CODE varchar(50), USER_ID varchar(50), USER_NAME varchar(50), STATE_ID varchar(50)) distributed by hash(INTERNAL_CODE) properties('replication_num'='1'); """
 
      sql """ insert into t_ods_tpisyncjpa4_2 values('1', '2', '3', '1');"""
 
@@ -41,4 +42,4 @@
         select DISTINCT         tpisyncjpa4.USER_ID as USER_ID,            tpisyncjpa4.USER_NAME as USER_NAME,       tpisyncjpp1.POST_ID AS "T4_POST_ID"   FROM t_ods_tpisyncjpa4_2 tpisyncjpa4 cross join [shuffle]       t_ods_tpisyncjpp1_2 tpisyncjpp1            inner join            (       SELECT         USER_ID,         MAX(INTERNAL_CODE) as INTERNAL_CODE       FROM        t_ods_tpisyncjpa4_2              WHERE         STATE_ID = '1'       GROUP BY         USER_ID     )jpa4         on tpisyncjpa4.USER_ID=jpa4.USER_ID and tpisyncjpa4.INTERNAL_CODE=jpa4.INTERNAL_CODE         inner join [shuffle]         (       SELECT         POST_ID,         MAX(INTERNAL_CODE) as INTERNAL_CODE       FROM             t_ods_tpisyncjpp1_2       WHERE         STATE_ID = '1'       GROUP BY         POST_ID     )jpp1         on tpisyncjpp1.POST_ID=jpp1.POST_ID and  tpisyncjpp1.INTERNAL_CODE=jpp1.INTERNAL_CODE         where tpisyncjpa4.USER_ID = tpisyncjpp1.USER_ID AND tpisyncjpp1.STATE_ID ='1' AND tpisyncjpa4.STATE_ID ='1'         AND tpisyncjpp1.POST_ID='BSDSAE1018';     
      """
 
- }
+}

regression-test/suites/query_p0/join/test_runtimefilter_on_datev2.groovy

@@ -23,6 +23,8 @@ suite("test_runtimefilter_on_datev2", "query_p0") {
     def dateV2Table2 = "dateV2Table2"
     def dateTimeV2Table2 = "dateTimeV2Table2"
 
+    sql "set enable_runtime_filter_prune=false;"
+
     sql "DROP TABLE IF EXISTS ${dateTable}"
     sql """
             CREATE TABLE IF NOT EXISTS ${dateTable} (