[feature](be) Add OLAP scan filter profile counters

### What problem does this PR solve?

Issue Number: N/A

Related PR: N/A

Problem Summary: Existing OLAP scan profiles only expose aggregate filter counters, making it hard to understand the selectivity of each conjunct, runtime filter, key range, and index-related filtering step. This change adds scan-local filter ids and materializes detailed ScanFilterInfo and KeyRangeInfo profile sections. Per-filter counters now show input rows, output rows, filtered rows, stage participation, source expression or runtime filter debug text, and runtime filter wait or always-true data where applicable. Runtime filter dynamic partition pruning is exposed as a separate ScanFilterInfo child because it prunes partitions/tablets before segment-level row filtering. Empty mechanism-specific fields are omitted when they do not participate, and runtime filter ids remain distinct from scan filter ids.

### Release note

Add detailed OLAP scan filter profile counters in query profile.

### Check List (For Author)

- Test: Manual test
    - ./build.sh --be
    - git diff --cached --check
- Behavior changed: Yes. Query profile includes detailed OLAP ScanFilterInfo and KeyRangeInfo counters.
- Does this need documentation: No

Author: Copilot

be/src/exec/operator/olap_scan_operator.cpp

@@ -19,6 +19,7 @@
 
 #include <fmt/format.h>
 
+#include <algorithm>
 #include <memory>
 #include <numeric>
 #include <optional>
@@ -54,6 +55,50 @@
 
 namespace doris {
 
+namespace {
+
+constexpr int64_t MAX_PROFILE_KEY_RANGES = 32;
+
+int64_t key_range_count(const std::vector<std::unique_ptr<OlapScanRange>>& ranges) {
+    int64_t count = 0;
+    for (const auto& range : ranges) {
+        if (range->has_lower_bound) {
+            ++count;
+        }
+    }
+    return count;
+}
+
+std::string scan_keys_profile_string(const std::vector<std::unique_ptr<OlapScanRange>>& ranges) {
+    fmt::memory_buffer scan_keys_buffer;
+    int64_t range_count = 0;
+    int64_t printed = 0;
+    for (const auto& range : ranges) {
+        if (!range->has_lower_bound) {
+            continue;
+        }
+        if (printed < MAX_PROFILE_KEY_RANGES) {
+            if (printed > 0) {
+                fmt::format_to(scan_keys_buffer, "; ");
+            }
+            fmt::format_to(scan_keys_buffer, "{}{} : {}{}", range->begin_include ? "[" : "(",
+                           range->begin_scan_range.debug_string(),
+                           range->end_scan_range.debug_string(), range->end_include ? "]" : ")");
+            ++printed;
+        }
+        ++range_count;
+    }
+    if (range_count > printed) {
+        if (printed > 0) {
+            fmt::format_to(scan_keys_buffer, "; ");
+        }
+        fmt::format_to(scan_keys_buffer, "... {} more", range_count - printed);
+    }
+    return fmt::to_string(scan_keys_buffer);
+}
+
+} // namespace
+
 Status OlapScanLocalState::init(RuntimeState* state, LocalStateInfo& info) {
     const TOlapScanNode& olap_scan_node = _parent->cast<OlapScanOperatorX>()._olap_scan_node;
 
@@ -582,6 +627,27 @@ bool OlapScanLocalState::_read_mor_as_dup() {
     return p._olap_scan_node.__isset.read_mor_as_dup && p._olap_scan_node.read_mor_as_dup;
 }
 
+void OlapScanLocalState::_register_key_range_scan_filter() {
+    if (_scan_filter_profile == nullptr || _key_range_scan_filter) {
+        return;
+    }
+
+    std::vector<int32_t> source_filter_ids;
+    for (const auto& [_, filter_ids] : _slot_id_to_scan_filter_ids_for_key_range) {
+        source_filter_ids.insert(source_filter_ids.end(), filter_ids.begin(), filter_ids.end());
+    }
+    std::ranges::sort(source_filter_ids);
+    source_filter_ids.erase(std::ranges::unique(source_filter_ids).begin(),
+                            source_filter_ids.end());
+
+    ScanFilterDesc desc;
+    desc.kind = ScanFilterKind::KEY_RANGE;
+    desc.compact_info = scan_keys_profile_string(_cond_ranges);
+    desc.source_filter_ids = std::move(source_filter_ids);
+    desc.range_count = key_range_count(_cond_ranges);
+    _key_range_scan_filter = _scan_filter_profile->register_filter(std::move(desc));
+}
+
 Status OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
     if (_scan_ranges.empty()) {
         _eos = true;
@@ -605,6 +671,10 @@ Status OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
     if (_cond_ranges.empty()) {
         _cond_ranges.emplace_back(new doris::OlapScanRange());
     }
+    if (std::ranges::any_of(_cond_ranges,
+                            [](const auto& range) { return range->has_lower_bound; })) {
+        _register_key_range_scan_filter();
+    }
 
     // Filter out tablets whose partitions have been pruned by runtime filters.
     //
@@ -1191,6 +1261,16 @@ Status OlapScanLocalState::_build_key_ranges_and_filters() {
                 for (const auto& it : _slot_id_to_predicates[*key_to_erase]) {
                     if (!can_erase_predicate(*it)) {
                         new_predicates.push_back(it);
+                    } else if (_scan_filter_profile != nullptr) {
+                        const auto& handle = it->scan_filter_handle();
+                        if (handle) {
+                            auto& source_ids =
+                                    _slot_id_to_scan_filter_ids_for_key_range[*key_to_erase];
+                            if (std::find(source_ids.begin(), source_ids.end(), handle.filter_id) ==
+                                source_ids.end()) {
+                                source_ids.push_back(handle.filter_id);
+                            }
+                        }
                     }
                 }
                 if (new_predicates.empty()) {
@@ -1211,7 +1291,9 @@ Status OlapScanLocalState::_build_key_ranges_and_filters() {
     }
 
     if (state()->enable_profile()) {
-        custom_profile()->add_info_string("KeyRanges", _scan_keys.debug_string());
+        if (_scan_filter_profile == nullptr) {
+            custom_profile()->add_info_string("KeyRanges", _scan_keys.debug_string());
+        }
         custom_profile()->add_info_string("TabletIds", tablets_id_to_string(_scan_ranges));
     }
     VLOG_CRITICAL << _scan_keys.debug_string();
@@ -1280,4 +1362,11 @@ void OlapScanLocalState::_attach_partition_boundaries() {
     COUNTER_SET(_total_partitions_rf_counter, parsed->total_partitions());
 }
 
+ScanRuntimeFilterPartitionPruningStats OlapScanLocalState::_runtime_filter_partition_pruning_stats()
+        const {
+    auto stats = ScanLocalStateBase::_runtime_filter_partition_pruning_stats();
+    stats.pruned_tablets = _tablets_pruned_by_rf_counter->value();
+    return stats;
+}
+
 } // namespace doris

be/src/exec/operator/olap_scan_operator.h

@@ -127,6 +127,7 @@ class OlapScanLocalState final : public ScanLocalState<OlapScanLocalState> {
     Status _init_scanners(std::list<ScannerSPtr>* scanners) override;
 
     Status _build_key_ranges_and_filters();
+    void _register_key_range_scan_filter();
 
     std::vector<std::unique_ptr<TPaloScanRange>> _scan_ranges;
     std::vector<SyncRowsetStats> _sync_statistics;
@@ -138,6 +139,7 @@ class OlapScanLocalState final : public ScanLocalState<OlapScanLocalState> {
     std::atomic_bool _sync_tablet = false;
     std::vector<std::unique_ptr<doris::OlapScanRange>> _cond_ranges;
     OlapScanKeys _scan_keys;
+    ScanFilterHandle _key_range_scan_filter;
     // If column id in this set, indicate that we need to read data after index filtering
     std::set<int32_t> _output_column_ids;
 
@@ -337,6 +339,7 @@ class OlapScanLocalState final : public ScanLocalState<OlapScanLocalState> {
     // OlapScanOperatorX (parsed once in OperatorX::prepare()). Cheap: pointer
     // assignment plus a counter set, no parsing work.
     void _attach_partition_boundaries();
+    ScanRuntimeFilterPartitionPruningStats _runtime_filter_partition_pruning_stats() const override;
 
     RuntimeProfile::Counter* _tablets_pruned_by_rf_counter = nullptr;
 };

be/src/exec/operator/scan_operator.cpp

@@ -21,6 +21,7 @@
 #include <gen_cpp/Exprs_types.h>
 #include <gen_cpp/Metrics_types.h>
 
+#include <algorithm>
 #include <cstdint>
 #include <memory>
 
@@ -77,6 +78,14 @@ Status ScanLocalStateBase::update_late_arrival_runtime_filter(RuntimeState* stat
     size_t conjuncts_before = _conjuncts.size();
     RETURN_IF_ERROR(_helper.try_append_late_arrival_runtime_filter(state, _parent->row_descriptor(),
                                                                    arrived_rf_num, _conjuncts));
+    if (_scan_filter_profile != nullptr) {
+        for (size_t i = conjuncts_before; i < _conjuncts.size(); ++i) {
+            if (_conjuncts[i]->root() != nullptr && !_conjuncts[i]->scan_filter_handle()) {
+                _conjuncts[i]->attach_scan_filter(
+                        _register_scan_filter(_conjuncts[i]->root(), nullptr));
+            }
+        }
+    }
     if (state->enable_adjust_conjunct_order_by_cost()) {
         std::ranges::sort(_conjuncts, [](const auto& a, const auto& b) {
             return a->execute_cost() < b->execute_cost();
@@ -134,6 +143,12 @@ Status ScanLocalStateBase::_do_partition_pruning_by_rf() {
     return Status::OK();
 }
 
+ScanRuntimeFilterPartitionPruningStats ScanLocalStateBase::_runtime_filter_partition_pruning_stats()
+        const {
+    return {.total_partitions = _total_partitions_rf_counter->value(),
+            .pruned_partitions = _partitions_pruned_by_rf_counter->value()};
+}
+
 int ScanLocalStateBase::max_scanners_concurrency(RuntimeState* state) const {
     // For select * from table limit 10; should just use one thread.
     if (should_run_serial()) {
@@ -180,6 +195,9 @@ Status ScanLocalState<Derived>::init(RuntimeState* state, LocalStateInfo& info)
     SCOPED_TIMER(_init_timer);
     auto& p = _parent->cast<typename Derived::Parent>();
     _max_pushdown_conditions_per_column = p._max_pushdown_conditions_per_column;
+    if (state->enable_profile() && state->profile_level() >= 1) {
+        _scan_filter_profile = std::make_shared<ScanFilterProfile>();
+    }
     RETURN_IF_ERROR(_helper.init(state, p.is_serial_operator(), p.node_id(), p.operator_id(),
                                  _filter_dependencies, p.get_name() + "_FILTER_DEPENDENCY"));
     RETURN_IF_ERROR(_init_profile());
@@ -207,6 +225,30 @@ static std::string predicates_to_string(
     return fmt::to_string(debug_string_buffer);
 }
 
+ScanFilterHandle ScanLocalStateBase::_register_scan_filter(const VExprSPtr& root,
+                                                           const SlotDescriptor* slot) {
+    if (_scan_filter_profile == nullptr || root == nullptr) {
+        return {};
+    }
+
+    ScanFilterDesc desc;
+    desc.kind = ScanFilterKind::NORMAL;
+    if (root->is_rf_wrapper()) {
+        desc.kind = ScanFilterKind::RUNTIME_FILTER;
+        desc.runtime_filter_id = assert_cast<RuntimeFilterExpr*>(root.get())->filter_id();
+    } else if (root->is_topn_filter()) {
+        desc.kind = ScanFilterKind::TOPN_FILTER;
+    }
+    if (slot != nullptr) {
+        desc.slot_id = slot->id();
+        desc.column_name = slot->col_name();
+        desc.column_id = _parent->intermediate_row_desc().get_column_id(slot->id());
+    }
+    desc.debug_string = root->debug_string();
+    desc.compact_info = desc.debug_string;
+    return _scan_filter_profile->register_filter(std::move(desc));
+}
+
 template <typename Derived>
 Status ScanLocalState<Derived>::open(RuntimeState* state) {
     SCOPED_TIMER(exec_time_counter());
@@ -251,7 +293,7 @@ Status ScanLocalState<Derived>::open(RuntimeState* state) {
 
     RETURN_IF_ERROR(_process_conjuncts(state));
 
-    if (state->enable_profile()) {
+    if (state->enable_profile() && _scan_filter_profile == nullptr) {
         custom_profile()->add_info_string("PushDownPredicates",
                                           predicates_to_string(_slot_id_to_predicates));
     }
@@ -354,6 +396,9 @@ Status ScanLocalState<Derived>::_normalize_conjuncts(RuntimeState* state) {
             RETURN_IF_ERROR(_normalize_predicate(conjunct.get(), conjunct->root(), new_root));
             if (new_root) {
                 conjunct->set_root(new_root);
+                if (!conjunct->scan_filter_handle()) {
+                    conjunct->attach_scan_filter(_register_scan_filter(conjunct->root(), nullptr));
+                }
                 if (_should_push_down_common_expr(conjunct->root())) {
                     _common_expr_ctxs_push_down.emplace_back(conjunct);
                     it = _conjuncts.erase(it);
@@ -369,7 +414,7 @@ Status ScanLocalState<Derived>::_normalize_conjuncts(RuntimeState* state) {
         ++it;
     }
 
-    if (state->enable_profile()) {
+    if (state->enable_profile() && _scan_filter_profile == nullptr) {
         std::string message;
         for (auto& conjunct : _conjuncts) {
             if (conjunct->root()) {
@@ -424,14 +469,16 @@ Status ScanLocalState<Derived>::_normalize_predicate(VExprContext* context, cons
     }
     if (_is_predicate_acting_on_slot(expr_root->children(), &slot, &range)) {
         Status status = Status::OK();
+        auto& slot_predicates = _slot_id_to_predicates[slot->id()];
+        const size_t predicates_before = slot_predicates.size();
         std::visit(
                 [&](auto& value_range) {
                     auto expr = root->is_rf_wrapper() ? root->get_impl() : root;
                     {
                         Defer attach_defer = [&]() {
                             if (pdt != PushDownType::UNACCEPTABLE && root->is_rf_wrapper()) {
                                 auto* rf_expr = assert_cast<RuntimeFilterExpr*>(root.get());
-                                _slot_id_to_predicates[slot->id()].back()->attach_profile_counter(
+                                slot_predicates.back()->attach_profile_counter(
                                         rf_expr->filter_id(),
                                         rf_expr->predicate_filtered_rows_counter(),
                                         rf_expr->predicate_input_rows_counter(),
@@ -442,43 +489,36 @@ Status ScanLocalState<Derived>::_normalize_predicate(VExprContext* context, cons
                         switch (expr->node_type()) {
                         case TExprNodeType::IN_PRED:
                             RETURN_IF_PUSH_DOWN(
-                                    _normalize_in_predicate(context, expr, slot,
-                                                            _slot_id_to_predicates[slot->id()],
+                                    _normalize_in_predicate(context, expr, slot, slot_predicates,
                                                             value_range, &pdt),
                                     status);
                             break;
                         case TExprNodeType::BINARY_PRED:
                             RETURN_IF_PUSH_DOWN(
                                     _normalize_binary_predicate(context, expr, slot,
-                                                                _slot_id_to_predicates[slot->id()],
-                                                                value_range, &pdt),
+                                                                slot_predicates, value_range, &pdt),
                                     status);
                             break;
                         case TExprNodeType::FUNCTION_CALL:
                             if (expr->is_topn_filter()) {
-                                RETURN_IF_PUSH_DOWN(
-                                        _normalize_topn_filter(context, expr, slot,
-                                                               _slot_id_to_predicates[slot->id()],
-                                                               &pdt),
-                                        status);
+                                RETURN_IF_PUSH_DOWN(_normalize_topn_filter(context, expr, slot,
+                                                                           slot_predicates, &pdt),
+                                                    status);
                             } else {
                                 RETURN_IF_PUSH_DOWN(_normalize_is_null_predicate(
-                                                            context, expr, slot,
-                                                            _slot_id_to_predicates[slot->id()],
+                                                            context, expr, slot, slot_predicates,
                                                             value_range, &pdt),
                                                     status);
                             }
                             break;
                         case TExprNodeType::BITMAP_PRED:
-                            RETURN_IF_PUSH_DOWN(_normalize_bitmap_filter(
-                                                        context, root, slot,
-                                                        _slot_id_to_predicates[slot->id()], &pdt),
+                            RETURN_IF_PUSH_DOWN(_normalize_bitmap_filter(context, root, slot,
+                                                                         slot_predicates, &pdt),
                                                 status);
                             break;
                         case TExprNodeType::BLOOM_PRED:
-                            RETURN_IF_PUSH_DOWN(_normalize_bloom_filter(
-                                                        context, root, slot,
-                                                        _slot_id_to_predicates[slot->id()], &pdt),
+                            RETURN_IF_PUSH_DOWN(_normalize_bloom_filter(context, root, slot,
+                                                                        slot_predicates, &pdt),
                                                 status);
                             break;
                         default:
@@ -493,6 +533,18 @@ Status ScanLocalState<Derived>::_normalize_predicate(VExprContext* context, cons
                 },
                 *range);
         RETURN_IF_ERROR(status);
+        if (pdt != PushDownType::UNACCEPTABLE) {
+            auto handle = context->scan_filter_handle();
+            if (!handle) {
+                handle = _register_scan_filter(root, slot);
+                context->attach_scan_filter(handle);
+            }
+            if (handle) {
+                for (size_t i = predicates_before; i < slot_predicates.size(); ++i) {
+                    slot_predicates[i]->attach_scan_filter(handle);
+                }
+            }
+        }
     }
     if (pdt == PushDownType::ACCEPTABLE && slotref != nullptr &&
         slotref->data_type()->get_primitive_type() == PrimitiveType::TYPE_VARIANT) {
@@ -614,7 +666,12 @@ Status ScanLocalStateBase::_normalize_function_filters(VExprContext* expr_ctx, S
                                                           expr_ctx, &val, &fn_ctx, temp_pdt));
         if (temp_pdt != PushDownType::UNACCEPTABLE) {
             std::string col = slot->col_name();
-            _push_down_functions.emplace_back(opposite, col, fn_ctx, val);
+            auto handle = expr_ctx->scan_filter_handle();
+            if (!handle) {
+                handle = _register_scan_filter(expr_ctx->root(), slot);
+                expr_ctx->attach_scan_filter(handle);
+            }
+            _push_down_functions.emplace_back(opposite, col, fn_ctx, val, handle);
             *pdt = temp_pdt;
         }
     }
@@ -1343,7 +1400,12 @@ Status ScanLocalState<Derived>::close(RuntimeState* state) {
     std::list<std::shared_ptr<ScannerDelegate>> {}.swap(_scanners);
     COUNTER_SET(_wait_for_dependency_timer, _scan_dependency->watcher_elapse_time());
     COUNTER_SET(_wait_for_rf_timer, rf_time);
-    _helper.collect_realtime_profile(custom_profile());
+    _helper.collect_realtime_profile(custom_profile(), _scan_filter_profile.get());
+    if (_scan_filter_profile != nullptr) {
+        _scan_filter_profile->set_runtime_filter_partition_pruning_stats(
+                _runtime_filter_partition_pruning_stats());
+        _scan_filter_profile->materialize(custom_profile(), state->profile_level());
+    }
     return PipelineXLocalState<>::close(state);
 }