This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 32accf9301c [improvement](topn) Refresh TopN runtime filter page
ranges (#70020)
32accf9301c is described below
commit 32accf9301cb2db4f3197287f54971e2a45787dd
Author: TengJianPing <[email protected]>
AuthorDate: Fri Nov 6 09:55:52 2026 +0800
[improvement](topn) Refresh TopN runtime filter page ranges (#70020)
### What problem does this PR solve?
Issue Number: None
Problem Summary:
Segment readers keep row ranges computed before a TopN runtime bound
arrives or tightens. Scan-key normalization can also remove a shared
runtime predicate on a leading storage key. Readers then miss later
bound updates.
Publish versioned predicate snapshots and use page Zone Maps to refresh
unread row ranges between batches. Apply the first bound immediately and
throttle later refreshes by the number of raw rows read. Preserve scan
progress, bitmap holes, and read direction. Refresh the bitmap cursor,
cached cardinality, and filter counters together. Keep shared runtime
predicates during static scan-key normalization while still removing
static predicates covered by scan keys.
In the existing ascending TopN check with 1,048,576 rows and three sort
keys, ScanRows fell from 1,048,576 to 16,384 and TotalPagesNum fell from
372 to 6. These figures compare filtering disabled and enabled on the
same patched build. The query results were identical.
### Release note
TopN scans can skip unread pages when runtime bounds arrive or tighten.
Runtime predicates remain active after scan-key normalization.
### Check List (For Author)
- Test: Manual checks of commit boundaries, dependencies,
generated-result preservation, whitespace, clang-format v16, and header
hygiene. Include BE unit tests and two TopN regression suites. Builds
and tests were not rerun for this commit split.
- Behavior changed: Yes. TopN scans can read fewer rows and pages. This
commit retains inclusive runtime bounds.
- Does this need documentation: No
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
be/src/exec/operator/olap_scan_operator.cpp | 6 +
be/src/storage/predicate/shared_predicate.h | 23 +-
be/src/storage/segment/segment_iterator.cpp | 93 +++++-
be/src/storage/segment/segment_iterator.h | 15 +-
be/test/exec/operator/olap_scan_operator_test.cpp | 130 ++++++++
.../segment_iterator_topn_runtime_filter_test.cpp | 332 +++++++++++++++++++++
.../topn_runtime_predicate_key_range.out | 221 ++++++++++++++
.../topn_optimize/topn_runtime_range_update.out | 177 +++++++++++
.../topn_runtime_predicate_key_range.groovy | 46 +++
.../topn_optimize/topn_runtime_range_update.groovy | 39 +++
10 files changed, 1068 insertions(+), 14 deletions(-)
diff --git a/be/src/exec/operator/olap_scan_operator.cpp
b/be/src/exec/operator/olap_scan_operator.cpp
index 65c36930152..6197e3f4ec5 100644
--- a/be/src/exec/operator/olap_scan_operator.cpp
+++ b/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();
diff --git a/be/src/storage/predicate/shared_predicate.h
b/be/src/storage/predicate/shared_predicate.h
index 6d1a4aead08..48acb9a9cdc 100644
--- a/be/src/storage/predicate/shared_predicate.h
+++ b/be/src/storage/predicate/shared_predicate.h
@@ -17,6 +17,7 @@
#pragma once
+#include <atomic>
#include <cstdint>
#include <memory>
@@ -39,10 +40,6 @@ public:
: 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 @@ public:
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 @@ public:
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 @@ private:
mutable std::shared_ptr<std::shared_mutex> _mtx;
std::shared_ptr<ColumnPredicate> _nested;
+ std::atomic<uint64_t> _version {0};
};
} //namespace doris
diff --git a/be/src/storage/segment/segment_iterator.cpp
b/be/src/storage/segment/segment_iterator.cpp
index 1d627b129f1..cb77918e8c8 100644
--- a/be/src/storage/segment/segment_iterator.cpp
+++ b/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);
diff --git a/be/src/storage/segment/segment_iterator.h
b/be/src/storage/segment/segment_iterator.h
index 30fba1af4ba..0de8a0b6752 100644
--- a/be/src/storage/segment/segment_iterator.h
+++ b/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 @@ private:
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 @@ private:
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
diff --git a/be/test/exec/operator/olap_scan_operator_test.cpp
b/be/test/exec/operator/olap_scan_operator_test.cpp
index 068d38f982c..5e7653107e0 100644
--- a/be/test/exec/operator/olap_scan_operator_test.cpp
+++ b/be/test/exec/operator/olap_scan_operator_test.cpp
@@ -19,17 +19,25 @@
#include <gtest/gtest.h>
+#include <array>
#include <memory>
#include <optional>
+#include <tuple>
#include "common/object_pool.h"
#include "core/column/column_const.h"
+#include "core/column/column_vector.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
+#include "exec/pipeline/thrift_builder.h"
#include "exprs/vexpr_context.h"
+#include "exprs/vtopn_pred.h"
#include "gen_cpp/PlanNodes_types.h"
#include "gen_cpp/QueryCache_types.h"
+#include "runtime/query_context.h"
+#include "runtime/runtime_predicate.h"
+#include "storage/predicate/predicate_creator.h"
#include "storage/rowset/beta_rowset.h"
#include "testutil/column_helper.h"
#include "testutil/desc_tbl_builder.h"
@@ -360,4 +368,126 @@ TEST_F(OlapScanOperatorTsoPruningTest,
EmptyBootstrapDoesNotStartMinDeltaScanner
EXPECT_EQ(_local_state->_tablets_pruned_by_tso_counter->value(), 1);
}
+class OlapScanRuntimePredicateTest : public
testing::TestWithParam<std::tuple<bool, bool>> {
+protected:
+ void SetUp() override {
+ _state = std::make_shared<MockRuntimeState>();
+ DescriptorTblBuilder desc_builder(&_pool);
+ desc_builder.declare_tuple()
+ << TupleDescBuilder::SlotType
{std::make_shared<DataTypeInt32>(), "k1"};
+ _descs = desc_builder.build();
+ _slot = _descs->get_tuple_descriptor(0)->slots()[0];
+
+ TOlapScanNode scan_node;
+ scan_node.__set_tuple_id(0);
+ scan_node.__set_keyType(TKeysType::DUP_KEYS);
+ scan_node.__set_key_column_name({"k1"});
+ scan_node.__set_key_column_type({TPrimitiveType::INT});
+ TPlanNode plan_node;
+ plan_node.__set_node_id(0);
+ plan_node.__set_node_type(TPlanNodeType::OLAP_SCAN_NODE);
+ plan_node.__set_num_children(0);
+ plan_node.__set_row_tuples({0});
+ plan_node.__set_limit(-1);
+ plan_node.__set_olap_scan_node(scan_node);
+ _parent = std::make_shared<OlapScanOperatorX>(&_pool, plan_node, 0,
*_descs, 1,
+ TQueryCacheParam {});
+ _parent->_push_down_agg_type = TPushAggOp::NONE;
+ _parent->_colname_to_slot_id = {{"k1", _slot->id()}};
+ // The normalization capability check needs a physical column UID,
without opening tablets.
+ TColumn column;
+ column.__set_col_unique_id(0);
+ _parent->_olap_scan_node.__set_columns_desc({column});
+ _local_state = OlapScanLocalState::create_shared(_state.get(),
_parent.get());
+ _local_state->_custom_profile =
std::make_unique<RuntimeProfile>("RuntimePredicateTest");
+
+ TTopnFilterDesc desc;
+ desc.__set_source_node_id(10);
+ desc.__set_is_asc(std::get<0>(GetParam()));
+ desc.__set_null_first(false);
+ desc.__set_target_node_id_to_target_expr(
+ {{0, TRuntimeFilterDescBuilder::get_default_expr()}});
+ _state->get_query_ctx()->init_runtime_predicates({desc});
+ _runtime_predicate =
&_state->get_query_ctx()->get_runtime_predicate(10);
+ _runtime_predicate->set_detected_source();
+ ASSERT_TRUE(_runtime_predicate->init_target(0, {{_slot->id(), _slot}},
0).ok());
+ _shared_predicate = _runtime_predicate->get_predicate(0);
+ if (std::get<1>(GetParam())) {
+ ASSERT_TRUE(_runtime_predicate
+ ->update(Field::create_field<TYPE_INT>(
+ std::get<0>(GetParam()) ? 60 : 40))
+ .ok());
+ }
+ }
+
+ void check_retained(ColumnValueRange<TYPE_INT> range,
+ std::shared_ptr<ColumnPredicate> static_predicate =
nullptr) {
+ auto& predicates = _local_state->_slot_id_to_predicates[_slot->id()];
+ if (static_predicate) {
+ predicates.push_back(std::move(static_predicate));
+ }
+ _local_state->_slot_id_to_value_range[_slot->id()] = std::move(range);
+ VExprSPtr expr;
+
ASSERT_TRUE(VTopNPred::create_vtopn_pred(_runtime_predicate->get_texpr(0), 10,
expr).ok());
+ PushDownType push_down = PushDownType::UNACCEPTABLE;
+ ASSERT_TRUE(
+ _local_state->_normalize_topn_filter(nullptr, expr, _slot,
predicates, &push_down)
+ .ok());
+ ASSERT_EQ(push_down, PushDownType::ACCEPTABLE);
+ ASSERT_EQ(predicates.back(), _shared_predicate);
+
+ ASSERT_TRUE(_local_state->_build_key_ranges_and_filters().ok());
+ // Reaching an exact scan key must still remove its static range and
static predicate.
+ EXPECT_EQ(_local_state->_slot_id_to_value_range.count(_slot->id()), 0);
+ ASSERT_EQ(_local_state->_slot_id_to_predicates.count(_slot->id()), 1);
+ const auto& retained =
_local_state->_slot_id_to_predicates.at(_slot->id());
+ ASSERT_EQ(retained.size(), 1);
+ EXPECT_EQ(retained.front(), _shared_predicate);
+ EXPECT_TRUE(retained.front()->is_runtime_filter());
+ EXPECT_EQ(retained.front()->clone(0), _shared_predicate);
+
+ auto column = ColumnInt32::create();
+ column->get_data().assign({49, 50, 51});
+ std::array<bool, 3> matches {};
+ retained.front()->evaluate_vec(*column, matches.size(),
matches.data());
+ EXPECT_EQ(matches, (std::array<bool, 3> {true, true, true}));
+
ASSERT_TRUE(_runtime_predicate->update(Field::create_field<TYPE_INT>(50)).ok());
+ retained.front()->evaluate_vec(*column, matches.size(),
matches.data());
+ const bool ascending = std::get<0>(GetParam());
+ EXPECT_EQ(matches, (std::array<bool, 3> {ascending, true,
!ascending}));
+ }
+
+ ObjectPool _pool;
+ DescriptorTbl* _descs = nullptr;
+ SlotDescriptor* _slot = nullptr;
+ std::shared_ptr<MockRuntimeState> _state;
+ std::shared_ptr<OlapScanOperatorX> _parent;
+ std::shared_ptr<OlapScanLocalState> _local_state;
+ RuntimePredicate* _runtime_predicate = nullptr;
+ std::shared_ptr<ColumnPredicate> _shared_predicate;
+};
+
+TEST_P(OlapScanRuntimePredicateTest, WholeRangePreservesRuntimePredicate) {
+ check_retained(ColumnValueRange<TYPE_INT>("k1", false, 0, 0));
+}
+
+TEST_P(OlapScanRuntimePredicateTest, ScopeRangeRemovesOnlyStaticPredicate) {
+ ColumnValueRange<TYPE_INT> range("k1", false, 0, 0);
+ ASSERT_TRUE(range.add_range(SQLFilterOp::FILTER_LESS_OR_EQUAL, 100).ok());
+ check_retained(std::move(range),
+ create_comparison_predicate<PredicateType::LE>(
+ 0, "k1", _slot->type(),
Field::create_field<TYPE_INT>(100), false));
+}
+
+TEST_P(OlapScanRuntimePredicateTest, FixedRangeRemovesOnlyStaticPredicate) {
+ ColumnValueRange<TYPE_INT> range("k1", false, 0, 0);
+ ASSERT_TRUE(range.add_fixed_value(50).ok());
+ check_retained(std::move(range),
+ create_comparison_predicate<PredicateType::EQ>(
+ 0, "k1", _slot->type(),
Field::create_field<TYPE_INT>(50), false));
+}
+
+INSTANTIATE_TEST_SUITE_P(Bounds, OlapScanRuntimePredicateTest,
+ testing::Combine(testing::Bool(), testing::Bool()));
+
} // namespace doris
diff --git
a/be/test/storage/segment/segment_iterator_topn_runtime_filter_test.cpp
b/be/test/storage/segment/segment_iterator_topn_runtime_filter_test.cpp
new file mode 100644
index 00000000000..7ffd083b9d6
--- /dev/null
+++ b/be/test/storage/segment/segment_iterator_topn_runtime_filter_test.cpp
@@ -0,0 +1,332 @@
+// 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.
+
+#include <gtest/gtest.h>
+
+#include <algorithm>
+#include <memory>
+#include <set>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_number.h"
+#include "exec/pipeline/thrift_builder.h"
+#include "io/fs/file_writer.h"
+#include "io/fs/local_file_system.h"
+#include "runtime/descriptors.h"
+#include "runtime/runtime_predicate.h"
+#include "runtime/thread_context.h"
+#include "storage/index/zone_map/zone_map_index.h"
+#include "storage/predicate/block_column_predicate.h"
+#include "storage/row_cursor.h"
+#include "storage/segment/segment.h"
+#include "storage/segment/segment_iterator.h"
+#include "storage/segment/test_segment_writer.h"
+#include "storage/tablet/tablet_schema_helper.h"
+#include "testutil/mock/mock_runtime_state.h"
+
+namespace doris::segment_v2 {
+namespace {
+constexpr auto kTestDir = "./ut_dir/topn_runtime_ranges";
+constexpr int kRows = 131072;
+constexpr int kBound = kRows / 2 - 1;
+
+struct ScanResult {
+ std::vector<int32_t> ids;
+ int64_t raw_rows = 0;
+ int64_t pages = 0;
+ int64_t range_filtered = 0;
+};
+} // namespace
+
+class SegmentIteratorTopNRuntimeFilterTest : public testing::Test {
+protected:
+ void SetUp() override {
+ auto fs = io::global_local_filesystem();
+ ASSERT_TRUE(fs->delete_directory(kTestDir).ok());
+ ASSERT_TRUE(fs->create_directory(kTestDir).ok());
+ _tablet_schema = std::make_shared<TabletSchema>();
+ _tablet_schema->append_column(*create_int_key(0, false));
+ _tablet_schema->set_storage_page_size(4096);
+ _schema = std::make_shared<ReadSchema>(_tablet_schema->columns());
+ _slot._id = 0;
+ _slot._col_name = _tablet_schema->column(0).name();
+ _slot._type = std::make_shared<DataTypeInt32>();
+ ASSERT_NO_FATAL_FAILURE(write_segment());
+ }
+
+ void TearDown() override {
+ _reader.reset();
+ _segment.reset();
+
EXPECT_TRUE(io::global_local_filesystem()->delete_directory(kTestDir).ok());
+ }
+
+ void write_segment() {
+ const auto path = std::string(kTestDir) + "/segment_7.dat";
+ auto fs = io::global_local_filesystem();
+ io::FileWriterPtr file;
+ ASSERT_TRUE(fs->create_file(path, &file).ok());
+ VerticalSegmentWriterOptions options;
+ options.num_rows_per_block = 1024;
+ TestVerticalSegmentWriter writer(file.get(), 7, _tablet_schema,
nullptr, nullptr, options,
+ nullptr);
+ ASSERT_TRUE(writer.init().ok());
+ RowCursor row;
+ ASSERT_TRUE(row.init_scan_key(_tablet_schema,
{Field::create_field<TYPE_INT>(0)}).ok());
+ for (int rid = 0; rid < kRows; ++rid) {
+ row.mutable_field(0) = Field::create_field<TYPE_INT>(rid);
+ ASSERT_TRUE(writer.append_row(row).ok());
+ }
+ uint64_t index_size = 0;
+ uint64_t file_size = 0;
+ ASSERT_TRUE(writer.finalize_columns(&index_size).ok());
+ ASSERT_TRUE(writer.finalize_footer(&file_size).ok());
+ ASSERT_TRUE(file->close().ok());
+ ASSERT_TRUE(Segment::open(fs, path, 100, 7, RowsetId {.version = 1},
_tablet_schema,
+ io::FileReaderOptions {}, &_segment)
+ .ok());
+ }
+
+ void start(bool reverse = false, const Field& initial = Field(), bool
with_hole = false,
+ int batch_rows = 1024) {
+ _reader.reset();
+ _stats = OlapReaderStatistics {};
+ auto target = TRuntimeFilterDescBuilder::get_default_expr();
+ target.nodes[0].__set_type(create_type_desc(TYPE_INT));
+ TTopnFilterDesc desc;
+ desc.__set_source_node_id(10);
+ desc.__set_is_asc(!reverse);
+ desc.__set_null_first(false);
+ desc.__set_target_node_id_to_target_expr({{20, target}});
+ _predicate = std::make_unique<RuntimePredicate>(desc);
+ ASSERT_TRUE(_predicate->init_target(20, {{0, &_slot}}, 0).ok());
+ if (!initial.is_null()) {
+ ASSERT_TRUE(_predicate->update(initial).ok());
+ }
+ StorageReadOptions options(_stats);
+ options.runtime_state = &_state;
+ options.io_ctx.reader_type = ReaderType::READER_QUERY;
+ options.block_row_max = batch_rows;
+ options.read_orderby_key_reverse = reverse;
+ if (with_hole) {
+ options.row_ranges.add({0, 2048});
+ options.row_ranges.add({4096, kRows});
+ }
+ options.column_predicates = {_predicate->get_predicate(20)};
+ auto block_predicate = AndBlockColumnPredicate::create_shared();
+ block_predicate->add_column_predicate(
+
SingleColumnBlockPredicate::create_unique(options.column_predicates.front()));
+ options.col_id_to_predicates.emplace(0, std::move(block_predicate));
+ ASSERT_TRUE(_segment->new_iterator(_schema, options, &_reader).ok());
+ _block = _schema->create_read_block();
+ }
+
+ Status read(std::vector<int32_t>* ids) {
+ _block.clear_column_data();
+ RETURN_IF_ERROR(_reader->next_batch(&_block));
+ const auto& column = assert_cast<const
ColumnInt32&>(*_block.get_by_position(0).column);
+ ids->insert(ids->end(), column.get_data().begin(),
column.get_data().end());
+ return Status::OK();
+ }
+
+ void drain(std::vector<int32_t>* ids) {
+ while (true) {
+ const auto status = read(ids);
+ if (status.is<ErrorCode::END_OF_FILE>()) {
+ return;
+ }
+ ASSERT_TRUE(status.ok()) << status;
+ }
+ }
+
+ void late_scan(bool reverse, bool dynamic, ScanResult* result, int
batch_rows = 1024) {
+ ASSERT_NO_FATAL_FAILURE(start(reverse, Field(), false, batch_rows));
+ ASSERT_TRUE(read(&result->ids).ok());
+ ASSERT_EQ(batch_rows, result->ids.size());
+ auto* iterator = assert_cast<SegmentIterator*>(_reader.get());
+ if (!dynamic) {
+ // Preserve the old behavior as a control: current row predicates
still update.
+ iterator->_topn_runtime_filters.clear();
+ }
+ ASSERT_TRUE(_predicate->update(Field::create_field<TYPE_INT>(reverse ?
kBound + 1 : kBound))
+ .ok());
+ ASSERT_NO_FATAL_FAILURE(drain(&result->ids));
+ result->raw_rows = _stats.raw_rows_read;
+ result->pages = _stats.total_pages_num;
+ result->range_filtered = _stats.rows_stats_rp_filtered;
+ }
+
+ void expect_same_scan_result(const ScanResult& baseline, const ScanResult&
dynamic) {
+ EXPECT_EQ(dynamic.ids, baseline.ids);
+ EXPECT_EQ(kRows / 2, dynamic.ids.size());
+ EXPECT_EQ(kRows, baseline.raw_rows);
+ EXPECT_LT(dynamic.raw_rows, baseline.raw_rows);
+ EXPECT_LT(dynamic.pages, baseline.pages);
+ EXPECT_EQ(kRows - dynamic.raw_rows, dynamic.range_filtered);
+ const std::set<int32_t> unique(dynamic.ids.begin(), dynamic.ids.end());
+ EXPECT_EQ(unique.size(), dynamic.ids.size());
+ }
+
+ void check_late_scan(bool reverse) {
+ ScanResult baseline;
+ ScanResult dynamic;
+ ASSERT_NO_FATAL_FAILURE(late_scan(reverse, false, &baseline));
+ ASSERT_NO_FATAL_FAILURE(late_scan(reverse, true, &dynamic));
+ expect_same_scan_result(baseline, dynamic);
+ RecordProperty(reverse ? "reverse_raw_rows" : "forward_raw_rows",
+ std::to_string(dynamic.raw_rows));
+ RecordProperty(reverse ? "reverse_pages" : "forward_pages",
std::to_string(dynamic.pages));
+ RecordProperty("baseline_raw_rows", std::to_string(baseline.raw_rows));
+ RecordProperty("baseline_pages", std::to_string(baseline.pages));
+ }
+
+ TabletSchemaSPtr _tablet_schema;
+ ReadSchemaSPtr _schema;
+ std::shared_ptr<Segment> _segment;
+ std::unique_ptr<RowwiseIterator> _reader;
+ std::unique_ptr<RuntimePredicate> _predicate;
+ SlotDescriptor _slot;
+ OlapReaderStatistics _stats;
+ MockRuntimeState _state;
+ Block _block;
+};
+
+TEST_F(SegmentIteratorTopNRuntimeFilterTest,
LateArrivalSkipsPagesAndMatchesRowFiltering) {
+ ASSERT_NO_FATAL_FAILURE(check_late_scan(false));
+ ASSERT_NO_FATAL_FAILURE(check_late_scan(true));
+}
+
+TEST_F(SegmentIteratorTopNRuntimeFilterTest,
RebuildingRefreshesCachedBitmapCardinality) {
+ for (const bool reverse : {false, true}) {
+ SCOPED_TRACE(reverse);
+ ASSERT_NO_FATAL_FAILURE(start(reverse));
+ std::vector<int32_t> ids;
+ ASSERT_TRUE(read(&ids).ok());
+ const auto consumed = ids.size();
+ auto* iterator = assert_cast<SegmentIterator*>(_reader.get());
+ EXPECT_EQ(kRows, iterator->_row_bitmap_cardinality);
+
+ ASSERT_TRUE(_predicate->update(Field::create_field<TYPE_INT>(reverse ?
kBound + 1 : kBound))
+ .ok());
+ ASSERT_TRUE(iterator->_update_topn_runtime_filter_ranges().ok());
+ EXPECT_EQ(kRows / 2 - consumed, iterator->_row_bitmap_cardinality);
+ EXPECT_EQ(iterator->_row_bitmap.cardinality(),
iterator->_row_bitmap_cardinality);
+ ASSERT_TRUE(read(&ids).ok());
+ EXPECT_EQ(2 * consumed, ids.size());
+ }
+}
+
+TEST_F(SegmentIteratorTopNRuntimeFilterTest,
RebuildingInsideAPageKeepsOnlyItsUnreadRows) {
+ for (const bool reverse : {false, true}) {
+ ScanResult result;
+ ASSERT_NO_FATAL_FAILURE(late_scan(reverse, true, &result, 127));
+ const std::set<int32_t> unique(result.ids.begin(), result.ids.end());
+ EXPECT_EQ(kRows / 2, result.ids.size());
+ EXPECT_EQ(unique.size(), result.ids.size());
+ EXPECT_EQ(reverse ? kBound + 1 : 0, *unique.begin());
+ EXPECT_EQ(reverse ? kRows - 1 : kBound, *unique.rbegin());
+ }
+}
+
+TEST_F(SegmentIteratorTopNRuntimeFilterTest,
TighteningWaitsUntilRawRowThresholdIsExceeded) {
+ ASSERT_NO_FATAL_FAILURE(start(false,
Field::create_field<TYPE_INT>(kRows)));
+ std::vector<int32_t> ids;
+ for (int batch = 0; batch < 40; ++batch) {
+ ASSERT_TRUE(read(&ids).ok());
+ }
+ ASSERT_EQ(40960, _stats.raw_rows_read);
+
ASSERT_TRUE(_predicate->update(Field::create_field<TYPE_INT>(kBound)).ok());
+ ASSERT_TRUE(read(&ids).ok());
+ EXPECT_EQ(0, _stats.rows_stats_rp_filtered);
+ ASSERT_TRUE(read(&ids).ok());
+ EXPECT_GT(_stats.rows_stats_rp_filtered, 0);
+ ASSERT_NO_FATAL_FAILURE(drain(&ids));
+ EXPECT_EQ(kRows / 2, ids.size());
+ EXPECT_EQ(std::set<int32_t>(ids.begin(), ids.end()).size(), ids.size());
+}
+
+TEST_F(SegmentIteratorTopNRuntimeFilterTest,
UpdatesAtEofDoNotRewindOrCountConsumedRows) {
+ ScanResult result;
+ ASSERT_NO_FATAL_FAILURE(late_scan(false, true, &result));
+ const auto before = _stats.rows_stats_rp_filtered;
+ ASSERT_TRUE(_predicate->update(Field::create_field<TYPE_INT>(1)).ok());
+ auto* iterator = assert_cast<SegmentIterator*>(_reader.get());
+ ASSERT_TRUE(iterator->_update_topn_runtime_filter_ranges().ok());
+ EXPECT_EQ(before, _stats.rows_stats_rp_filtered);
+ EXPECT_EQ(kRows / 2, result.ids.size());
+}
+
+TEST_F(SegmentIteratorTopNRuntimeFilterTest,
RebuildingPreservesHolesFromInitialRestrictions) {
+ ASSERT_NO_FATAL_FAILURE(start(false, Field(), true));
+ std::vector<int32_t> ids;
+ ASSERT_TRUE(read(&ids).ok());
+ ASSERT_TRUE(_predicate->update(Field::create_field<TYPE_INT>(8191)).ok());
+ ASSERT_NO_FATAL_FAILURE(drain(&ids));
+ EXPECT_EQ(6144, ids.size());
+ EXPECT_EQ(std::set<int32_t>(ids.begin(), ids.end()).size(), ids.size());
+ EXPECT_TRUE(std::ranges::all_of(
+ ids, [](int id) { return (id < 2048 || id >= 4096) && id <= 8191;
}));
+}
+
+TEST_F(SegmentIteratorTopNRuntimeFilterTest,
SnapshotKeepsItsBoundAfterPublicationOfANewerVersion) {
+ ASSERT_NO_FATAL_FAILURE(start(false, Field::create_field<TYPE_INT>(100)));
+ auto shared =
std::dynamic_pointer_cast<SharedPredicate>(_predicate->get_predicate(20));
+ ASSERT_NE(nullptr, shared);
+ const auto snapshot = shared->snapshot();
+ ASSERT_EQ(1, snapshot.version);
+ ASSERT_TRUE(_predicate->update(Field::create_field<TYPE_INT>(100)).ok());
+ ASSERT_TRUE(_predicate->update(Field::create_field<TYPE_INT>(200)).ok());
+ EXPECT_EQ(1, shared->version());
+ ASSERT_TRUE(_predicate->update(Field::create_field<TYPE_INT>(50)).ok());
+ EXPECT_EQ(2, shared->version());
+ ZoneMap zone_map;
+ zone_map.has_not_null = true;
+ zone_map.min_value = Field::create_field<TYPE_INT>(75);
+ zone_map.max_value = zone_map.min_value;
+ EXPECT_TRUE(snapshot.predicate->evaluate_and(zone_map));
+ EXPECT_FALSE(shared->evaluate_and(zone_map));
+}
+
+TEST_F(SegmentIteratorTopNRuntimeFilterTest,
ConcurrentSnapshotsPairEachVersionWithItsBound) {
+ ASSERT_NO_FATAL_FAILURE(start(false,
Field::create_field<TYPE_INT>(kRows)));
+ auto shared =
std::dynamic_pointer_cast<SharedPredicate>(_predicate->get_predicate(20));
+ ASSERT_NE(nullptr, shared);
+ std::jthread writer([&] {
+ SCOPED_INIT_THREAD_CONTEXT();
+ for (int i = 1; i <= 1000; ++i) {
+ EXPECT_TRUE(_predicate->update(Field::create_field<TYPE_INT>(kRows
- i)).ok());
+ }
+ });
+ for (int i = 0; i < 1000; ++i) {
+ const auto snapshot = shared->snapshot();
+ const int boundary = kRows - static_cast<int>(snapshot.version - 1);
+ ZoneMap zone_map;
+ zone_map.has_not_null = true;
+ zone_map.min_value = Field::create_field<TYPE_INT>(boundary);
+ zone_map.max_value = zone_map.min_value;
+ EXPECT_TRUE(snapshot.predicate->evaluate_and(zone_map));
+ zone_map.min_value = Field::create_field<TYPE_INT>(boundary + 1);
+ zone_map.max_value = zone_map.min_value;
+ EXPECT_FALSE(snapshot.predicate->evaluate_and(zone_map));
+ }
+ writer.join();
+ EXPECT_EQ(1001, shared->version());
+}
+
+} // namespace doris::segment_v2
diff --git
a/regression-test/data/topn_optimize/topn_runtime_predicate_key_range.out
b/regression-test/data/topn_optimize/topn_runtime_predicate_key_range.out
new file mode 100644
index 00000000000..39b2dbb32f9
--- /dev/null
+++ b/regression-test/data/topn_optimize/topn_runtime_predicate_key_range.out
@@ -0,0 +1,221 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !asc_baseline --
+0 0 p081023
+0 0 p083199
+0 0 p085375
+0 0 p087551
+0 0 p089727
+0 0 p091903
+0 0 p094079
+0 0 p096255
+0 0 p098431
+0 0 p100607
+0 0 p102783
+0 0 p104959
+0 0 p107135
+0 0 p109311
+0 0 p111487
+0 0 p113663
+0 0 p115839
+0 0 p118015
+0 0 p120191
+0 0 p122367
+
+-- !desc_baseline --
+1008191 1 p000000
+1008190 14 p057345
+1008190 12 p049153
+1008190 10 p040961
+1008190 8 p032769
+1008190 6 p024577
+1008190 4 p016385
+1008190 2 p008193
+1008190 0 p000001
+1008189 16 p000002
+1008189 13 p057346
+1008189 11 p049154
+1008189 9 p040962
+1008189 7 p032770
+1008189 5 p024578
+1008189 3 p016386
+1008189 1 p008194
+1008188 15 p000003
+1008188 12 p057347
+1008188 10 p049155
+
+-- !null_baseline --
+\N 0 p000800
+\N 0 p002449
+\N 0 p004098
+\N 0 p005747
+\N 0 p007396
+\N 0 p009045
+\N 0 p010694
+\N 0 p012343
+\N 0 p013992
+\N 0 p015641
+\N 0 p017290
+\N 0 p018939
+\N 0 p020588
+\N 0 p022237
+\N 0 p023886
+\N 0 p025535
+\N 0 p027184
+\N 0 p028833
+\N 0 p030482
+\N 0 p032131
+
+-- !scope_baseline --
+0 0 p081023
+0 0 p083199
+0 0 p085375
+0 0 p087551
+0 0 p089727
+0 0 p091903
+0 0 p094079
+0 0 p096255
+0 0 p098431
+0 0 p100607
+0 0 p102783
+0 0 p104959
+0 0 p107135
+0 0 p109311
+0 0 p111487
+0 0 p113663
+0 0 p115839
+0 0 p118015
+0 0 p120191
+0 0 p122367
+
+-- !fixed_baseline --
+1 0 p081278
+1 0 p083454
+1 0 p085630
+1 0 p087806
+1 0 p089982
+1 0 p092158
+1 0 p094334
+1 0 p096510
+1 0 p098686
+1 0 p100862
+1 0 p105214
+1 0 p107390
+1 0 p109566
+1 0 p111742
+1 0 p113918
+1 0 p116094
+1 0 p118270
+1 0 p120446
+1 0 p122622
+1 0 p124798
+
+-- !asc_dynamic --
+0 0 p081023
+0 0 p083199
+0 0 p085375
+0 0 p087551
+0 0 p089727
+0 0 p091903
+0 0 p094079
+0 0 p096255
+0 0 p098431
+0 0 p100607
+0 0 p102783
+0 0 p104959
+0 0 p107135
+0 0 p109311
+0 0 p111487
+0 0 p113663
+0 0 p115839
+0 0 p118015
+0 0 p120191
+0 0 p122367
+
+-- !desc_dynamic --
+1008191 1 p000000
+1008190 14 p057345
+1008190 12 p049153
+1008190 10 p040961
+1008190 8 p032769
+1008190 6 p024577
+1008190 4 p016385
+1008190 2 p008193
+1008190 0 p000001
+1008189 16 p000002
+1008189 13 p057346
+1008189 11 p049154
+1008189 9 p040962
+1008189 7 p032770
+1008189 5 p024578
+1008189 3 p016386
+1008189 1 p008194
+1008188 15 p000003
+1008188 12 p057347
+1008188 10 p049155
+
+-- !null_dynamic --
+\N 0 p000800
+\N 0 p002449
+\N 0 p004098
+\N 0 p005747
+\N 0 p007396
+\N 0 p009045
+\N 0 p010694
+\N 0 p012343
+\N 0 p013992
+\N 0 p015641
+\N 0 p017290
+\N 0 p018939
+\N 0 p020588
+\N 0 p022237
+\N 0 p023886
+\N 0 p025535
+\N 0 p027184
+\N 0 p028833
+\N 0 p030482
+\N 0 p032131
+
+-- !scope_dynamic --
+0 0 p081023
+0 0 p083199
+0 0 p085375
+0 0 p087551
+0 0 p089727
+0 0 p091903
+0 0 p094079
+0 0 p096255
+0 0 p098431
+0 0 p100607
+0 0 p102783
+0 0 p104959
+0 0 p107135
+0 0 p109311
+0 0 p111487
+0 0 p113663
+0 0 p115839
+0 0 p118015
+0 0 p120191
+0 0 p122367
+
+-- !fixed_dynamic --
+1 0 p081278
+1 0 p083454
+1 0 p085630
+1 0 p087806
+1 0 p089982
+1 0 p092158
+1 0 p094334
+1 0 p096510
+1 0 p098686
+1 0 p100862
+1 0 p105214
+1 0 p107390
+1 0 p109566
+1 0 p111742
+1 0 p113918
+1 0 p116094
+1 0 p118270
+1 0 p120446
+1 0 p122622
+1 0 p124798
+
diff --git a/regression-test/data/topn_optimize/topn_runtime_range_update.out
b/regression-test/data/topn_optimize/topn_runtime_range_update.out
new file mode 100644
index 00000000000..881d0254cb7
--- /dev/null
+++ b/regression-test/data/topn_optimize/topn_runtime_range_update.out
@@ -0,0 +1,177 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !asc_baseline --
+8
+9
+10
+11
+12
+13
+14
+15
+16
+17
+18
+19
+20
+21
+22
+23
+24
+25
+26
+27
+
+-- !desc_baseline --
+1008191
+1008190
+1008190
+1008190
+1008190
+1008190
+1008190
+1008190
+1008190
+1008189
+1008189
+1008189
+1008189
+1008189
+1008189
+1008189
+1008189
+1008188
+1008188
+1008188
+
+-- !null_baseline --
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+
+-- !ties_baseline --
+1008191 131071
+1008190 73726
+1008190 81918
+1008190 90110
+1008190 98302
+1008190 106494
+1008190 114686
+1008190 122878
+1008190 131070
+1008189 73725
+1008189 81917
+1008189 90109
+1008189 98301
+1008189 106493
+1008189 114685
+1008189 122877
+1008189 131069
+1008188 73724
+1008188 81916
+1008188 90108
+
+-- !asc_dynamic --
+8
+9
+10
+11
+12
+13
+14
+15
+16
+17
+18
+19
+20
+21
+22
+23
+24
+25
+26
+27
+
+-- !desc_dynamic --
+1008191
+1008190
+1008190
+1008190
+1008190
+1008190
+1008190
+1008190
+1008190
+1008189
+1008189
+1008189
+1008189
+1008189
+1008189
+1008189
+1008189
+1008188
+1008188
+1008188
+
+-- !null_dynamic --
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+\N
+
+-- !ties_dynamic --
+1008191 131071
+1008190 73726
+1008190 81918
+1008190 90110
+1008190 98302
+1008190 106494
+1008190 114686
+1008190 122878
+1008190 131070
+1008189 73725
+1008189 81917
+1008189 90109
+1008189 98301
+1008189 106493
+1008189 114685
+1008189 122877
+1008189 131069
+1008188 73724
+1008188 81916
+1008188 90108
+
diff --git
a/regression-test/suites/topn_optimize/topn_runtime_predicate_key_range.groovy
b/regression-test/suites/topn_optimize/topn_runtime_predicate_key_range.groovy
new file mode 100644
index 00000000000..6fc92c2a006
--- /dev/null
+++
b/regression-test/suites/topn_optimize/topn_runtime_predicate_key_range.groovy
@@ -0,0 +1,46 @@
+// 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.
+
+suite("topn_runtime_predicate_key_range") {
+ sql "DROP TABLE IF EXISTS topn_runtime_predicate_key_range"
+ sql """CREATE TABLE topn_runtime_predicate_key_range (
+ k1 INT NULL, k2 INT NOT NULL, id BIGINT NOT NULL, payload VARCHAR(32)
NOT NULL
+ ) DUPLICATE KEY(k1,k2,id) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ('replication_num'='1')"""
+ sql """INSERT INTO topn_runtime_predicate_key_range
+ SELECT CASE WHEN number % 97 = 0 THEN NULL
+ WHEN number < 65536 THEN CAST(number % 128 AS INT)
+ ELSE CAST(1000000 + number % 8192 AS INT) END,
+ CAST(number % 17 AS INT), number,
+ CONCAT('p', LPAD(CAST(131071 - number AS STRING), 6, '0'))
+ FROM numbers('number'='131072')"""
+
+ // The third ORDER BY column is not the third storage key. Read the whole
prefix tie group.
+ sql "SET topn_filter_ratio=0"
+ qt_asc_baseline "SELECT k1,k2,payload FROM
topn_runtime_predicate_key_range ORDER BY k1 NULLS LAST,k2,payload LIMIT 20
OFFSET 7"
+ qt_desc_baseline "SELECT k1,k2,payload FROM
topn_runtime_predicate_key_range ORDER BY k1 DESC NULLS LAST,k2 DESC,payload
DESC LIMIT 20 OFFSET 7"
+ qt_null_baseline "SELECT k1,k2,payload FROM
topn_runtime_predicate_key_range ORDER BY k1 NULLS FIRST,k2,payload LIMIT 20"
+ qt_scope_baseline "SELECT k1,k2,payload FROM
topn_runtime_predicate_key_range WHERE k1 >= 0 AND k1 < 64 ORDER BY
k1,k2,payload LIMIT 20 OFFSET 7"
+ qt_fixed_baseline "SELECT k1,k2,payload FROM
topn_runtime_predicate_key_range WHERE k1 IN (1,2) ORDER BY k1,k2,payload LIMIT
20 OFFSET 7"
+
+ sql "SET topn_filter_ratio=1000000"
+ qt_asc_dynamic "SELECT k1,k2,payload FROM topn_runtime_predicate_key_range
ORDER BY k1 NULLS LAST,k2,payload LIMIT 20 OFFSET 7"
+ qt_desc_dynamic "SELECT k1,k2,payload FROM
topn_runtime_predicate_key_range ORDER BY k1 DESC NULLS LAST,k2 DESC,payload
DESC LIMIT 20 OFFSET 7"
+ qt_null_dynamic "SELECT k1,k2,payload FROM
topn_runtime_predicate_key_range ORDER BY k1 NULLS FIRST,k2,payload LIMIT 20"
+ qt_scope_dynamic "SELECT k1,k2,payload FROM
topn_runtime_predicate_key_range WHERE k1 >= 0 AND k1 < 64 ORDER BY
k1,k2,payload LIMIT 20 OFFSET 7"
+ qt_fixed_dynamic "SELECT k1,k2,payload FROM
topn_runtime_predicate_key_range WHERE k1 IN (1,2) ORDER BY k1,k2,payload LIMIT
20 OFFSET 7"
+}
diff --git
a/regression-test/suites/topn_optimize/topn_runtime_range_update.groovy
b/regression-test/suites/topn_optimize/topn_runtime_range_update.groovy
new file mode 100644
index 00000000000..0fe2f9b83c8
--- /dev/null
+++ b/regression-test/suites/topn_optimize/topn_runtime_range_update.groovy
@@ -0,0 +1,39 @@
+// 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.
+
+suite("topn_runtime_range_update") {
+ sql "DROP TABLE IF EXISTS topn_runtime_range_update"
+ sql """CREATE TABLE topn_runtime_range_update (
+ id BIGINT NOT NULL, k INT NULL
+ ) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ('replication_num'='1')"""
+ sql """INSERT INTO topn_runtime_range_update
+ SELECT number, CASE WHEN number % 97 = 0 THEN NULL
+ WHEN number < 65536 THEN CAST(number AS INT)
+ ELSE CAST(1000000 + number % 8192 AS INT) END
+ FROM numbers('number'='131072')"""
+ sql "SET topn_filter_ratio=0"
+ qt_asc_baseline "SELECT k FROM topn_runtime_range_update ORDER BY k NULLS
LAST LIMIT 20 OFFSET 7"
+ qt_desc_baseline "SELECT k FROM topn_runtime_range_update ORDER BY k DESC
NULLS LAST LIMIT 20 OFFSET 7"
+ qt_null_baseline "SELECT k FROM topn_runtime_range_update ORDER BY k NULLS
FIRST LIMIT 20"
+ qt_ties_baseline "SELECT k,id FROM topn_runtime_range_update ORDER BY k
DESC NULLS LAST,id LIMIT 20 OFFSET 7"
+ sql "SET topn_filter_ratio=1000000"
+ qt_asc_dynamic "SELECT k FROM topn_runtime_range_update ORDER BY k NULLS
LAST LIMIT 20 OFFSET 7"
+ qt_desc_dynamic "SELECT k FROM topn_runtime_range_update ORDER BY k DESC
NULLS LAST LIMIT 20 OFFSET 7"
+ qt_null_dynamic "SELECT k FROM topn_runtime_range_update ORDER BY k NULLS
FIRST LIMIT 20"
+ qt_ties_dynamic "SELECT k,id FROM topn_runtime_range_update ORDER BY k
DESC NULLS LAST,id LIMIT 20 OFFSET 7"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
