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..48acb9a9cdc750 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 @@ -39,10 +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()), - _nested(other._nested ? other._nested->clone(column_id) : nullptr) {} ~SharedPredicate() override = default; std::string debug_string() const override { std::shared_lock lock(*_mtx); @@ -56,6 +53,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 +74,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 +205,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" +}