pick 58636

Mryange · morningman · commit d9a931a68bc6 · 2025-12-26T17:59:54.000+08:00
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
@@ -1037,7 +1037,7 @@ DEFINE_mInt64(big_column_size_buffer, "65535");
 DEFINE_mInt64(small_column_size_buffer, "100");
 
 // Perform the always_true check at intervals determined by runtime_filter_sampling_frequency
-DEFINE_mInt32(runtime_filter_sampling_frequency, "64");
+DEFINE_mInt32(runtime_filter_sampling_frequency, "32");
 DEFINE_mInt32(execution_max_rpc_timeout_sec, "3600");
 DEFINE_mBool(execution_ignore_eovercrowded, "true");
 // cooldown task configs
diff --git a/be/src/olap/column_predicate.h b/be/src/olap/column_predicate.h
@@ -25,6 +25,7 @@
 #include "olap/rowset/segment_v2/bloom_filter.h"
 #include "olap/rowset/segment_v2/inverted_index_iterator.h"
 #include "runtime/define_primitive_type.h"
+#include "runtime_filter/runtime_filter_selectivity.h"
 #include "util/defer_op.h"
 #include "util/runtime_profile.h"
 #include "vec/columns/column.h"
@@ -372,8 +373,8 @@ class ColumnPredicate {
         if (!_always_true) {
             _judge_filter_rows += filter_rows;
             _judge_input_rows += input_rows;
-            vectorized::VRuntimeFilterWrapper::judge_selectivity(
-                    get_ignore_threshold(), _judge_filter_rows, _judge_input_rows, _always_true);
+            RuntimeFilterSelectivity::judge_selectivity(get_ignore_threshold(), _judge_filter_rows,
+                                                        _judge_input_rows, _always_true);
         }
     }
 
diff --git a/be/src/runtime_filter/runtime_filter_selectivity.h b/be/src/runtime_filter/runtime_filter_selectivity.h
@@ -0,0 +1,96 @@
+// 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 <cstdint>
+
+#include "common/config.h"
+#include "common/logging.h"
+
+namespace doris {
+
+// Used to track the selectivity of runtime filters
+// If the selectivity of a runtime filter is very low, it is considered ineffective and can be ignored
+// Considering that the selectivity of runtime filters may change with data variations
+// A dynamic selectivity tracking mechanism is needed
+// Note: this is not a thread-safe class
+
+class RuntimeFilterSelectivity {
+public:
+    RuntimeFilterSelectivity() = default;
+
+    RuntimeFilterSelectivity(const RuntimeFilterSelectivity&) = delete;
+    void update_judge_counter() {
+        if ((_judge_counter++) >= config::runtime_filter_sampling_frequency) {
+            reset_judge_selectivity();
+        }
+    }
+
+    void update_judge_selectivity(int filter_id, uint64_t filter_rows, uint64_t input_rows,
+                                  double ignore_thredhold) {
+        if (!_always_true) {
+            _judge_filter_rows += filter_rows;
+            _judge_input_rows += input_rows;
+            judge_selectivity(ignore_thredhold, _judge_filter_rows, _judge_input_rows,
+                              _always_true);
+        }
+
+        VLOG_ROW << fmt::format(
+                "Runtime filter[{}] selectivity update: filter_rows: {}, input_rows: {},  filter "
+                "rate: {}, "
+                "ignore_thredhold: {}, counter: {} , always_true: {}",
+                filter_id, _judge_filter_rows, _judge_input_rows,
+                static_cast<double>(_judge_filter_rows) / static_cast<double>(_judge_input_rows),
+                ignore_thredhold, _judge_counter, _always_true);
+    }
+
+    bool maybe_always_true_can_ignore() const {
+        /// TODO: maybe we can use session variable to control this behavior ?
+        if (config::runtime_filter_sampling_frequency <= 0) {
+            return false;
+        } else {
+            return _always_true;
+        }
+    }
+
+    static void judge_selectivity(double ignore_threshold, int64_t filter_rows, int64_t input_rows,
+                                  bool& always_true) {
+        // if the judged input rows is too small, we think the selectivity is not reliable
+        if (input_rows > min_judge_input_rows) {
+            always_true = (static_cast<double>(filter_rows) / static_cast<double>(input_rows)) <
+                          ignore_threshold;
+        }
+    }
+
+private:
+    void reset_judge_selectivity() {
+        _always_true = false;
+        _judge_counter = 0;
+        _judge_input_rows = 0;
+        _judge_filter_rows = 0;
+    }
+
+    int64_t _judge_input_rows = 0;
+    int64_t _judge_filter_rows = 0;
+    int _judge_counter = 0;
+    bool _always_true = false;
+
+    constexpr static int64_t min_judge_input_rows = 4096 * 10;
+};
+
+} // namespace doris
diff --git a/be/src/vec/exprs/vexpr.cpp b/be/src/vec/exprs/vexpr.cpp
@@ -1015,5 +1015,60 @@ bool VExpr::ann_dist_is_fulfilled() const {
     return _virtual_column_is_fulfilled;
 }
 
+Status VExpr::execute_filter(VExprContext* context, const Block* block,
+                             uint8_t* __restrict result_filter_data, size_t rows, bool accept_null,
+                             bool* can_filter_all) const {
+    ColumnPtr filter_column;
+    RETURN_IF_ERROR(execute_column(context, block, filter_column));
+    if (const auto* const_column = check_and_get_column<ColumnConst>(*filter_column)) {
+        // const(nullable) or const(bool)
+        const bool result = accept_null
+                                    ? (const_column->is_null_at(0) || const_column->get_bool(0))
+                                    : (!const_column->is_null_at(0) && const_column->get_bool(0));
+        if (!result) {
+            // filter all
+            *can_filter_all = true;
+            memset(result_filter_data, 0, rows);
+            return Status::OK();
+        }
+    } else if (const auto* nullable_column = check_and_get_column<ColumnNullable>(*filter_column)) {
+        // nullable(bool)
+        const ColumnPtr& nested_column = nullable_column->get_nested_column_ptr();
+        const IColumn::Filter& filter = assert_cast<const ColumnUInt8&>(*nested_column).get_data();
+        const auto* __restrict filter_data = filter.data();
+        const auto* __restrict null_map_data = nullable_column->get_null_map_data().data();
+
+        if (accept_null) {
+            for (size_t i = 0; i < rows; ++i) {
+                result_filter_data[i] &= (null_map_data[i]) || filter_data[i];
+            }
+        } else {
+            for (size_t i = 0; i < rows; ++i) {
+                result_filter_data[i] &= (!null_map_data[i]) & filter_data[i];
+            }
+        }
+
+        if ((memchr(result_filter_data, 0x1, rows) == nullptr)) {
+            *can_filter_all = true;
+            return Status::OK();
+        }
+    } else {
+        // bool
+        const IColumn::Filter& filter = assert_cast<const ColumnUInt8&>(*filter_column).get_data();
+        const auto* __restrict filter_data = filter.data();
+
+        for (size_t i = 0; i < rows; ++i) {
+            result_filter_data[i] &= filter_data[i];
+        }
+
+        if (memchr(result_filter_data, 0x1, rows) == nullptr) {
+            *can_filter_all = true;
+            return Status::OK();
+        }
+    }
+
+    return Status::OK();
+}
+
 #include "common/compile_check_end.h"
 } // namespace doris::vectorized
diff --git a/be/src/vec/exprs/vexpr.h b/be/src/vec/exprs/vexpr.h
@@ -147,6 +147,10 @@ class VExpr {
     // Therefore we need a function like this to return the actual type produced by execution.
     virtual DataTypePtr execute_type(const Block* block) const { return _data_type; }
 
+    virtual Status execute_filter(VExprContext* context, const Block* block,
+                                  uint8_t* __restrict result_filter_data, size_t rows,
+                                  bool accept_null, bool* can_filter_all) const;
+
     // `is_blockable` means this expr will be blocked in `execute` (e.g. AI Function, Remote Function)
     [[nodiscard]] virtual bool is_blockable() const {
         return std::any_of(_children.begin(), _children.end(),
@@ -204,12 +208,6 @@ class VExpr {
                                    [](VExprSPtr child) { return child->is_rf_wrapper(); });
     }
 
-    virtual void do_judge_selectivity(uint64_t filter_rows, uint64_t input_rows) {
-        for (auto child : _children) {
-            child->do_judge_selectivity(filter_rows, input_rows);
-        }
-    }
-
     static Status create_expr_tree(const TExpr& texpr, VExprContextSPtr& ctx);
 
     static Status create_expr_trees(const std::vector<TExpr>& texprs, VExprContextSPtrs& ctxs);
diff --git a/be/src/vec/exprs/vexpr_context.cpp b/be/src/vec/exprs/vexpr_context.cpp
@@ -199,7 +199,12 @@ Status VExprContext::execute_conjuncts(const VExprContextSPtrs& ctxs,
     return execute_conjuncts(ctxs, filters, false, block, result_filter, can_filter_all);
 }
 
-// TODO: Performance Optimization
+Status VExprContext::execute_filter(const Block* block, uint8_t* __restrict result_filter_data,
+                                    size_t rows, bool accept_null, bool* can_filter_all) {
+    return _root->execute_filter(this, block, result_filter_data, rows, accept_null,
+                                 can_filter_all);
+}
+
 Status VExprContext::execute_conjuncts(const VExprContextSPtrs& ctxs,
                                        const std::vector<IColumn::Filter*>* filters,
                                        bool accept_null, const Block* block,
@@ -209,85 +214,10 @@ Status VExprContext::execute_conjuncts(const VExprContextSPtrs& ctxs,
     *can_filter_all = false;
     auto* __restrict result_filter_data = result_filter->data();
     for (const auto& ctx : ctxs) {
-        // Statistics are only required when an rf wrapper exists in the expr.
-        bool is_rf_wrapper = ctx->root()->is_rf_wrapper();
-        ColumnPtr filter_column;
-        RETURN_IF_ERROR(ctx->execute(block, filter_column));
-        if (const auto* nullable_column = check_and_get_column<ColumnNullable>(*filter_column)) {
-            size_t column_size = nullable_column->size();
-            if (column_size == 0) {
-                *can_filter_all = true;
-                return Status::OK();
-            } else {
-                const ColumnPtr& nested_column = nullable_column->get_nested_column_ptr();
-                const IColumn::Filter& filter =
-                        assert_cast<const ColumnUInt8&>(*nested_column).get_data();
-                const auto* __restrict filter_data = filter.data();
-                const auto* __restrict null_map_data = nullable_column->get_null_map_data().data();
-
-                size_t input_rows =
-                        rows - (is_rf_wrapper
-                                        ? simd::count_zero_num((int8_t*)result_filter_data, rows)
-                                        : 0);
-
-                if (accept_null) {
-                    for (size_t i = 0; i < rows; ++i) {
-                        result_filter_data[i] &= (null_map_data[i]) || filter_data[i];
-                    }
-                } else {
-                    for (size_t i = 0; i < rows; ++i) {
-                        result_filter_data[i] &= (!null_map_data[i]) & filter_data[i];
-                    }
-                }
-
-                size_t output_rows =
-                        rows - (is_rf_wrapper
-                                        ? simd::count_zero_num((int8_t*)result_filter_data, rows)
-                                        : 0);
-
-                if (is_rf_wrapper) {
-                    ctx->root()->do_judge_selectivity(input_rows - output_rows, input_rows);
-                }
-
-                if ((is_rf_wrapper && output_rows == 0) ||
-                    (!is_rf_wrapper && memchr(result_filter_data, 0x1, rows) == nullptr)) {
-                    *can_filter_all = true;
-                    return Status::OK();
-                }
-            }
-        } else if (const auto* const_column = check_and_get_column<ColumnConst>(*filter_column)) {
-            // filter all
-            if (!const_column->get_bool(0)) {
-                *can_filter_all = true;
-                memset(result_filter_data, 0, result_filter->size());
-                return Status::OK();
-            }
-        } else {
-            const IColumn::Filter& filter =
-                    assert_cast<const ColumnUInt8&>(*filter_column).get_data();
-            const auto* __restrict filter_data = filter.data();
-
-            size_t input_rows =
-                    rows -
-                    (is_rf_wrapper ? simd::count_zero_num((int8_t*)result_filter_data, rows) : 0);
-
-            for (size_t i = 0; i < rows; ++i) {
-                result_filter_data[i] &= filter_data[i];
-            }
-
-            size_t output_rows =
-                    rows -
-                    (is_rf_wrapper ? simd::count_zero_num((int8_t*)result_filter_data, rows) : 0);
-
-            if (is_rf_wrapper) {
-                ctx->root()->do_judge_selectivity(input_rows - output_rows, input_rows);
-            }
-
-            if ((is_rf_wrapper && output_rows == 0) ||
-                (!is_rf_wrapper && memchr(result_filter_data, 0x1, rows) == nullptr)) {
-                *can_filter_all = true;
-                return Status::OK();
-            }
+        RETURN_IF_ERROR(
+                ctx->execute_filter(block, result_filter_data, rows, accept_null, can_filter_all));
+        if (*can_filter_all) {
+            return Status::OK();
         }
     }
     if (filters != nullptr) {
diff --git a/be/src/vec/exprs/vexpr_context.h b/be/src/vec/exprs/vexpr_context.h
@@ -33,6 +33,7 @@
 #include "olap/rowset/segment_v2/inverted_index_reader.h"
 #include "runtime/runtime_state.h"
 #include "runtime/types.h"
+#include "runtime_filter/runtime_filter_selectivity.h"
 #include "udf/udf.h"
 #include "vec/columns/column.h"
 #include "vec/core/block.h"
@@ -210,6 +211,9 @@ class VExprContext {
 
     bool all_expr_inverted_index_evaluated();
 
+    Status execute_filter(const Block* block, uint8_t* __restrict result_filter_data, size_t rows,
+                          bool accept_null, bool* can_filter_all);
+
     [[nodiscard]] static Status filter_block(VExprContext* vexpr_ctx, Block* block);
 
     [[nodiscard]] static Status filter_block(const VExprContextSPtrs& expr_contexts, Block* block,
@@ -246,6 +250,8 @@ class VExprContext {
         return _last_result_column_id;
     }
 
+    RuntimeFilterSelectivity& get_runtime_filter_selectivity() { return *_rf_selectivity; }
+
     FunctionContext::FunctionStateScope get_function_state_scope() const {
         return _is_clone ? FunctionContext::THREAD_LOCAL : FunctionContext::FRAGMENT_LOCAL;
     }
@@ -337,5 +343,8 @@ class VExprContext {
 
     segment_v2::AnnRangeSearchRuntime _ann_range_search_runtime;
     bool _suitable_for_ann_index = true;
+
+    std::unique_ptr<RuntimeFilterSelectivity> _rf_selectivity =
+            std::make_unique<RuntimeFilterSelectivity>();
 };
 } // namespace doris::vectorized
diff --git a/be/src/vec/exprs/vruntimefilter_wrapper.cpp b/be/src/vec/exprs/vruntimefilter_wrapper.cpp
diff --git a/be/src/vec/exprs/vruntimefilter_wrapper.h b/be/src/vec/exprs/vruntimefilter_wrapper.h
diff --git a/be/test/runtime_filter/runtime_filter_selectivity_test.cpp b/be/test/runtime_filter/runtime_filter_selectivity_test.cpp
