From 1b8d5fd63c7547c2964ccf45cf6ce32a5540b659 Mon Sep 17 00:00:00 2001 From: jacktengg Date: Tue, 27 Oct 2026 19:44:22 +0800 Subject: [PATCH 1/2] [improvement](be) Refresh TopN runtime filter page ranges ### What problem does this PR solve? Issue Number: None Related PR: #69776 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 --- be/src/exec/operator/olap_scan_operator.cpp | 6 + be/src/storage/predicate/shared_predicate.h | 27 +- be/src/storage/segment/segment_iterator.cpp | 93 ++++- be/src/storage/segment/segment_iterator.h | 15 +- .../exec/operator/olap_scan_operator_test.cpp | 130 +++++++ ...ment_iterator_topn_runtime_filter_test.cpp | 332 ++++++++++++++++++ .../topn_runtime_predicate_key_range.out | 221 ++++++++++++ .../topn_runtime_range_update.out | 177 ++++++++++ .../topn_runtime_predicate_key_range.groovy | 46 +++ .../topn_runtime_range_update.groovy | 39 ++ 10 files changed, 1073 insertions(+), 13 deletions(-) create mode 100644 be/test/storage/segment/segment_iterator_topn_runtime_filter_test.cpp create mode 100644 regression-test/data/topn_optimize/topn_runtime_predicate_key_range.out create mode 100644 regression-test/data/topn_optimize/topn_runtime_range_update.out create mode 100644 regression-test/suites/topn_optimize/topn_runtime_predicate_key_range.groovy create mode 100644 regression-test/suites/topn_optimize/topn_runtime_range_update.groovy diff --git a/be/src/exec/operator/olap_scan_operator.cpp b/be/src/exec/operator/olap_scan_operator.cpp index 65c36930152f56..6197e3f4ec507a 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 6d1a4aead0885c..8cc50d2de49973 100644 --- a/be/src/storage/predicate/shared_predicate.h +++ b/be/src/storage/predicate/shared_predicate.h @@ -17,6 +17,7 @@ #pragma once +#include #include #include @@ -40,9 +41,11 @@ class SharedPredicate final : public ColumnPredicate { _mtx(std::make_shared()) {} SharedPredicate(const ColumnPredicate& other) = delete; SharedPredicate(const SharedPredicate& other, uint32_t column_id) - : ColumnPredicate(other, column_id), - _mtx(std::make_shared()), - _nested(other._nested ? other._nested->clone(column_id) : nullptr) {} + : ColumnPredicate(other, column_id), _mtx(std::make_shared()) { + const auto state = other.snapshot(); + _nested = state.predicate ? state.predicate->clone(column_id) : nullptr; + _version.store(state.version, std::memory_order_relaxed); + } ~SharedPredicate() override = default; std::string debug_string() const override { std::shared_lock lock(*_mtx); @@ -56,6 +59,8 @@ class SharedPredicate final : public ColumnPredicate { return std::const_pointer_cast(shared_from_this()); } + bool is_runtime_filter() const override { return true; } + PredicateType type() const override { std::shared_lock lock(*_mtx); if (!_nested) { @@ -75,6 +80,21 @@ class SharedPredicate final : public ColumnPredicate { void set_nested(const std::shared_ptr& nested) { std::unique_lock 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 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 lock(*_mtx); + return {.version = _version.load(std::memory_order_relaxed), .predicate = _nested}; } Status evaluate(const IndexFieldNameAndTypePair& name_with_type, IndexIterator* iterator, @@ -191,6 +211,7 @@ class SharedPredicate final : public ColumnPredicate { mutable std::shared_ptr _mtx; std::shared_ptr _nested; + std::atomic _version {0}; }; } //namespace doris diff --git a/be/src/storage/segment/segment_iterator.cpp b/be/src/storage/segment/segment_iterator.cpp index 1d627b129f1e2b..cb77918e8c8057 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(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 cids; for (auto& entry : _opts.col_id_to_predicates) { @@ -2665,6 +2733,11 @@ Status SegmentIterator::_read_columns_by_index(const std::vector& 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 30fba1af4baa61..0de8a0b6752e1f 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 @@ class SegmentIterator : public RowwiseIterator { std::vector>& 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> _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 _range_iter; + struct TopNRuntimeFilter { + std::shared_ptr predicate; + uint64_t applied_version; + uint64_t last_update_rows = 0; + }; + std::vector _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 068d38f982c598..5e7653107e0462 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 +#include #include #include +#include #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> { +protected: + void SetUp() override { + _state = std::make_shared(); + DescriptorTblBuilder desc_builder(&_pool); + desc_builder.declare_tuple() + << TupleDescBuilder::SlotType {std::make_shared(), "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(&_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("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( + std::get<0>(GetParam()) ? 60 : 40)) + .ok()); + } + } + + void check_retained(ColumnValueRange range, + std::shared_ptr 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 matches {}; + retained.front()->evaluate_vec(*column, matches.size(), matches.data()); + EXPECT_EQ(matches, (std::array {true, true, true})); + ASSERT_TRUE(_runtime_predicate->update(Field::create_field(50)).ok()); + retained.front()->evaluate_vec(*column, matches.size(), matches.data()); + const bool ascending = std::get<0>(GetParam()); + EXPECT_EQ(matches, (std::array {ascending, true, !ascending})); + } + + ObjectPool _pool; + DescriptorTbl* _descs = nullptr; + SlotDescriptor* _slot = nullptr; + std::shared_ptr _state; + std::shared_ptr _parent; + std::shared_ptr _local_state; + RuntimePredicate* _runtime_predicate = nullptr; + std::shared_ptr _shared_predicate; +}; + +TEST_P(OlapScanRuntimePredicateTest, WholeRangePreservesRuntimePredicate) { + check_retained(ColumnValueRange("k1", false, 0, 0)); +} + +TEST_P(OlapScanRuntimePredicateTest, ScopeRangeRemovesOnlyStaticPredicate) { + ColumnValueRange 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( + 0, "k1", _slot->type(), Field::create_field(100), false)); +} + +TEST_P(OlapScanRuntimePredicateTest, FixedRangeRemovesOnlyStaticPredicate) { + ColumnValueRange range("k1", false, 0, 0); + ASSERT_TRUE(range.add_fixed_value(50).ok()); + check_retained(std::move(range), + create_comparison_predicate( + 0, "k1", _slot->type(), Field::create_field(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 00000000000000..7ffd083b9d60cc --- /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 + +#include +#include +#include +#include +#include +#include + +#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 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(); + _tablet_schema->append_column(*create_int_key(0, false)); + _tablet_schema->set_storage_page_size(4096); + _schema = std::make_shared(_tablet_schema->columns()); + _slot._id = 0; + _slot._col_name = _tablet_schema->column(0).name(); + _slot._type = std::make_shared(); + 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(0)}).ok()); + for (int rid = 0; rid < kRows; ++rid) { + row.mutable_field(0) = Field::create_field(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(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* ids) { + _block.clear_column_data(); + RETURN_IF_ERROR(_reader->next_batch(&_block)); + const auto& column = assert_cast(*_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* ids) { + while (true) { + const auto status = read(ids); + if (status.is()) { + 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(_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(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 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; + std::unique_ptr _reader; + std::unique_ptr _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 ids; + ASSERT_TRUE(read(&ids).ok()); + const auto consumed = ids.size(); + auto* iterator = assert_cast(_reader.get()); + EXPECT_EQ(kRows, iterator->_row_bitmap_cardinality); + + ASSERT_TRUE(_predicate->update(Field::create_field(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 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(kRows))); + std::vector 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(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(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(1)).ok()); + auto* iterator = assert_cast(_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 ids; + ASSERT_TRUE(read(&ids).ok()); + ASSERT_TRUE(_predicate->update(Field::create_field(8191)).ok()); + ASSERT_NO_FATAL_FAILURE(drain(&ids)); + EXPECT_EQ(6144, ids.size()); + EXPECT_EQ(std::set(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(100))); + auto shared = std::dynamic_pointer_cast(_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(100)).ok()); + ASSERT_TRUE(_predicate->update(Field::create_field(200)).ok()); + EXPECT_EQ(1, shared->version()); + ASSERT_TRUE(_predicate->update(Field::create_field(50)).ok()); + EXPECT_EQ(2, shared->version()); + ZoneMap zone_map; + zone_map.has_not_null = true; + zone_map.min_value = Field::create_field(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(kRows))); + auto shared = std::dynamic_pointer_cast(_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(kRows - i)).ok()); + } + }); + for (int i = 0; i < 1000; ++i) { + const auto snapshot = shared->snapshot(); + const int boundary = kRows - static_cast(snapshot.version - 1); + ZoneMap zone_map; + zone_map.has_not_null = true; + zone_map.min_value = Field::create_field(boundary); + zone_map.max_value = zone_map.min_value; + EXPECT_TRUE(snapshot.predicate->evaluate_and(zone_map)); + zone_map.min_value = Field::create_field(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 00000000000000..39b2dbb32f90d2 --- /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 00000000000000..881d0254cb72d4 --- /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 00000000000000..6fc92c2a006544 --- /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 00000000000000..0fe2f9b83c8a43 --- /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" +} From 0e20c230d359fab1102ca1491d004643f72ecfb9 Mon Sep 17 00:00:00 2001 From: jacktengg Date: Tue, 3 Nov 2026 21:37:46 +0800 Subject: [PATCH 2/2] [refactor](be) Remove unused SharedPredicate copy constructor ### What problem does this PR solve? Issue Number: None Related PR: #70020 Problem Summary: The SharedPredicate constructor taking another SharedPredicate and a column ID has no callers in production code or tests. RuntimePredicate creates instances with a column ID and name, while clone() returns the existing shared instance. Remove the unused overload without changing runtime behavior. ### Release note None ### Check List (For Author) - Test: Manual test: searched all tracked references and verified the factory construction and clone paths; clang-format 16 and check-format passed for the modified header; BE build-hygiene and git diff --check passed. No build or unit tests run for this unused-code deletion. clang-tidy was attempted but could not complete because the existing analysis environment cannot locate stddef.h. - Behavior changed: No - Does this need documentation: No --- be/src/storage/predicate/shared_predicate.h | 6 ------ 1 file changed, 6 deletions(-) diff --git a/be/src/storage/predicate/shared_predicate.h b/be/src/storage/predicate/shared_predicate.h index 8cc50d2de49973..48acb9a9cdc750 100644 --- a/be/src/storage/predicate/shared_predicate.h +++ b/be/src/storage/predicate/shared_predicate.h @@ -40,12 +40,6 @@ class SharedPredicate final : public ColumnPredicate { : ColumnPredicate(column_id, col_name, PrimitiveType::INVALID_TYPE), _mtx(std::make_shared()) {} SharedPredicate(const ColumnPredicate& other) = delete; - SharedPredicate(const SharedPredicate& other, uint32_t column_id) - : ColumnPredicate(other, column_id), _mtx(std::make_shared()) { - const auto state = other.snapshot(); - _nested = state.predicate ? state.predicate->clone(column_id) : nullptr; - _version.store(state.version, std::memory_order_relaxed); - } ~SharedPredicate() override = default; std::string debug_string() const override { std::shared_lock lock(*_mtx);