[improvement](topn) Refresh TopN runtime filter page ranges

be/src/exec/operator/olap_scan_operator.cpp

@@ -1417,6 +1417,12 @@ Status OlapScanLocalState::_build_key_ranges_and_filters() {
                 // In either case, predicates with negation semantics (effective NE / NOT_IN_LIST)
                 // are never subsumed by scan key ranges and must always be preserved.
                 auto can_erase_predicate = [is_fixed_value_range](const ColumnPredicate& pred) {
+                    // Static scan keys do not capture runtime bounds. For example, a full-range
+                    // key cannot replace a TopN predicate that later tightens to k1 <= 100.
+                    // Keep the shared predicate installed so readers can observe later updates.
+                    if (pred.is_runtime_filter()) {
+                        return false;
+                    }
                     PredicateType pt = pred.type();
                     bool opposite = pred.opposite();
 

be/src/storage/predicate/shared_predicate.h

@@ -17,6 +17,7 @@
 
 #pragma once
 
+#include <atomic>
 #include <cstdint>
 #include <memory>
 
@@ -39,10 +40,6 @@ class SharedPredicate final : public ColumnPredicate {
             : ColumnPredicate(column_id, col_name, PrimitiveType::INVALID_TYPE),
               _mtx(std::make_shared<std::shared_mutex>()) {}
     SharedPredicate(const ColumnPredicate& other) = delete;
-    SharedPredicate(const SharedPredicate& other, uint32_t column_id)
-            : ColumnPredicate(other, column_id),
-              _mtx(std::make_shared<std::shared_mutex>()),
-              _nested(other._nested ? other._nested->clone(column_id) : nullptr) {}
     ~SharedPredicate() override = default;
     std::string debug_string() const override {
         std::shared_lock<std::shared_mutex> lock(*_mtx);
@@ -56,6 +53,8 @@ class SharedPredicate final : public ColumnPredicate {
         return std::const_pointer_cast<ColumnPredicate>(shared_from_this());
     }
 
+    bool is_runtime_filter() const override { return true; }
+
     PredicateType type() const override {
         std::shared_lock<std::shared_mutex> lock(*_mtx);
         if (!_nested) {
@@ -75,6 +74,21 @@ class SharedPredicate final : public ColumnPredicate {
     void set_nested(const std::shared_ptr<ColumnPredicate>& nested) {
         std::unique_lock<std::shared_mutex> lock(*_mtx);
         _nested = nested;
+        _version.fetch_add(1, std::memory_order_release);
+    }
+
+    uint64_t version() const { return _version.load(std::memory_order_acquire); }
+
+    struct Snapshot {
+        uint64_t version;
+        std::shared_ptr<const ColumnPredicate> predicate;
+    };
+
+    // Keep the immutable comparison alive after releasing the lock. A concurrent update must
+    // not pair version 2 with version 1's bound, or hold the producer lock during index I/O.
+    Snapshot snapshot() const {
+        std::shared_lock<std::shared_mutex> lock(*_mtx);
+        return {.version = _version.load(std::memory_order_relaxed), .predicate = _nested};
     }
 
     Status evaluate(const IndexFieldNameAndTypePair& name_with_type, IndexIterator* iterator,
@@ -191,6 +205,7 @@ class SharedPredicate final : public ColumnPredicate {
 
     mutable std::shared_ptr<std::shared_mutex> _mtx;
     std::shared_ptr<ColumnPredicate> _nested;
+    std::atomic<uint64_t> _version {0};
 };
 
 } //namespace doris

be/src/storage/segment/segment_iterator.cpp

@@ -98,6 +98,7 @@
 #include "storage/olap_common.h"
 #include "storage/predicate/bloom_filter_predicate.h"
 #include "storage/predicate/column_predicate.h"
+#include "storage/predicate/shared_predicate.h"
 #include "storage/schema.h"
 #include "storage/segment/column_reader.h"
 #include "storage/segment/column_reader_cache.h"
@@ -460,6 +461,11 @@ Status SegmentIterator::_init_impl(const StorageReadOptions& opts) {
             continue;
         }
         _col_predicates.emplace_back(predicate);
+        if (auto shared = std::dynamic_pointer_cast<SharedPredicate>(predicate)) {
+            const auto version = shared->version();
+            _topn_runtime_filters.push_back(
+                    {.predicate = std::move(shared), .applied_version = version});
+        }
     }
     _init_column_states();
     _tablet_id = opts.tablet_id;
@@ -617,15 +623,8 @@ Status SegmentIterator::_lazy_init(Block* block) {
 
     RETURN_IF_ERROR(_apply_ann_topn_predicate());
 
-    // _row_bitmap is immutable after _range_iter is created. Cache its cardinality to avoid
-    // traversing all roaring containers for every batch.
-    _row_bitmap_cardinality = _row_bitmap.cardinality();
-
-    if (_opts.read_orderby_key_reverse) {
-        _range_iter.reset(new BackwardBitmapRangeIterator(_row_bitmap));
-    } else {
-        _range_iter.reset(new BitmapRangeIterator(_row_bitmap));
-    }
+    _reset_range_iterator();
+    _topn_read_position = _opts.read_orderby_key_reverse ? num_rows() : 0;
 
     // Reserve columns for _initial_block_row_max (the original max before any adaptive
     // prediction) because the predictor may increase block_row_max on subsequent batches
@@ -1282,6 +1281,75 @@ Status SegmentIterator::_apply_ann_topn_predicate() {
     return Status::OK();
 }
 
+void SegmentIterator::_reset_range_iterator() {
+    // TopN pruning can replace the bitmap between batches. Refresh the cached cardinality with
+    // the cursor so batch sizing does not use the larger range from before the bound arrived.
+    _row_bitmap_cardinality = _row_bitmap.cardinality();
+    if (_opts.read_orderby_key_reverse) {
+        _range_iter.reset(new BackwardBitmapRangeIterator(_row_bitmap));
+    } else {
+        _range_iter.reset(new BitmapRangeIterator(_row_bitmap));
+    }
+}
+
+Status SegmentIterator::_update_topn_runtime_filter_ranges() {
+    // Apply the first bound immediately and throttle later changes by raw rows read.
+    // A 40960-row delta is not enough; it must exceed the limit.
+    constexpr uint64_t update_threshold = 4096 * 10;
+    auto needs_update = [&](const TopNRuntimeFilter& filter) {
+        return filter.predicate->version() > filter.applied_version &&
+               (filter.applied_version == 0 ||
+                _topn_rows_read - filter.last_update_rows > update_threshold);
+    };
+    if (std::ranges::none_of(_topn_runtime_filters, needs_update)) {
+        return Status::OK();
+    }
+
+    SCOPED_RAW_TIMER(&_opts.stats->generate_row_ranges_by_zonemap_ns);
+    roaring::Roaring remaining = _row_bitmap;
+    // Exclude already consumed rows BEFORE intersecting page ranges. For example, after reading
+    // [0,100), a surviving page [50,150) may only add [100,150) to the rebuilt forward iterator.
+    if (_opts.read_orderby_key_reverse) {
+        remaining.removeRange(_topn_read_position, num_rows());
+    } else {
+        remaining.removeRange(0, _topn_read_position);
+    }
+    const auto before = remaining.cardinality();
+    // The last returned batch may have exhausted the cursor before the producer publishes a
+    // tighter bound. There are no unread rows to test against page metadata in that case.
+    if (before == 0) {
+        return Status::OK();
+    }
+    for (auto& filter : _topn_runtime_filters) {
+        if (!needs_update(filter)) {
+            continue;
+        }
+        const auto snapshot = filter.predicate->snapshot();
+        DCHECK(snapshot.predicate != nullptr);
+        AndBlockColumnPredicate predicate;
+        predicate.add_column_predicate(
+                SingleColumnBlockPredicate::create_unique(snapshot.predicate));
+        RowRanges ranges = RowRanges::create_single(num_rows());
+        RETURN_IF_ERROR(
+                _column_iterators[filter.predicate->column_id()]->get_row_ranges_by_zone_map(
+                        &predicate, nullptr, &ranges));
+        remaining &= RowRanges::ranges_to_roaring(ranges);
+        filter.applied_version = snapshot.version;
+        filter.last_update_rows = _topn_rows_read;
+    }
+    const auto filtered = before - remaining.cardinality();
+    if (filtered != 0) {
+        // Roaring iterators borrow bitmap containers. Destroy the old cursor before replacing
+        // its bitmap, then bind a new cursor to the remaining rows in the same read direction.
+        _range_iter.reset();
+        _row_bitmap = std::move(remaining);
+        _reset_range_iterator();
+        _opts.stats->rows_stats_filtered += filtered;
+        _opts.stats->rows_stats_rp_filtered += filtered;
+    }
+    return Status::OK();
+}
+
 Status SegmentIterator::_get_row_ranges_from_conditions(RowRanges* condition_row_ranges) {
     std::set<int32_t> cids;
     for (auto& entry : _opts.col_id_to_predicates) {
@@ -2665,6 +2733,11 @@ Status SegmentIterator::_read_columns_by_index(const std::vector<ColumnId>& read
     SCOPED_RAW_TIMER(&_opts.stats->predicate_column_read_ns);
 
     nrows_read = (uint16_t)_range_iter->read_batch_rowids(_block_rowids.data(), nrows_read_limit);
+    if (nrows_read > 0) {
+        _topn_read_position = _opts.read_orderby_key_reverse
+                                      ? _block_rowids[0]
+                                      : uint64_t(_block_rowids[nrows_read - 1]) + 1;
+    }
     bool is_continuous = (nrows_read > 1) &&
                          (_block_rowids[nrows_read - 1] - _block_rowids[0] == nrows_read - 1);
     VLOG_DEBUG << fmt::format(
@@ -3148,6 +3221,7 @@ Status SegmentIterator::_next_batch_internal(Block* block) {
     DCHECK(is_mem_reuse);
 
     RETURN_IF_ERROR(_lazy_init(block));
+    RETURN_IF_ERROR(_update_topn_runtime_filter_ranges());
 
     SCOPED_RAW_TIMER(&_opts.stats->block_load_ns);
 
@@ -3197,6 +3271,7 @@ Status SegmentIterator::_next_batch_internal(Block* block) {
 
     _opts.stats->blocks_load += 1;
     _opts.stats->raw_rows_read += _selected_size;
+    _topn_rows_read += _selected_size;
 
     if (_selected_size == 0) {
         return _process_eof(block);

be/src/storage/segment/segment_iterator.h

@@ -61,6 +61,7 @@ namespace doris {
 
 class VExpr;
 class VExprContext;
+class SharedPredicate;
 
 namespace segment_v2 {
 
@@ -161,6 +162,9 @@ class SegmentIterator : public RowwiseIterator {
             std::vector<std::shared_ptr<ColumnPredicate>>& remaining_predicates,
             bool* continue_apply);
     [[nodiscard]] Status _apply_ann_topn_predicate();
+
+    Status _update_topn_runtime_filter_ranges();
+    void _reset_range_iterator();
     [[nodiscard]] Status _apply_index_expr();
     // Publish _row_bitmap as IndexQueryContext::candidate_rows when it is
     // below the configured engage ratio; refreshed at conjunct boundaries as
@@ -380,10 +384,19 @@ class SegmentIterator : public RowwiseIterator {
     std::vector<std::unique_ptr<IndexIterator>> _index_iterators;
     // after init(), `_row_bitmap` contains all rowid to scan
     roaring::Roaring _row_bitmap;
-    // Cardinality cached after all row bitmap filters are applied in _lazy_init().
+    // Cardinality cached when the range iterator is created or rebuilt after TopN pruning.
     uint64_t _row_bitmap_cardinality = 0;
     // an iterator for `_row_bitmap` that can be used to extract row range to scan
     std::unique_ptr<BitmapRangeIterator> _range_iter;
+    struct TopNRuntimeFilter {
+        std::shared_ptr<SharedPredicate> predicate;
+        uint64_t applied_version;
+        uint64_t last_update_rows = 0;
+    };
+    std::vector<TopNRuntimeFilter> _topn_runtime_filters;
+    uint64_t _topn_rows_read = 0;
+    // Forward: first unread rowid. Reverse: exclusive upper bound of unread rowids.
+    uint64_t _topn_read_position = 0;
     // the next rowid to read
     rowid_t _cur_rowid;
     // members related to lazy materialization read