[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/always-true data where applicable. Empty runtime filter fields are omitted for scans without runtime filters, 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
    - Ran local profile_level=2 scan and runtime-filter join queries on scan_filter_profile_db and checked generated profiles
- 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

@@ -55,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;
 
@@ -598,9 +642,9 @@ void OlapScanLocalState::_register_key_range_scan_filter() {
 
     ScanFilterDesc desc;
     desc.kind = ScanFilterKind::KEY_RANGE;
-    desc.compact_info = _scan_keys.debug_string();
-    desc.debug_string = desc.compact_info;
+    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));
 }
 
@@ -1217,6 +1261,14 @@ 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();
+                        DORIS_CHECK(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()) {
@@ -1237,7 +1289,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();

be/src/exec/operator/scan_operator.cpp

@@ -287,7 +287,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));
     }
@@ -408,7 +408,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()) {
@@ -537,13 +537,6 @@ Status ScanLocalState<Derived>::_normalize_predicate(VExprContext* context, cons
                 for (size_t i = predicates_before; i < slot_predicates.size(); ++i) {
                     slot_predicates[i]->attach_scan_filter(handle);
                 }
-                if (_is_key_column(slot->col_name())) {
-                    auto& source_ids = _slot_id_to_scan_filter_ids_for_key_range[slot->id()];
-                    if (std::find(source_ids.begin(), source_ids.end(), handle.filter_id) ==
-                        source_ids.end()) {
-                        source_ids.push_back(handle.filter_id);
-                    }
-                }
             }
         }
     }
@@ -1401,7 +1394,7 @@ 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->materialize(custom_profile(), state->profile_level());
     }

be/src/exec/runtime_filter/runtime_filter_consumer.cpp

@@ -23,6 +23,7 @@
 #include "exprs/vbloom_predicate.h"
 #include "exprs/vdirect_in_predicate.h"
 #include "runtime/runtime_profile.h"
+#include "runtime/scan_filter_profile.h"
 
 namespace doris {
 Status RuntimeFilterConsumer::_apply_ready_expr(std::vector<RuntimeFilterExprPtr>& push_exprs) {
@@ -258,4 +259,14 @@ void RuntimeFilterConsumer::collect_realtime_profile(RuntimeProfile* parent_oper
     c->update(_always_true_counter->value());
 }
 
+void RuntimeFilterConsumer::collect_scan_filter_profile(ScanFilterProfile* scan_filter_profile) {
+    std::unique_lock<std::recursive_mutex> l(_rmtx);
+    DCHECK(scan_filter_profile != nullptr);
+    scan_filter_profile->set_runtime_filter_profile_stats(
+            {.runtime_filter_id = _wrapper->filter_id(),
+             .wait_time_ns = _wait_timer->value(),
+             .always_true_filter_rows = _always_true_counter->value(),
+             .debug_string = debug_string()});
+}
+
 } // namespace doris

be/src/exec/runtime_filter/runtime_filter_consumer.h

@@ -28,6 +28,8 @@
 #include "runtime/runtime_profile.h"
 
 namespace doris {
+class ScanFilterProfile;
+
 // Work on ScanNode or MultiCastDataStreamSource, RuntimeFilterConsumerHelper will manage all RuntimeFilterConsumer
 // Used to create RuntimeFilterExpr to filter data
 class RuntimeFilterConsumer : public RuntimeFilter {
@@ -68,6 +70,7 @@ class RuntimeFilterConsumer : public RuntimeFilter {
 
     // Called by RuntimeFilterConsumerHelper
     void collect_realtime_profile(RuntimeProfile* parent_operator_profile);
+    void collect_scan_filter_profile(ScanFilterProfile* scan_filter_profile);
 
     static std::string to_string(const State& state) {
         switch (state) {

be/src/exec/runtime_filter/runtime_filter_consumer_helper.cpp

@@ -19,6 +19,7 @@
 
 #include "exec/runtime_filter/runtime_filter_consumer.h"
 #include "runtime/runtime_profile.h"
+#include "runtime/scan_filter_profile.h"
 
 namespace doris {
 RuntimeFilterConsumerHelper::RuntimeFilterConsumerHelper(
@@ -129,8 +130,21 @@ Status RuntimeFilterConsumerHelper::try_append_late_arrival_runtime_filter(
     return Status::OK();
 }
 
-void RuntimeFilterConsumerHelper::collect_realtime_profile(
-        RuntimeProfile* parent_operator_profile) {
+void RuntimeFilterConsumerHelper::collect_realtime_profile(RuntimeProfile* parent_operator_profile,
+                                                           ScanFilterProfile* scan_filter_profile) {
+    if (_consumers.empty()) {
+        return;
+    }
+
+    if (scan_filter_profile != nullptr) {
+        scan_filter_profile->set_runtime_filter_acquire_time(
+                _acquire_runtime_filter_timer->value());
+        for (const auto& consumer : _consumers) {
+            consumer->collect_scan_filter_profile(scan_filter_profile);
+        }
+        return;
+    }
+
     std::ignore = parent_operator_profile->add_counter("RuntimeFilterInfo", TUnit::NONE,
                                                        RuntimeProfile::ROOT_COUNTER, 1);
     RuntimeProfile::Counter* c = parent_operator_profile->add_counter(
@@ -142,4 +156,4 @@ void RuntimeFilterConsumerHelper::collect_realtime_profile(
     }
 }
 
-} // namespace doris
+} // namespace doris

be/src/exec/runtime_filter/runtime_filter_consumer_helper.h

@@ -24,6 +24,8 @@
 #include "runtime/runtime_profile.h"
 
 namespace doris {
+class ScanFilterProfile;
+
 // this class used in ScanNode or MultiCastDataStreamSource
 /**
  * init -> acquire_runtime_filter -> try_append_late_arrival_runtime_filter
@@ -49,7 +51,8 @@ class RuntimeFilterConsumerHelper {
 
     // Called by XXXLocalState::close()
     // parent_operator_profile is owned by LocalState so update it is safe at here.
-    void collect_realtime_profile(RuntimeProfile* parent_operator_profile);
+    void collect_realtime_profile(RuntimeProfile* parent_operator_profile,
+                                  ScanFilterProfile* scan_filter_profile = nullptr);
 
     size_t runtime_filter_nums() const { return _runtime_filter_descs.size(); }
 
@@ -71,4 +74,4 @@ class RuntimeFilterConsumerHelper {
     std::unique_ptr<RuntimeProfile::Counter> _acquire_runtime_filter_timer =
             std::make_unique<RuntimeProfile::Counter>(TUnit::TIME_NS, 0);
 };
-} // namespace doris
+} // namespace doris