[feature](runtime filter) Support single-column runtime filter bucket pruning (#65837)

### What problem does this PR solve?

Problem Summary: Runtime filters could prune table partitions but still
created and scheduled scanners for every hash-distributed bucket. Add FE
eligibility metadata for direct targets on a single HASH distribution
column and use exact IN values in BE to compute Doris CRC bucket
indexes. Initial and late filters skip nonmatching tablet scanners.
Composite distribution and non-invertible Bloom filters conservatively
fall back.

### Release note

Support runtime-filter bucket pruning for exact filters on single-column
HASH-distributed OLAP scans. It can be disabled with
enable_runtime_filter_bucket_prune.

Author: HappenLee

be/src/exec/operator/olap_scan_operator.cpp

@@ -130,6 +130,8 @@ Status OlapScanLocalState::_init_profile() {
     _scan_rows = ADD_COUNTER(custom_profile(), "ScanRows", TUnit::UNIT);
     _tablets_pruned_by_rf_counter =
             ADD_COUNTER(custom_profile(), "TabletsPrunedByRuntimeFilter", TUnit::UNIT);
+    _buckets_pruned_by_rf_counter =
+            ADD_COUNTER(custom_profile(), "BucketsPrunedByRuntimeFilter", TUnit::UNIT);
 
     // 1. init segment profile
     _segment_profile.reset(new RuntimeProfile("SegmentIterator"));
@@ -655,7 +657,7 @@ Status OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
         _cond_ranges.emplace_back(new doris::OlapScanRange());
     }
 
-    // Filter out tablets whose partitions have been pruned by runtime filters.
+    // Filter out tablets whose partitions or buckets have been pruned by runtime filters.
     //
     // TODO(rf-partition-prune): this happens after OlapScanLocalState::init()
     // has already executed _sync_cloud_tablets() (in cloud mode that performs
@@ -670,13 +672,16 @@ Status OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
     // the tablet, (b) acquire ready-at-start RFs before _sync_cloud_tablets()
     // and run partition pruning there to filter _scan_ranges by partition_id
     // so the heavy per-tablet work is skipped for pruned partitions.
-    if (_rf_partition_pruner.pruned_partition_count() > 0) {
+    if (_rf_partition_pruner.pruned_partition_count() > 0 ||
+        _rf_bucket_pruner.pruned_tablet_count() > 0) {
         DCHECK_EQ(_tablets.size(), _scan_ranges.size());
         DCHECK_EQ(_tablets.size(), _read_sources.size());
         size_t write_idx = 0;
         for (size_t read_idx = 0; read_idx < _tablets.size(); ++read_idx) {
             int64_t pid = _tablets[read_idx].tablet->partition_id();
-            if (!_rf_partition_pruner.is_partition_pruned(pid)) {
+            const auto& scan_range = *_scan_ranges[read_idx];
+            if (!_is_tablet_pruned_by_runtime_filter(pid, scan_range.bucket_seq,
+                                                     scan_range.bucket_num)) {
                 if (write_idx != read_idx) {
                     _tablets[write_idx] = std::move(_tablets[read_idx]);
                     _scan_ranges[write_idx] = std::move(_scan_ranges[read_idx]);
@@ -740,9 +745,9 @@ Status OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
             key_ranges.emplace_back(range.get());
         }
 
-        ParallelScannerBuilder scanner_builder(this, _tablets, _read_sources, _scanner_profile,
-                                               key_ranges, state(), p._limit, true,
-                                               p._olap_scan_node.is_preaggregation);
+        ParallelScannerBuilder scanner_builder(this, _tablets, _read_sources, _scan_ranges,
+                                               _scanner_profile, key_ranges, state(), p._limit,
+                                               true, p._olap_scan_node.is_preaggregation);
 
         int max_scanners_count = state()->parallel_scan_max_scanners_count();
 
@@ -831,6 +836,8 @@ Status OlapScanLocalState::_init_scanners(std::list<ScannerSPtr>* scanners) {
                                   p._olap_scan_node.is_preaggregation,
                                   read_row_binlog,
                                   resolve_binlog_scan_type(palo_scan_range),
+                                  palo_scan_range.bucket_seq,
+                                  palo_scan_range.bucket_num,
                                   palo_scan_range.__isset.start_tso
                                           ? std::make_optional(palo_scan_range.start_tso)
                                           : std::nullopt,
@@ -1111,13 +1118,54 @@ void OlapScanLocalState::set_scan_ranges(RuntimeState* state,
         }
     }
 
-    for (auto& scan_range : scan_ranges) {
-        DCHECK(scan_range.scan_range.__isset.palo_scan_range);
+    bool bucket_prune_metadata_initialized = !_scan_ranges.empty();
+    for (const auto& scan_range : scan_ranges) {
+        DORIS_CHECK(scan_range.scan_range.__isset.palo_scan_range);
         _scan_ranges.emplace_back(new TPaloScanRange(scan_range.scan_range.palo_scan_range));
+        const auto& palo_scan_range = *_scan_ranges.back();
+        DORIS_CHECK_EQ(palo_scan_range.__isset.bucket_seq, palo_scan_range.__isset.bucket_num);
+        if (!bucket_prune_metadata_initialized) {
+            _has_rf_bucket_prune_metadata = palo_scan_range.__isset.bucket_seq;
+            bucket_prune_metadata_initialized = true;
+        } else {
+            DORIS_CHECK_EQ(palo_scan_range.__isset.bucket_seq, _has_rf_bucket_prune_metadata);
+        }
+        if (_has_rf_bucket_prune_metadata) {
+            DORIS_CHECK_GT(palo_scan_range.bucket_num, 0);
+            DORIS_CHECK_GE(palo_scan_range.bucket_seq, 0);
+            DORIS_CHECK_LT(palo_scan_range.bucket_seq, palo_scan_range.bucket_num);
+        }
         COUNTER_UPDATE(_tablet_counter, 1);
     }
 }
 
+Status OlapScanLocalState::_on_runtime_filter_update(const VExprContextSPtrs& new_conjuncts) {
+    RETURN_IF_ERROR(Base::_on_runtime_filter_update(new_conjuncts));
+    if (!state()->query_options().enable_runtime_filter_bucket_prune ||
+        !_has_rf_bucket_prune_metadata || _scan_ranges.empty()) {
+        return Status::OK();
+    }
+
+    int64_t newly_pruned = 0;
+    RETURN_IF_ERROR(_rf_bucket_pruner.prune_by_runtime_filters(
+            _scan_ranges, new_conjuncts, _parent->runtime_filter_descs(), _parent->node_id(),
+            state()->runtime_filter_max_in_num(), &newly_pruned));
+    if (newly_pruned > 0) {
+        COUNTER_SET(_buckets_pruned_by_rf_counter, _rf_bucket_pruner.pruned_tablet_count());
+    }
+    return Status::OK();
+}
+
+bool OlapScanLocalState::_is_tablet_pruned_by_runtime_filter(int64_t partition_id,
+                                                             int32_t bucket_seq,
+                                                             int32_t bucket_num) const {
+    if (_rf_partition_pruner.is_partition_pruned(partition_id)) {
+        return true;
+    }
+    return _has_rf_bucket_prune_metadata &&
+           _rf_bucket_pruner.is_bucket_pruned(bucket_seq, bucket_num);
+}
+
 static std::string tablets_id_to_string(
         const std::vector<std::unique_ptr<TPaloScanRange>>& scan_ranges) {
     if (scan_ranges.empty()) {

be/src/exec/operator/olap_scan_operator.h

@@ -27,6 +27,7 @@
 #include "common/status.h"
 #include "exec/operator/operator.h"
 #include "exec/operator/scan_operator.h"
+#include "exec/runtime_filter/runtime_filter_bucket_pruner.h"
 #include "runtime/runtime_profile.h"
 #include "storage/index/snii/snii_prx_profile.h"
 #include "storage/olap_scan_common.h"
@@ -78,6 +79,7 @@ class OlapScanLocalState final : public ScanLocalState<OlapScanLocalState> {
                          const std::vector<TScanRangeParams>& scan_ranges) override;
     Status _init_profile() override;
     Status _process_conjuncts(RuntimeState* state) override;
+    Status _on_runtime_filter_update(const VExprContextSPtrs& new_conjuncts) override;
     bool _is_key_column(const std::string& col_name) override;
 
     bool can_push_down_column_predicate(const SlotDescriptor* slot) override;
@@ -135,7 +137,12 @@ class OlapScanLocalState final : public ScanLocalState<OlapScanLocalState> {
 
     Status _build_key_ranges_and_filters();
 
+    bool _is_tablet_pruned_by_runtime_filter(int64_t partition_id, int32_t bucket_seq,
+                                             int32_t bucket_num) const;
+
     std::vector<std::unique_ptr<TPaloScanRange>> _scan_ranges;
+    bool _has_rf_bucket_prune_metadata = false;
+    RuntimeFilterBucketPruner _rf_bucket_pruner;
     std::vector<SyncRowsetStats> _sync_statistics;
     MonotonicStopWatch _sync_cloud_tablets_watcher;
     std::shared_ptr<Dependency> _cloud_tablet_dependency;
@@ -154,6 +161,7 @@ class OlapScanLocalState final : public ScanLocalState<OlapScanLocalState> {
     snii::SniiPhraseRuntimeProfileCounters _snii_phrase_profile_counters;
 
     RuntimeProfile::Counter* _tablet_counter = nullptr;
+    RuntimeProfile::Counter* _buckets_pruned_by_rf_counter = nullptr;
     RuntimeProfile::Counter* _key_range_counter = nullptr;
     RuntimeProfile::Counter* _reader_init_timer = nullptr;
     RuntimeProfile::Counter* _scanner_init_timer = nullptr;

be/src/exec/operator/scan_operator.cpp

@@ -74,25 +74,26 @@ bool ScanLocalState<Derived>::should_run_serial() const {
 
 Status ScanLocalStateBase::update_late_arrival_runtime_filter(RuntimeState* state,
                                                               int& arrived_rf_num) {
-    // Lock needed because _conjuncts can be accessed concurrently by multiple scanner threads
+    // Lock needed because _conjuncts can be accessed concurrently by multiple scanner threads.
     LockGuard lock(_conjuncts_lock);
     size_t conjuncts_before = _conjuncts.size();
     RETURN_IF_ERROR(_helper.try_append_late_arrival_runtime_filter(
             state, _parent->operator_row_desc_before_projection(), arrived_rf_num, _conjuncts));
+    VExprContextSPtrs new_conjuncts;
+    if (_conjuncts.size() > conjuncts_before) {
+        new_conjuncts.assign(_conjuncts.begin() + conjuncts_before, _conjuncts.end());
+    }
     if (state->enable_adjust_conjunct_order_by_cost()) {
         std::ranges::stable_sort(_conjuncts, [](const auto& a, const auto& b) {
             return a->execute_cost() < b->execute_cost();
         });
-    };
-    // Only re-run partition pruning when try_append_late_arrival_runtime_filter
-    // actually appended new conjuncts. Otherwise this hook would re-scan all
-    // partition boundaries on every scheduler pass while there are still
-    // unapplied RFs (Scanner::_applied_rf_num is not advanced here), wasting
-    // CPU re-evaluating the same set of RFs against the same boundaries.
-    if (_conjuncts.size() > conjuncts_before) {
-        RETURN_IF_ERROR(_on_runtime_filter_update());
     }
-    return Status::OK();
+    if (new_conjuncts.empty()) {
+        return Status::OK();
+    }
+    // Partition projection executes the shared expression tree. Keep it serialized with
+    // clone_conjunct_ctxs(), whose VExprContext::clone() opens that same tree.
+    return _on_runtime_filter_update(new_conjuncts);
 }
 
 Status ScanLocalStateBase::clone_conjunct_ctxs(VExprContextSPtrs& scanner_conjuncts) {
@@ -105,19 +106,15 @@ Status ScanLocalStateBase::clone_conjunct_ctxs(VExprContextSPtrs& scanner_conjun
     return Status::OK();
 }
 
-bool ScanLocalStateBase::is_partition_pruned(int64_t partition_id) const {
-    return _rf_partition_pruner.is_partition_pruned(partition_id);
-}
-
-Status ScanLocalStateBase::_on_runtime_filter_update() {
+Status ScanLocalStateBase::_on_runtime_filter_update(const VExprContextSPtrs& new_conjuncts) {
     const auto* parsed = _parent->parsed_partition_boundaries();
     if (parsed != nullptr && !parsed->empty()) {
-        RETURN_IF_ERROR(_do_partition_pruning_by_rf());
+        RETURN_IF_ERROR(_do_partition_pruning_by_rf(new_conjuncts));
     }
     return Status::OK();
 }
 
-Status ScanLocalStateBase::_do_partition_pruning_by_rf() {
+Status ScanLocalStateBase::_do_partition_pruning_by_rf(const VExprContextSPtrs& conjuncts) {
     if (!_state->query_options().enable_runtime_filter_partition_prune) {
         return Status::OK();
     }
@@ -127,7 +124,7 @@ Status ScanLocalStateBase::_do_partition_pruning_by_rf() {
     }
     int64_t newly_pruned = 0;
     RETURN_IF_ERROR(_rf_partition_pruner.prune_by_runtime_filters(
-            *parsed, _conjuncts, _parent->runtime_filter_descs(), _parent->node_id(),
+            *parsed, conjuncts, _parent->runtime_filter_descs(), _parent->node_id(),
             &newly_pruned));
     if (newly_pruned > 0) {
         COUNTER_SET(_partitions_pruned_by_rf_counter,
@@ -236,7 +233,8 @@ Status ScanLocalState<Derived>::open(RuntimeState* state) {
     RETURN_IF_ERROR(_helper.acquire_runtime_filter(state, _conjuncts,
                                                    p.operator_row_desc_before_projection()));
     if (_conjuncts.size() > conjuncts_before) {
-        RETURN_IF_ERROR(_on_runtime_filter_update());
+        VExprContextSPtrs new_conjuncts(_conjuncts.begin() + conjuncts_before, _conjuncts.end());
+        RETURN_IF_ERROR(_on_runtime_filter_update(new_conjuncts));
     }
 
     // Disable condition cache in topn filter valid. TODO:: Try to support the topn filter in condition cache

be/src/exec/operator/scan_operator.h

@@ -98,10 +98,6 @@ class ScanLocalStateBase : public PipelineXLocalState<> {
     [[nodiscard]] virtual int min_scanners_concurrency(RuntimeState* state) const;
     [[nodiscard]] virtual ScannerScheduler* scan_scheduler(RuntimeState* state) const;
 
-    // Thread-safe check whether a partition has been pruned by runtime filter.
-    // Callable from any scan type's scanner in scheduling threads.
-    bool is_partition_pruned(int64_t partition_id) const;
-
     [[nodiscard]] std::string get_name() { return _parent->get_name(); }
 
     uint64_t get_condition_cache_digest() const { return _condition_cache_digest; }
@@ -116,12 +112,12 @@ class ScanLocalStateBase : public PipelineXLocalState<> {
 
     virtual Status _init_profile() = 0;
 
-    // Hook for subclasses to react after new runtime filters are appended.
-    // Called inside update_late_arrival_runtime_filter() while _conjuncts_lock is held.
-    // Default implementation runs partition pruning on the newly appended RFs.
-    virtual Status _on_runtime_filter_update();
+    // Hook for subclasses to process only the runtime-filter conjuncts appended by the current
+    // update. Late-arrival updates call this while holding _conjuncts_lock because pruning may
+    // execute expression nodes shared with scanner clones.
+    virtual Status _on_runtime_filter_update(const VExprContextSPtrs& new_conjuncts);
 
-    Status _do_partition_pruning_by_rf();
+    Status _do_partition_pruning_by_rf(const VExprContextSPtrs& conjuncts);
 
     std::atomic<bool> _opened {false};
 
@@ -296,10 +292,7 @@ class ScanLocalState : public ScanLocalStateBase {
     friend class Scanner;
 
     Status _init_profile() override;
-    virtual Status _process_conjuncts(RuntimeState* state) {
-        RETURN_IF_ERROR(_do_partition_pruning_by_rf());
-        return _normalize_conjuncts(state);
-    }
+    virtual Status _process_conjuncts(RuntimeState* state) { return _normalize_conjuncts(state); }
     virtual bool _should_push_down_common_expr(const VExprSPtr&) { return false; }
 
     virtual bool can_push_down_column_predicate(const SlotDescriptor* slot) {