[improvement](parquet) Optimize V2 sparse runtime-filter scans

be/src/core/data_type_serde/parquet_decode_source.cpp

@@ -19,19 +19,21 @@
 
 #include <cstring>
 #include <limits>
+#include <type_traits>
 
+#include "core/column/column_decimal.h"
 #include "core/column/column_string.h"
 #include "core/column/column_vector.h"
 #include "util/simd/parquet_kernels.h"
 
 namespace doris {
 namespace {
 
-template <PrimitiveType TYPE>
-bool try_gather_vector(IColumn& destination, const IColumn& dictionary, const uint32_t* indices,
-                       size_t num_values) {
-    using ColumnType = ColumnVector<TYPE>;
+template <typename ColumnType>
+bool try_gather_fixed_width(IColumn& destination, const IColumn& dictionary,
+                            const uint32_t* indices, size_t num_values) {
     using ValueType = typename ColumnType::value_type;
+    static_assert(std::is_trivially_copyable_v<ValueType>);
     if constexpr (sizeof(ValueType) != 4 && sizeof(ValueType) != 8) {
         return false;
     } else {
@@ -57,6 +59,12 @@ bool try_gather_vector(IColumn& destination, const IColumn& dictionary, const ui
     }
 }
 
+template <PrimitiveType TYPE>
+bool try_gather_vector(IColumn& destination, const IColumn& dictionary, const uint32_t* indices,
+                       size_t num_values) {
+    return try_gather_fixed_width<ColumnVector<TYPE>>(destination, dictionary, indices, num_values);
+}
+
 template <typename Offset>
 bool try_gather_strings(IColumn& destination, const IColumn& dictionary, const uint32_t* indices,
                         size_t num_values) {
@@ -108,11 +116,18 @@ bool try_simd_insert_parquet_dictionary_indices(IColumn& destination, const ICol
     TRY_PARQUET_GATHER(TYPE_DATETIME);
     TRY_PARQUET_GATHER(TYPE_DATEV2);
     TRY_PARQUET_GATHER(TYPE_DATETIMEV2);
+    TRY_PARQUET_GATHER(TYPE_TIMESTAMPTZ);
     TRY_PARQUET_GATHER(TYPE_IPV4);
     TRY_PARQUET_GATHER(TYPE_TIMEV2);
     TRY_PARQUET_GATHER(TYPE_UINT32);
     TRY_PARQUET_GATHER(TYPE_UINT64);
 #undef TRY_PARQUET_GATHER
+    // Keep every 4/8-byte POD column on this path: aliases such as TIMESTAMPTZ and decimals do not
+    // participate in the ordinary numeric dispatch and otherwise silently fall back to Field insertion.
+    if (try_gather_fixed_width<ColumnDecimal32>(destination, dictionary, indices, num_values) ||
+        try_gather_fixed_width<ColumnDecimal64>(destination, dictionary, indices, num_values)) {
+        return true;
+    }
     // String survivors have variable widths, so pre-size both buffers and copy each selected
     // dictionary slice exactly once instead of routing every id through generic Field insertion.
     if (try_gather_strings<UInt32>(destination, dictionary, indices, num_values) ||

be/src/exprs/bloom_filter_func.h

@@ -17,9 +17,12 @@
 
 #pragma once
 
+#include <cstring>
+
 #include "common/exception.h"
 #include "common/status.h"
 #include "core/column/column_dictionary.h"
+#include "core/field.h"
 #include "exec/runtime_filter/runtime_filter_definitions.h"
 #include "exprs/bloom_filter_func_impl.h"
 #include "exprs/filter_base.h"
@@ -136,6 +139,13 @@ class BloomFilterFuncBase : public FilterBase {
     virtual void find_fixed_len(const ColumnPtr& column, uint8_t* results,
                                 const uint8_t* __restrict filter = nullptr) = 0;
 
+    virtual PrimitiveType primitive_type() const = 0;
+    virtual bool supports_raw_fixed_values() const = 0;
+    virtual size_t raw_fixed_value_size() const = 0;
+    virtual Status find_batch_raw_fixed(const uint8_t* values, size_t rows, size_t value_width,
+                                        uint8_t* matches) const = 0;
+    virtual bool test_field(const Field& field) const = 0;
+
     virtual uint16_t find_fixed_len_olap_engine(const IColumn& column, const uint8_t* nullmap,
                                                 uint16_t* offsets, int number,
                                                 bool is_parse_column) = 0;
@@ -181,6 +191,56 @@ class BloomFilterFunc final : public BloomFilterFuncBase {
         OpV2::find_batch(*_bloom_filter, column, results, filter);
     }
 
+    PrimitiveType primitive_type() const override { return type; }
+
+    bool supports_raw_fixed_values() const override { return !is_string_type(type); }
+
+    size_t raw_fixed_value_size() const override {
+        if constexpr (is_string_type(type)) {
+            return 0;
+        } else {
+            return sizeof(typename PrimitiveTypeTraits<type>::CppType);
+        }
+    }
+
+    Status find_batch_raw_fixed(const uint8_t* values, size_t rows, size_t value_width,
+                                uint8_t* matches) const override {
+        if constexpr (is_string_type(type)) {
+            return Status::NotSupported("String Bloom filter cannot probe fixed-width values");
+        } else {
+            using ValueType = typename PrimitiveTypeTraits<type>::CppType;
+            if (_bloom_filter == nullptr) {
+                return Status::InternalError("Bloom filter is not initialized");
+            }
+            if (value_width != sizeof(ValueType)) {
+                return Status::Corruption("Raw Bloom filter width {} does not match expected {}",
+                                          value_width, sizeof(ValueType));
+            }
+            DORIS_CHECK(values != nullptr || rows == 0);
+            DORIS_CHECK(matches != nullptr || rows == 0);
+            for (size_t row = 0; row < rows; ++row) {
+                ValueType value;
+                std::memcpy(&value, values + row * sizeof(ValueType), sizeof(ValueType));
+                matches[row] &= _bloom_filter->test_element<fixed_len_to_uint32_v2>(value) ? 1 : 0;
+            }
+            return Status::OK();
+        }
+    }
+
+    bool test_field(const Field& field) const override {
+        DORIS_CHECK(_bloom_filter != nullptr);
+        if (field.is_null()) {
+            return _bloom_filter->contain_null();
+        }
+        if constexpr (is_string_type(type)) {
+            const auto& value = field.get<type>();
+            return _bloom_filter->test_element<fixed_len_to_uint32_v2>(
+                    StringRef(value.data(), value.size()));
+        } else {
+            return _bloom_filter->test_element<fixed_len_to_uint32_v2>(field.get<type>());
+        }
+    }
+
     template <bool is_nullable>
     uint16_t find_dict_olap_engine(const ColumnDictI32* column, const uint8_t* nullmap,
                                    uint16_t* offsets, int number) {

be/src/exprs/hybrid_set.h

@@ -20,6 +20,8 @@
 #include <gen_cpp/internal_service.pb.h>
 #include <pdqsort.h>
 
+#include <cstring>
+
 #include "common/object_pool.h"
 #include "core/column/column_nullable.h"
 #include "core/column/column_string.h"
@@ -204,6 +206,13 @@ class HybridSetBase : public FilterBase {
     // use in vectorize execute engine
     virtual bool find(const void* data, size_t) const = 0;
 
+    virtual void find_batch_raw_fixed(const uint8_t* values, size_t rows, size_t value_width,
+                                      uint8_t* matches) const {
+        for (size_t row = 0; row < rows; ++row) {
+            matches[row] &= find(values + row * value_width) ? 1 : 0;
+        }
+    }
+
     virtual void find_batch(const doris::IColumn& column, size_t rows,
                             doris::ColumnUInt8::Container& results,
                             const uint8_t* __restrict filter = nullptr) = 0;
@@ -297,6 +306,16 @@ 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 {
+        DORIS_CHECK_EQ(value_width, sizeof(ElementType));
+        for (size_t row = 0; row < rows; ++row) {
+            ElementType value;
+            std::memcpy(&value, values + row * sizeof(ElementType), sizeof(ElementType));
+            matches[row] &= _set.find(value) ? 1 : 0;
+        }
+    }
+
     void find_batch(const doris::IColumn& column, size_t rows,
                     doris::ColumnUInt8::Container& results,
                     const uint8_t* __restrict filter = nullptr) override {

be/src/exprs/runtime_filter_expr.cpp

@@ -236,6 +236,37 @@ bool RuntimeFilterExpr::can_evaluate_zonemap_filter() const {
     return _impl->can_evaluate_zonemap_filter();
 }
 
+bool RuntimeFilterExpr::can_execute_on_raw_fixed_values(const DataTypePtr& data_type,
+                                                        int column_id) const {
+    // Raw and dictionary streams omit NULL payloads and currently map NULL rows to false. A
+    // null-aware RF must therefore stay on execute_filter(), which restores its NULL semantics.
+    return !_null_aware && _impl->can_execute_on_raw_fixed_values(data_type, column_id);
+}
+
+Status RuntimeFilterExpr::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 {
+    if (!can_execute_on_raw_fixed_values(data_type, column_id)) {
+        return Status::NotSupported("Runtime filter {} cannot evaluate raw fixed-width values",
+                                    _filter_id);
+    }
+    return _impl->execute_on_raw_fixed_values(values, num_values, value_width, data_type, column_id,
+                                              matches);
+}
+
+ZoneMapFilterResult RuntimeFilterExpr::evaluate_dictionary_filter(
+        const DictionaryEvalContext& ctx) const {
+    if (!can_evaluate_dictionary_filter()) {
+        return ZoneMapFilterResult::kUnsupported;
+    }
+    return _impl->evaluate_dictionary_filter(ctx);
+}
+
+bool RuntimeFilterExpr::can_evaluate_dictionary_filter() const {
+    return !_null_aware && _impl->can_evaluate_dictionary_filter();
+}
+
 void RuntimeFilterExpr::collect_slot_column_ids(std::set<int>& column_ids) const {
     _impl->collect_slot_column_ids(column_ids);
 }

be/src/exprs/runtime_filter_expr.h

@@ -107,6 +107,13 @@ class RuntimeFilterExpr final : public VExpr {
 
     ZoneMapFilterResult evaluate_zonemap_filter(const ZoneMapEvalContext& ctx) const override;
     bool can_evaluate_zonemap_filter() const override;
+    bool can_execute_on_raw_fixed_values(const DataTypePtr& data_type,
+                                         int column_id) const override;
+    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;
+    ZoneMapFilterResult evaluate_dictionary_filter(const DictionaryEvalContext& ctx) const override;
+    bool can_evaluate_dictionary_filter() const override;
     void collect_slot_column_ids(std::set<int>& column_ids) const override;
 
     int filter_id() const { return _filter_id; }

be/src/exprs/vbloom_predicate.cpp

@@ -31,6 +31,8 @@
 #include "core/data_type/data_type_nullable.h"
 #include "core/types.h"
 #include "exprs/bloom_filter_func.h"
+#include "exprs/expr_zonemap_filter.h"
+#include "exprs/vslot_ref.h"
 #include "runtime/runtime_state.h"
 
 namespace doris {
@@ -106,6 +108,71 @@ Status VBloomPredicate::execute_runtime_filter(VExprContext* context, const Bloc
                                                ColumnPtr* arg_column) const {
     return _do_execute(context, block, filter, nullptr, count, result_column);
 }
+
+namespace {
+
+bool bloom_filter_type_matches(PrimitiveType filter_type, const DataTypePtr& data_type) {
+    if (data_type == nullptr) {
+        return false;
+    }
+    const auto value_type = remove_nullable(data_type)->get_primitive_type();
+    return filter_type == value_type || (is_string_type(filter_type) && is_string_type(value_type));
+}
+
+} // namespace
+
+bool VBloomPredicate::can_execute_on_raw_fixed_values(const DataTypePtr& data_type,
+                                                      int column_id) const {
+    if (_filter == nullptr || !_filter->supports_raw_fixed_values() || _children.size() != 1) {
+        return false;
+    }
+    const auto slot = std::dynamic_pointer_cast<VSlotRef>(_children[0]);
+    return slot != nullptr && slot->column_id() == column_id &&
+           bloom_filter_type_matches(_filter->primitive_type(), slot->data_type()) &&
+           bloom_filter_type_matches(_filter->primitive_type(), data_type);
+}
+
+Status VBloomPredicate::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 {
+    if (!can_execute_on_raw_fixed_values(data_type, column_id)) {
+        return Status::NotSupported("Bloom predicate cannot evaluate raw fixed-width values");
+    }
+    // Hash physical values inside BloomFilterFunc<T>; reconstructing an untyped hash here could
+    // disagree with the build-side hash for dates, decimals, and other fixed-width wrappers.
+    return _filter->find_batch_raw_fixed(values, num_values, value_width, matches);
+}
+
+ZoneMapFilterResult VBloomPredicate::evaluate_dictionary_filter(
+        const DictionaryEvalContext& ctx) const {
+    if (!can_evaluate_dictionary_filter()) {
+        return ZoneMapFilterResult::kUnsupported;
+    }
+    const auto slot = std::dynamic_pointer_cast<VSlotRef>(_children[0]);
+    DORIS_CHECK(slot != nullptr);
+    const auto* dictionary = ctx.slot(slot->column_id());
+    if (dictionary == nullptr ||
+        !bloom_filter_type_matches(_filter->primitive_type(), dictionary->data_type)) {
+        return ZoneMapFilterResult::kUnsupported;
+    }
+    for (const auto& value : dictionary->values) {
+        if (_filter->test_field(value)) {
+            return ZoneMapFilterResult::kMayMatch;
+        }
+    }
+    return ZoneMapFilterResult::kNoMatch;
+}
+
+bool VBloomPredicate::can_evaluate_dictionary_filter() const {
+    if (_filter == nullptr || _children.size() != 1) {
+        return false;
+    }
+    const auto slot = std::dynamic_pointer_cast<VSlotRef>(_children[0]);
+    return slot != nullptr &&
+           bloom_filter_type_matches(_filter->primitive_type(), slot->data_type());
+}
+
 const std::string& VBloomPredicate::expr_name() const {
     return EXPR_NAME;
 }
@@ -125,4 +192,4 @@ uint64_t VBloomPredicate::get_digest(uint64_t seed) const {
     return 0;
 }
 
-} // namespace doris
+} // namespace doris

be/src/exprs/vbloom_predicate.h

@@ -49,6 +49,14 @@ class VBloomPredicate final : public VExpr {
                                   const uint8_t* __restrict filter, size_t count,
                                   ColumnPtr& result_column, ColumnPtr* arg_column) const override;
 
+    bool can_execute_on_raw_fixed_values(const DataTypePtr& data_type,
+                                         int column_id) const override;
+    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;
+    ZoneMapFilterResult evaluate_dictionary_filter(const DictionaryEvalContext& ctx) const override;
+    bool can_evaluate_dictionary_filter() const override;
+
     Status prepare(RuntimeState* state, const RowDescriptor& desc, VExprContext* context) override;
     Status open(RuntimeState* state, VExprContext* context,
                 FunctionContext::FunctionStateScope scope) override;