From b783446272a1eba618d43638cb7075cd7effd608 Mon Sep 17 00:00:00 2001 From: hui lai Date: Sun, 20 Sep 2026 15:28:30 +0800 Subject: [PATCH 1/7] [fix](mow) Lazily load PK indexes for picked segments --- be/src/storage/tablet/base_tablet.cpp | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/be/src/storage/tablet/base_tablet.cpp b/be/src/storage/tablet/base_tablet.cpp index c7ad7a6c639e93..e40c815452a07f 100644 --- a/be/src/storage/tablet/base_tablet.cpp +++ b/be/src/storage/tablet/base_tablet.cpp @@ -530,8 +530,11 @@ Status BaseTablet::lookup_row_key(const Slice& encoded_key, TabletSchema* latest if (UNLIKELY(segment_caches[i] == nullptr)) { segment_caches[i] = std::make_unique(); + // Keep segment handles for reuse, but load PK indexes and bloom filters only + // when lookup_row_key visits a picked segment. Eagerly loading the whole rowset + // can retain large PK index pages even for segments excluded by key bounds. RETURN_IF_ERROR(SegmentLoader::instance()->load_segments( - std::static_pointer_cast(rs), segment_caches[i].get(), true, true, + std::static_pointer_cast(rs), segment_caches[i].get(), true, false, stats, io_ctx)); } auto& segments = segment_caches[i]->get_segments(); From 50f630b52d456423b5cb714004bb283ba2398879 Mon Sep 17 00:00:00 2001 From: laihui <1353307710@qq.com> Date: Sun, 20 Sep 2026 23:25:57 +0800 Subject: [PATCH 2/7] [fix](mow) Check bloom filter before loading PK index --- be/src/storage/segment/segment.cpp | 22 +- be/src/storage/segment/segment.h | 4 +- .../segment_primary_key_lookup_test.cpp | 238 ++++++++++++++++++ 3 files changed, 256 insertions(+), 8 deletions(-) create mode 100644 be/test/storage/segment/segment_primary_key_lookup_test.cpp diff --git a/be/src/storage/segment/segment.cpp b/be/src/storage/segment/segment.cpp index 8978d7e01eac03..43f0e4eaf00518 100644 --- a/be/src/storage/segment/segment.cpp +++ b/be/src/storage/segment/segment.cpp @@ -383,6 +383,11 @@ Status Segment::_open(OlapReaderStatistics* stats, const io::IOContext* source_i footer_pb_shared->has_primary_key_index_meta() ? new PrimaryKeyIndexMetaPB(footer_pb_shared->primary_key_index_meta()) : nullptr); + if (_tablet_schema->keys_type() == UNIQUE_KEYS && _pk_index_meta != nullptr) { + // Create the shared reader before publishing the segment. Its index and bloom filter + // can then be initialized independently without replacing each other's state. + _pk_index_reader = std::make_unique(); + } // delete_bitmap_calculator_test.cpp // DCHECK(footer.has_short_key_index_page()); _sk_index_page = footer_pb_shared->short_key_index_page(); @@ -780,7 +785,7 @@ Status Segment::load_pk_index_and_bf(OlapReaderStatistics* index_load_stats, Status Segment::load_index(OlapReaderStatistics* stats, const io::IOContext* source_io_ctx) { return _load_index_once.call([this, stats, source_io_ctx] { if (_tablet_schema->keys_type() == UNIQUE_KEYS && _pk_index_meta != nullptr) { - _pk_index_reader = std::make_unique(); + DCHECK(_pk_index_reader != nullptr); RETURN_IF_ERROR(_pk_index_reader->parse_index(_file_reader, *_pk_index_meta, stats, source_io_ctx)); // _meta_mem_usage += _pk_index_reader->get_memory_size(); @@ -1155,7 +1160,6 @@ Status Segment::lookup_row_key(const Slice& key, const TabletSchema* latest_sche bool with_seq_col, bool with_rowid, RowLocation* row_location, OlapReaderStatistics* stats, std::string* encoded_seq_value, const io::IOContext* io_ctx) { - RETURN_IF_ERROR(load_pk_index_and_bf(stats, io_ctx)); bool has_seq_col = latest_schema->has_sequence_col(); bool has_rowid = !latest_schema->cluster_key_uids().empty(); size_t seq_col_length = 0; @@ -1168,10 +1172,16 @@ Status Segment::lookup_row_key(const Slice& key, const TabletSchema* latest_sche Slice(key.get_data(), key.get_size() - (with_seq_col ? seq_col_length : 0) - (with_rowid ? rowid_length : 0)); - DCHECK(_pk_index_reader != nullptr); - if (!_pk_index_reader->check_present(key_without_seq)) { - return Status::Error(""); - } + // A bloom-filter miss must not load the potentially large PK index root pages. + // Preserve the exception boundary of load_pk_index_and_bf: DorisCallOnce can rethrow + // an initialization exception on a later call as well as on the first call. + RETURN_IF_CATCH_EXCEPTION({ + RETURN_IF_ERROR(_load_pk_bloom_filter(stats, io_ctx)); + if (!_pk_index_reader->check_present(key_without_seq)) { + return Status::Error(""); + } + RETURN_IF_ERROR(load_index(stats, io_ctx)); + }); bool exact_match = false; std::unique_ptr index_iterator; RETURN_IF_ERROR(_pk_index_reader->new_iterator(&index_iterator, stats, io_ctx)); diff --git a/be/src/storage/segment/segment.h b/be/src/storage/segment/segment.h index a39114187c2021..9ed9cb77fbb874 100644 --- a/be/src/storage/segment/segment.h +++ b/be/src/storage/segment/segment.h @@ -309,7 +309,7 @@ class Segment : public std::enable_shared_from_this, public MetadataAdd // map column unique id ---> it's inner data type std::map> _file_column_types; - // used to guarantee that short key index will be loaded at most once in a thread-safe way + // used to guarantee that the short key or primary key index is loaded at most once DorisCallOnce _load_index_once; // used to guarantee that primary key bloom filter will be loaded at most once in a thread-safe way DorisCallOnce _load_pk_bf_once; @@ -326,7 +326,7 @@ class Segment : public std::enable_shared_from_this, public MetadataAdd // short key index decoder // all content is in memory std::unique_ptr _sk_index_decoder; - // primary key index reader + // Created in _open before the segment is shared; PK index and BF are loaded independently. std::unique_ptr _pk_index_reader; std::mutex _open_lock; // inverted index file reader diff --git a/be/test/storage/segment/segment_primary_key_lookup_test.cpp b/be/test/storage/segment/segment_primary_key_lookup_test.cpp new file mode 100644 index 00000000000000..52354e06ee862a --- /dev/null +++ b/be/test/storage/segment/segment_primary_key_lookup_test.cpp @@ -0,0 +1,238 @@ +// 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 "core/field.h" +#include "io/fs/local_file_system.h" +#include "runtime/exec_env.h" +#include "storage/index/primary_key_index.h" +#include "storage/segment/segment.h" +#include "storage/segment/segment_writer.h" +#include "storage/storage_engine.h" +#include "storage/tablet/tablet_schema_helper.h" + +namespace doris { + +TabletSchemaSPtr create_schema(const std::vector& columns, KeysType keys_type); +using Generator = std::function; +void build_segment(SegmentWriterOptions opts, TabletSchemaSPtr build_schema, size_t segment_id, + TabletSchemaSPtr query_schema, size_t nrows, Generator generator, + std::shared_ptr* res, std::string segment_dir); + +class SegmentPrimaryKeyLookupTest : public testing::Test { +protected: + void SetUp() override { + auto fs = io::global_local_filesystem(); + ASSERT_TRUE(fs->delete_directory(_dir).ok()); + ASSERT_TRUE(fs->create_directory(_dir).ok()); + ExecEnv::GetInstance()->set_storage_engine( + std::make_unique(EngineOptions {})); + _schema = create_schema( + {create_varchar_key(0, false), + create_int_value(1, FieldAggregationMethod::OLAP_FIELD_AGGREGATION_REPLACE, + false)}, + UNIQUE_KEYS); + SegmentWriterOptions opts; + opts.enable_unique_key_merge_on_write = true; + auto generator = [](size_t rid, int cid, Field& field) { + if (cid == 0) { + // Each key exceeds the normal PK data-page target, producing a large value index. + field = Field::create_field(std::string(50000, 'a' + rid)); + } else { + field = Field::create_field(static_cast(rid)); + } + }; + build_segment(opts, _schema, 0, _schema, 8, generator, &_segment, _dir); + ASSERT_NE(_segment, nullptr); + ASSERT_NE(_segment->_pk_index_reader, nullptr); + ASSERT_FALSE(_segment->_load_index_once.has_called()); + ASSERT_FALSE(_segment->_load_pk_bf_once.has_called()); + ASSERT_FALSE(_segment->_pk_index_meta->primary_key_index() + .value_index_meta() + .is_root_data_page()); + _present_key = _segment->min_key(); + + // Find a definite BF miss without initializing the segment's own reader. Do not assume + // that an arbitrary absent key is rejected, because bloom filters allow false positives. + PrimaryKeyIndexReader probe; + ASSERT_TRUE( + probe.parse_bf(_segment->file_reader(), *_segment->_pk_index_meta, nullptr).ok()); + for (int i = 0; i < 10000; ++i) { + auto candidate = _present_key + std::to_string(i); + if (!probe.check_present(Slice(candidate))) { + _missing_key = std::move(candidate); + break; + } + } + ASSERT_FALSE(_missing_key.empty()); + } + + void TearDown() override { + _segment.reset(); + ExecEnv::GetInstance()->set_storage_engine(nullptr); + EXPECT_TRUE(io::global_local_filesystem()->delete_directory(_dir).ok()); + } + + Status lookup(const std::string& key, RowLocation* location) { + return _segment->lookup_row_key(Slice(key), _schema.get(), false, false, location, nullptr); + } + + const std::string _dir = "./ut_dir/segment_primary_key_lookup_test"; + TabletSchemaSPtr _schema; + std::shared_ptr _segment; + std::string _present_key; + std::string _missing_key; +}; + +TEST_F(SegmentPrimaryKeyLookupTest, BloomFilterMissDoesNotLoadIndex) { + RowLocation location; + for (int i = 0; i < 3; ++i) { + auto st = lookup(_missing_key, &location); + EXPECT_TRUE(st.is()) << st; + EXPECT_TRUE(_segment->_load_pk_bf_once.has_called()); + EXPECT_FALSE(_segment->_load_index_once.has_called()); + EXPECT_EQ(_segment->_pk_index_reader->_index_reader, nullptr); + EXPECT_TRUE(_segment->healthy_status().ok()); + } +} + +TEST_F(SegmentPrimaryKeyLookupTest, HitAfterMissPreservesBloomFilter) { + RowLocation location; + ASSERT_TRUE(lookup(_missing_key, &location).is()); + auto* reader = _segment->_pk_index_reader.get(); + auto* bf = reader->_bf.get(); + ASSERT_FALSE(_segment->_load_index_once.has_called()); + + ASSERT_TRUE(lookup(_present_key, &location).ok()); + EXPECT_EQ(location.segment_id, _segment->id()); + EXPECT_EQ(location.row_id, 0); + EXPECT_TRUE(_segment->_load_index_once.has_called()); + EXPECT_EQ(_segment->_pk_index_reader.get(), reader); + EXPECT_EQ(reader->_bf.get(), bf); + auto* index = reader->_index_reader.get(); + ASSERT_NE(index, nullptr); + + ASSERT_TRUE(lookup(_present_key, &location).ok()); + EXPECT_EQ(reader->_index_reader.get(), index); + EXPECT_EQ(reader->_bf.get(), bf); + EXPECT_TRUE(lookup(_missing_key, &location).is()); +} + +TEST_F(SegmentPrimaryKeyLookupTest, BloomFilterPositiveStillChecksExactKey) { + ASSERT_TRUE(_segment->_load_pk_bloom_filter(nullptr).ok()); + // Deliberately make the BF positive for an absent key to exercise the false-positive path. + _segment->_pk_index_reader->_bf->add_bytes(_missing_key.data(), _missing_key.size()); + ASSERT_TRUE(_segment->_pk_index_reader->check_present(Slice(_missing_key))); + RowLocation location; + auto st = lookup(_missing_key, &location); + EXPECT_TRUE(st.is()) << st; + EXPECT_TRUE(_segment->_load_index_once.has_called()); +} + +TEST_F(SegmentPrimaryKeyLookupTest, EagerLoadThenLookupReusesReader) { + auto* reader = _segment->_pk_index_reader.get(); + ASSERT_TRUE(_segment->load_pk_index_and_bf(nullptr).ok()); + auto* index = reader->_index_reader.get(); + auto* bf = reader->_bf.get(); + RowLocation location; + ASSERT_TRUE(lookup(_present_key, &location).ok()); + EXPECT_EQ(location.row_id, 0); + EXPECT_TRUE(lookup(_missing_key, &location).is()); + EXPECT_EQ(_segment->_pk_index_reader.get(), reader); + EXPECT_EQ(reader->_index_reader.get(), index); + EXPECT_EQ(reader->_bf.get(), bf); +} + +TEST_F(SegmentPrimaryKeyLookupTest, AddedSequenceColumnUsesUnsuffixedBloomKey) { + TabletSchema latest_schema; + latest_schema.copy_from(*_schema); + latest_schema._sequence_col_idx = 1; + std::string sequence_suffix(latest_schema.column(1).length() + 1, '\0'); + RowLocation location; + auto missing = _missing_key + sequence_suffix; + auto st = _segment->lookup_row_key(Slice(missing), &latest_schema, true, false, &location, + nullptr); + ASSERT_TRUE(st.is()) << st; + EXPECT_FALSE(_segment->_load_index_once.has_called()); + + auto present = _present_key + sequence_suffix; + std::string encoded_sequence = "not cleared"; + st = _segment->lookup_row_key(Slice(present), &latest_schema, true, false, &location, nullptr, + &encoded_sequence); + ASSERT_TRUE(st.ok()) << st; + EXPECT_EQ(location.row_id, 0); + EXPECT_TRUE(encoded_sequence.empty()); // The original segment has no sequence column. +} + +TEST_F(SegmentPrimaryKeyLookupTest, BloomMissStripsSequenceAndRowIdSuffixes) { + TabletSchema latest_schema; + latest_schema.copy_from(*_schema); + latest_schema._sequence_col_idx = 1; + latest_schema._cluster_key_uids = {0}; + auto missing = _missing_key + std::string(latest_schema.column(1).length() + 1 + + PrimaryKeyIndexReader::ROW_ID_LENGTH, + '\0'); + RowLocation location; + auto st = _segment->lookup_row_key(Slice(missing), &latest_schema, true, true, &location, + nullptr); + EXPECT_TRUE(st.is()) << st; + EXPECT_FALSE(_segment->_load_index_once.has_called()); +} + +TEST_F(SegmentPrimaryKeyLookupTest, ConcurrentIndexAndBloomFilterInitialization) { + std::promise start; + auto ready = start.get_future().share(); + std::vector> tasks; + for (int i = 0; i < 12; ++i) { + tasks.emplace_back(std::async(std::launch::async, [&, ready, i] { + ready.wait(); + if (i % 3 == 0) { + // This path need not initialize BF, and can race with BF-only lookup misses. + return _segment->load_index(nullptr); + } + RowLocation location; + auto st = lookup(i % 3 == 1 ? _present_key : _missing_key, &location); + if (i % 3 == 2) { + if (st.is()) { + return Status::OK(); + } + return st.ok() ? Status::InternalError("Expected a bloom-filter miss") : st; + } + if (st.ok() && location.row_id != 0) { + return Status::InternalError("Unexpected primary key row id"); + } + return st; + })); + } + start.set_value(); + for (auto& task : tasks) { + auto st = task.get(); + EXPECT_TRUE(st.ok()) << st; + } + EXPECT_TRUE(_segment->_load_index_once.has_called()); + EXPECT_TRUE(_segment->_load_pk_bf_once.has_called()); + EXPECT_TRUE(_segment->healthy_status().ok()); +} + +} // namespace doris From e5cd8ac31bbb491555427f4c96926b1bed901a96 Mon Sep 17 00:00:00 2001 From: laihui <1353307710@qq.com> Date: Mon, 21 Sep 2026 09:46:11 +0800 Subject: [PATCH 3/7] [fix](mow) Protect template comma in exception macro --- be/src/storage/segment/segment.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/be/src/storage/segment/segment.cpp b/be/src/storage/segment/segment.cpp index 43f0e4eaf00518..1bb755e16c5ae4 100644 --- a/be/src/storage/segment/segment.cpp +++ b/be/src/storage/segment/segment.cpp @@ -1178,7 +1178,7 @@ Status Segment::lookup_row_key(const Slice& key, const TabletSchema* latest_sche RETURN_IF_CATCH_EXCEPTION({ RETURN_IF_ERROR(_load_pk_bloom_filter(stats, io_ctx)); if (!_pk_index_reader->check_present(key_without_seq)) { - return Status::Error(""); + return (Status::Error("")); } RETURN_IF_ERROR(load_index(stats, io_ctx)); }); From d486cb00ca889296cfeb6b9b600450334e3de7da Mon Sep 17 00:00:00 2001 From: laihui <1353307710@qq.com> Date: Mon, 21 Sep 2026 15:58:29 +0800 Subject: [PATCH 4/7] [fix](mow) Charge pinned PK roots to segment cache eviction ### What problem does this PR solve? Related PR: #68304 SegmentCache retained large PK index roots while its LRU charge only reflected estimated metadata. Charge eagerly loaded roots at insertion and refresh the entry after lazy initialization, without charging the physical allocation twice. Grow LRU weights under the shard lock and preserve live handles and replacement identity. Discard partial readers on failed index initialization. ### Release note Reduce persistent PK root retention by accounting for pinned roots in SegmentCache eviction. Active readers can still retain pages beyond the cache budget, and more frequent eviction can increase index reloads. ### Check List (For Author) - Test: 11 new BE unit tests; execution blocked by dependency download timeouts. Actual low-level LRU source passed a standalone ASan/UBSan harness, including concurrent charge growth. Changed-file clang-format 16 and build hygiene pass. Full BE compilation and clang-tidy require CI (no compilation database locally). - Behavior changed: Yes, SegmentCache eviction includes loaded PK root bytes. - Does this need documentation: No. --- be/src/storage/index/primary_key_index.cpp | 12 ++- be/src/storage/segment/segment.cpp | 6 +- be/src/storage/segment/segment.h | 16 ++++ be/src/storage/segment/segment_loader.cpp | 21 ++++- be/src/util/lru_cache.cpp | 35 ++++++++ be/src/util/lru_cache.h | 7 ++ be/test/storage/cache/lru_cache_test.cpp | 88 +++++++++++++++++++ .../segment_primary_key_lookup_test.cpp | 83 +++++++++++++++++ 8 files changed, 262 insertions(+), 6 deletions(-) diff --git a/be/src/storage/index/primary_key_index.cpp b/be/src/storage/index/primary_key_index.cpp index 28fa377e4b4aee..080e19d503b727 100644 --- a/be/src/storage/index/primary_key_index.cpp +++ b/be/src/storage/index/primary_key_index.cpp @@ -112,12 +112,16 @@ Status PrimaryKeyIndexReader::parse_index(io::FileReaderSPtr file_reader, OlapReaderStatistics* pk_index_load_stats, const io::IOContext* source_io_ctx) { // parse primary key index - _index_reader.reset(new segment_v2::IndexedColumnReader(file_reader, meta.primary_key_index())); - _index_reader->set_is_pk_index(true); + auto index_reader = std::make_unique(file_reader, + meta.primary_key_index()); + index_reader->set_is_pk_index(true); auto io_ctx = create_index_io_context(source_io_ctx, pk_index_load_stats); - RETURN_IF_ERROR(_index_reader->load(!config::disable_pk_storage_page_cache, false, - pk_index_load_stats, &io_ctx)); + RETURN_IF_ERROR(index_reader->load(!config::disable_pk_storage_page_cache, false, + pk_index_load_stats, &io_ctx)); + // A failed initialization must not leave partially loaded root pages pinned in + // a cached segment without a corresponding eviction charge. + _index_reader = std::move(index_reader); _index_parsed = true; return Status::OK(); } diff --git a/be/src/storage/segment/segment.cpp b/be/src/storage/segment/segment.cpp index 1bb755e16c5ae4..72413b9f21f60f 100644 --- a/be/src/storage/segment/segment.cpp +++ b/be/src/storage/segment/segment.cpp @@ -788,7 +788,11 @@ Status Segment::load_index(OlapReaderStatistics* stats, const io::IOContext* sou DCHECK(_pk_index_reader != nullptr); RETURN_IF_ERROR(_pk_index_reader->parse_index(_file_reader, *_pk_index_meta, stats, source_io_ctx)); - // _meta_mem_usage += _pk_index_reader->get_memory_size(); + _pk_index_cache_bytes.store(_pk_index_reader->get_memory_size(), + std::memory_order_relaxed); + if (_cache_charge_callback) { + _cache_charge_callback(cache_charge()); + } return Status::OK(); } else { // read and parse short key index page diff --git a/be/src/storage/segment/segment.h b/be/src/storage/segment/segment.h index 9ed9cb77fbb874..92c168f0e4144d 100644 --- a/be/src/storage/segment/segment.h +++ b/be/src/storage/segment/segment.h @@ -22,7 +22,9 @@ #include #include +#include #include +#include #include #include // for unique_ptr #include @@ -181,6 +183,18 @@ class Segment : public std::enable_shared_from_this, public MetadataAdd // another method `get_metadata_size` not include the column reader, only the segment object itself. int64_t meta_mem_usage() const { return _meta_mem_usage; } + // PK pages are tracked by PKIndexPageCache, but their pinned bytes must also + // count toward SegmentCache eviction. Do not add these bytes to metadata tracking. + size_t cache_charge() const { + return _meta_mem_usage + _pk_index_cache_bytes.load(std::memory_order_relaxed); + } + + // Install before publishing this segment in SegmentCache; immutable afterwards. + void set_cache_charge_callback(std::function callback) { + DCHECK(!_cache_charge_callback); + _cache_charge_callback = std::move(callback); + } + // Variant paths use segment metadata; other columns use `read_type`. std::shared_ptr get_data_type_of(const TabletColumn& read_column, const DataTypePtr& read_type, @@ -292,6 +306,8 @@ class Segment : public std::enable_shared_from_this, public MetadataAdd // The memory consumed by querying is tracked in segment iterator. int64_t _meta_mem_usage; int64_t _tracked_meta_mem_usage = 0; + std::atomic _pk_index_cache_bytes {0}; + std::function _cache_charge_callback; RowsetId _rowset_id; TabletSchemaSPtr _tablet_schema; diff --git a/be/src/storage/segment/segment_loader.cpp b/be/src/storage/segment/segment_loader.cpp index e1c12cbf486131..c7f4a1a6d3b2ef 100644 --- a/be/src/storage/segment/segment_loader.cpp +++ b/be/src/storage/segment/segment_loader.cpp @@ -43,8 +43,27 @@ bool SegmentCache::lookup(const SegmentCache::CacheKey& key, SegmentCacheHandle* void SegmentCache::insert(const SegmentCache::CacheKey& key, SegmentCache::CacheValue& value, SegmentCacheHandle* handle) { + // Keep the cache weakly referenced: the cache owns the segment. The raw segment + // pointer is only compared while its own initialization invokes the callback. + value.segment->set_cache_charge_callback([cache = std::weak_ptr(_cache), + encoded_key = key.encode(), + segment = value.segment.get()](size_t charge) { + auto owner = cache.lock(); + if (!owner) { + return; // The Segment may outlive the cache during shutdown. + } + auto* entry = owner->lookup(encoded_key); + if (entry == nullptr) { + return; // Already evicted. + } + // Another instance of this segment may have replaced the cached entry. + if (static_cast(owner->value(entry))->segment.get() == segment) { + owner->update_charge(entry, charge); + } + owner->release(entry); + }); auto* lru_handle = - LRUCachePolicy::insert(key.encode(), &value, value.segment->meta_mem_usage(), + LRUCachePolicy::insert(key.encode(), &value, value.segment->cache_charge(), value.segment->meta_mem_usage(), CachePriority::NORMAL); handle->push_segment(this, lru_handle); } diff --git a/be/src/util/lru_cache.cpp b/be/src/util/lru_cache.cpp index 353c7303212376..9bc5b9914746b7 100644 --- a/be/src/util/lru_cache.cpp +++ b/be/src/util/lru_cache.cpp @@ -335,6 +335,36 @@ Cache::Handle* LRUCache::lookup(const CacheKey& key, uint32_t hash) { return reinterpret_cast(e); } +void LRUCache::update_charge(Cache::Handle* handle, size_t charge) { + auto* e = reinterpret_cast(handle); + LRUHandle* to_remove_head = nullptr; + { + std::lock_guard l(_mutex); + // Erase/replacement may have removed this entry while its handle was held. + if (!e->in_cache || charge <= e->charge) { + return; + } + DCHECK_GT(e->refs, 1); + size_t delta = charge - e->charge; + e->charge = charge; + e->total_size += delta; + _usage += delta; + if (_cache_value_check_timestamp) { + _evict_from_lru_with_time(0, &to_remove_head); + } else { + _evict_from_lru(0, &to_remove_head); + } + } + // Cache values can release other caches; never destroy them under the shard lock. + while (to_remove_head != nullptr) { + LRUHandle* next = to_remove_head->next; + to_remove_head->free(); + to_remove_head = next; + } + // If the updated entry itself exceeds capacity, release() evicts it after the + // last outstanding cache handle is released. +} + void LRUCache::release(Cache::Handle* handle) { if (handle == nullptr) { return; @@ -769,6 +799,11 @@ void ShardedLRUCache::release(Handle* handle) { _shards[_shard(h->hash)]->release(handle); } +void ShardedLRUCache::update_charge(Handle* handle, size_t charge) { + auto* h = reinterpret_cast(handle); + _shards[_shard(h->hash)]->update_charge(handle, charge); +} + void ShardedLRUCache::erase(const CacheKey& key) { const uint32_t hash = _hash_slice(key); _shards[_shard(hash)]->erase(key, hash); diff --git a/be/src/util/lru_cache.h b/be/src/util/lru_cache.h index f01fd0cd6cf5e2..718baabba7f275 100644 --- a/be/src/util/lru_cache.h +++ b/be/src/util/lru_cache.h @@ -193,6 +193,10 @@ class Cache { // REQUIRES: handle must have been returned by a method on *this. virtual void release(Handle* handle) = 0; + // Grow an entry's eviction charge after lazy initialization, without changing its + // memory tracker. The caller must hold a live handle. Stale smaller charges are ignored. + virtual void update_charge(Handle* handle, size_t charge) = 0; + // Return the value encapsulated in a handle returned by a // successful lookup(). // REQUIRES: handle must not have been released yet. @@ -348,6 +352,7 @@ class LRUCache { CachePriority priority = CachePriority::NORMAL); Cache::Handle* lookup(const CacheKey& key, uint32_t hash); void release(Cache::Handle* handle); + void update_charge(Cache::Handle* handle, size_t charge); void erase(const CacheKey& key, uint32_t hash); PrunedInfo prune(); PrunedInfo prune_if(CachePrunePredicate pred, bool lazy_mode = false); @@ -420,6 +425,7 @@ class ShardedLRUCache : public Cache { CachePriority priority = CachePriority::NORMAL) override; Handle* lookup(const CacheKey& key) override; void release(Handle* handle) override; + void update_charge(Handle* handle, size_t charge) override; void erase(const CacheKey& key) override; void* value(Handle* handle) override; uint64_t new_id() override; @@ -484,6 +490,7 @@ class DummyLRUCache : public Cache { CachePriority priority = CachePriority::NORMAL) override; Handle* lookup(const CacheKey& key) override { return nullptr; }; void release(Handle* handle) override; + void update_charge(Handle* handle, size_t charge) override {} void erase(const CacheKey& key) override {}; void* value(Handle* handle) override; uint64_t new_id() override { return 0; }; diff --git a/be/test/storage/cache/lru_cache_test.cpp b/be/test/storage/cache/lru_cache_test.cpp index a31d532a5a4686..76c19eb03461d4 100644 --- a/be/test/storage/cache/lru_cache_test.cpp +++ b/be/test/storage/cache/lru_cache_test.cpp @@ -21,6 +21,7 @@ #include #include +#include #include #include @@ -297,6 +298,93 @@ TEST_F(CacheTest, EvictionPolicyWithDurable) { EXPECT_EQ(201, Lookup(200)); } +TEST_F(CacheTest, UpdateChargeDoesNotDoubleTrackMemory) { + init_size_cache(1024 * 1024); + const CacheKey key("growing"); + auto* handle = cache()->insert(key, new CacheValue(EncodeValue(1)), 100, 100); + const auto usage = cache()->get_usage(); + const auto tracked = cache()->mem_consumption(); + cache()->_cache->update_charge(handle, 500); + EXPECT_EQ(cache()->get_usage(), usage + 400); + EXPECT_EQ(cache()->mem_consumption(), tracked); + // Repeated initialization and older snapshots cannot undo the larger charge. + cache()->_cache->update_charge(handle, 500); + cache()->_cache->update_charge(handle, 200); + EXPECT_EQ(cache()->get_usage(), usage + 400); + cache()->release(handle); + cache()->prune_all(true); + EXPECT_EQ(cache()->get_usage(), 0); + EXPECT_EQ(cache()->mem_consumption(), 0); +} + +TEST_F(CacheTest, UpdateChargeEvictsIdleEntriesAndReleasesOversizedEntry) { + LRUCache cache(LRUCacheType::SIZE); + cache.set_capacity(1024); + CacheKey idle("idle"), growing("growing"); + cache.release(cache.insert(idle, 1, new CacheValue(EncodeValue(1)), 100)); + auto* handle = cache.insert(growing, 2, new CacheValue(EncodeValue(2)), 100); + cache.update_charge(handle, 2048); + EXPECT_EQ(cache.lookup(idle, 1), nullptr); + EXPECT_GT(cache.get_usage(), 2048); + // The caller's live handle remains usable even while over capacity. + EXPECT_EQ(DecodeValue( + static_cast(reinterpret_cast(handle)->value)->value), + 2); + cache.release(handle); + EXPECT_EQ(cache.get_usage(), 0); + EXPECT_EQ(cache.lookup(growing, 2), nullptr); +} + +TEST_F(CacheTest, UpdateChargeIgnoresReplacedEntry) { + LRUCache cache(LRUCacheType::SIZE); + cache.set_capacity(4096); + CacheKey key("replaced"); + auto* old = cache.insert(key, 1, new CacheValue(EncodeValue(1)), 100); + auto* current = cache.insert(key, 1, new CacheValue(EncodeValue(2)), 200); + const auto usage = cache.get_usage(); + cache.update_charge(old, 8192); + EXPECT_EQ(cache.get_usage(), usage); + cache.release(old); + cache.release(current); + EXPECT_EQ(cache.get_usage(), usage); +} + +TEST_F(CacheTest, UpdateChargeConcurrentGrowth) { + LRUCache cache(LRUCacheType::SIZE); + cache.set_capacity(1024 * 1024); + CacheKey key("concurrent"); + auto* handle = cache.insert(key, 1, new CacheValue(EncodeValue(1)), 100); + const auto usage = cache.get_usage(); + std::vector> updates; + for (size_t charge = 200; charge <= 900; charge += 100) { + auto* held = cache.lookup(key, 1); + ASSERT_NE(held, nullptr); + updates.emplace_back(std::async(std::launch::async, [&, held, charge] { + cache.update_charge(held, charge); + cache.release(held); + })); + } + for (auto& update : updates) { + update.get(); + } + EXPECT_EQ(cache.get_usage(), usage + 800); + cache.release(handle); +} + +TEST_F(CacheTest, UpdateChargePreservesOtherOutstandingHandles) { + LRUCache cache(LRUCacheType::SIZE); + cache.set_capacity(1024); + CacheKey key("shared"); + auto* first = cache.insert(key, 1, new CacheValue(EncodeValue(1)), 100); + auto* second = cache.lookup(key, 1); + ASSERT_NE(second, nullptr); + cache.update_charge(first, 2048); + cache.release(first); + EXPECT_GT(cache.get_usage(), 2048); + cache.release(second); + EXPECT_EQ(cache.get_usage(), 0); +} + TEST_F(CacheTest, Usage) { LRUCache cache(LRUCacheType::SIZE); cache.set_capacity(1040); diff --git a/be/test/storage/segment/segment_primary_key_lookup_test.cpp b/be/test/storage/segment/segment_primary_key_lookup_test.cpp index 52354e06ee862a..a346ddb0e20abf 100644 --- a/be/test/storage/segment/segment_primary_key_lookup_test.cpp +++ b/be/test/storage/segment/segment_primary_key_lookup_test.cpp @@ -28,6 +28,7 @@ #include "runtime/exec_env.h" #include "storage/index/primary_key_index.h" #include "storage/segment/segment.h" +#include "storage/segment/segment_loader.h" #include "storage/segment/segment_writer.h" #include "storage/storage_engine.h" #include "storage/tablet/tablet_schema_helper.h" @@ -105,6 +106,88 @@ class SegmentPrimaryKeyLookupTest : public testing::Test { std::string _missing_key; }; +TEST_F(SegmentPrimaryKeyLookupTest, CachedRootPagesIncreaseChargeOnlyOnce) { + SegmentCache cache(1024 * 1024 * 1024, 10000); + SegmentCacheHandle handle; + const SegmentCache::CacheKey key(_segment->rowset_id(), _segment->id()); + cache.insert(key, *new SegmentCache::CacheValue(_segment), &handle); + const auto usage = cache.get_usage(); + const auto tracked = cache.mem_consumption(); + RowLocation location; + ASSERT_TRUE(lookup(_missing_key, &location).is()); + EXPECT_EQ(cache.get_usage(), usage); + ASSERT_TRUE(lookup(_present_key, &location).ok()); + const auto roots = _segment->_pk_index_reader->get_memory_size(); + EXPECT_GT(roots, 300000); + EXPECT_EQ(cache.get_usage(), usage + roots); + EXPECT_EQ(cache.mem_consumption(), tracked); + ASSERT_TRUE(_segment->load_index(nullptr).ok()); + EXPECT_EQ(cache.get_usage(), usage + roots); +} + +TEST_F(SegmentPrimaryKeyLookupTest, LazyRootLoadingEvictsOversizedSegment) { + // SegmentCache has 64 shards. Metadata fits; the large PK roots do not. + SegmentCache cache((_segment->cache_charge() + 65536) * 64, 10000); + SegmentCacheHandle handle; + const SegmentCache::CacheKey key(_segment->rowset_id(), _segment->id()); + cache.insert(key, *new SegmentCache::CacheValue(_segment), &handle); + ASSERT_EQ(cache.get_element_count(), 1); + ASSERT_TRUE(_segment->load_index(nullptr).ok()); + EXPECT_EQ(cache.get_element_count(), 0); + EXPECT_EQ(cache.get_usage(), 0); + // Eviction drops cache ownership, not the active operation's shared_ptr. + RowLocation location; + ASSERT_TRUE(lookup(_present_key, &location).ok()); + EXPECT_EQ(location.row_id, 0); +} + +TEST_F(SegmentPrimaryKeyLookupTest, EagerRootLoadingChargesOnInsert) { + const auto capacity = (_segment->cache_charge() + 65536) * 64; + ASSERT_TRUE(_segment->load_index(nullptr).ok()); + SegmentCache cache(capacity, 10000); + SegmentCacheHandle handle; + const SegmentCache::CacheKey key(_segment->rowset_id(), _segment->id()); + cache.insert(key, *new SegmentCache::CacheValue(_segment), &handle); + EXPECT_EQ(cache.get_usage(), 0); + EXPECT_EQ(cache.get_element_count(), 0); + EXPECT_EQ(handle.get_segments().front(), _segment); +} + +TEST_F(SegmentPrimaryKeyLookupTest, OldSegmentDoesNotChargeItsReplacement) { + SegmentCache cache(1024 * 1024 * 1024, 10000); + const SegmentCache::CacheKey key(_segment->rowset_id(), _segment->id()); + SegmentCacheHandle old_handle, new_handle; + cache.insert(key, *new SegmentCache::CacheValue(_segment), &old_handle); + auto replacement = std::make_shared(_segment->id(), _segment->rowset_id(), + _schema, InvertedIndexFileInfo {}); + cache.insert(key, *new SegmentCache::CacheValue(replacement), &new_handle); + const auto usage = cache.get_usage(); + ASSERT_TRUE(_segment->load_index(nullptr).ok()); + EXPECT_EQ(cache.get_usage(), usage); + EXPECT_EQ(cache.get_element_count(), 1); +} + +TEST_F(SegmentPrimaryKeyLookupTest, SegmentCanOutliveCache) { + { + SegmentCache cache(1024 * 1024 * 1024, 10000); + SegmentCacheHandle handle; + const SegmentCache::CacheKey key(_segment->rowset_id(), _segment->id()); + cache.insert(key, *new SegmentCache::CacheValue(_segment), &handle); + } + ASSERT_TRUE(_segment->load_index(nullptr).ok()); + RowLocation location; + ASSERT_TRUE(lookup(_present_key, &location).ok()); +} + +TEST_F(SegmentPrimaryKeyLookupTest, FailedIndexLoadDoesNotPinPartialRoots) { + auto meta = *_segment->_pk_index_meta; + meta.mutable_primary_key_index()->mutable_value_index_meta()->mutable_root_page()->set_offset( + uint64_t {1} << 50); + PrimaryKeyIndexReader reader; + EXPECT_FALSE(reader.parse_index(_segment->file_reader(), meta, nullptr).ok()); + EXPECT_EQ(reader._index_reader, nullptr); +} + TEST_F(SegmentPrimaryKeyLookupTest, BloomFilterMissDoesNotLoadIndex) { RowLocation location; for (int i = 0; i < 3; ++i) { From 0314afd9c8b642734221f444c42a55ce8f36d8ea Mon Sep 17 00:00:00 2001 From: laihui <1353307710@qq.com> Date: Mon, 21 Sep 2026 16:10:30 +0800 Subject: [PATCH 5/7] [fix](mow) Isolate segment cache charging from lookup optimizations ### What problem does this PR solve? Related PR: #68304 Restore the original eager rowset PK-index loading and index-before-bloom-filter lookup order so the pinned-root SegmentCache charge change can be measured alone. Restore reader creation during index loading and replace BF-first tests with the six cache-charge tests. A definite bloom miss now verifies index-first loading and the corresponding cache charge increase. ### Release note Only the SegmentCache pinned-root accounting fix remains in this PR. ### Check List (For Author) - Test: Verified rowset loading and lookup implementation match baseline 695c88b5772. clang-format 16 and build hygiene passed. Full BE build/tests remain pending CI due to the previously observed dependency download failures. - Behavior changed: Yes, removes the earlier lazy-loading and BF-first changes. - Does this need documentation: No. --- be/src/storage/segment/segment.cpp | 22 +-- be/src/storage/segment/segment.h | 4 +- be/src/storage/tablet/base_tablet.cpp | 5 +- ...cpp => segment_primary_key_cache_test.cpp} | 153 ++---------------- 4 files changed, 20 insertions(+), 164 deletions(-) rename be/test/storage/segment/{segment_primary_key_lookup_test.cpp => segment_primary_key_cache_test.cpp} (54%) diff --git a/be/src/storage/segment/segment.cpp b/be/src/storage/segment/segment.cpp index 72413b9f21f60f..3b7bd70b851d35 100644 --- a/be/src/storage/segment/segment.cpp +++ b/be/src/storage/segment/segment.cpp @@ -383,11 +383,6 @@ Status Segment::_open(OlapReaderStatistics* stats, const io::IOContext* source_i footer_pb_shared->has_primary_key_index_meta() ? new PrimaryKeyIndexMetaPB(footer_pb_shared->primary_key_index_meta()) : nullptr); - if (_tablet_schema->keys_type() == UNIQUE_KEYS && _pk_index_meta != nullptr) { - // Create the shared reader before publishing the segment. Its index and bloom filter - // can then be initialized independently without replacing each other's state. - _pk_index_reader = std::make_unique(); - } // delete_bitmap_calculator_test.cpp // DCHECK(footer.has_short_key_index_page()); _sk_index_page = footer_pb_shared->short_key_index_page(); @@ -785,7 +780,7 @@ Status Segment::load_pk_index_and_bf(OlapReaderStatistics* index_load_stats, Status Segment::load_index(OlapReaderStatistics* stats, const io::IOContext* source_io_ctx) { return _load_index_once.call([this, stats, source_io_ctx] { if (_tablet_schema->keys_type() == UNIQUE_KEYS && _pk_index_meta != nullptr) { - DCHECK(_pk_index_reader != nullptr); + _pk_index_reader = std::make_unique(); RETURN_IF_ERROR(_pk_index_reader->parse_index(_file_reader, *_pk_index_meta, stats, source_io_ctx)); _pk_index_cache_bytes.store(_pk_index_reader->get_memory_size(), @@ -1164,6 +1159,7 @@ Status Segment::lookup_row_key(const Slice& key, const TabletSchema* latest_sche bool with_seq_col, bool with_rowid, RowLocation* row_location, OlapReaderStatistics* stats, std::string* encoded_seq_value, const io::IOContext* io_ctx) { + RETURN_IF_ERROR(load_pk_index_and_bf(stats, io_ctx)); bool has_seq_col = latest_schema->has_sequence_col(); bool has_rowid = !latest_schema->cluster_key_uids().empty(); size_t seq_col_length = 0; @@ -1176,16 +1172,10 @@ Status Segment::lookup_row_key(const Slice& key, const TabletSchema* latest_sche Slice(key.get_data(), key.get_size() - (with_seq_col ? seq_col_length : 0) - (with_rowid ? rowid_length : 0)); - // A bloom-filter miss must not load the potentially large PK index root pages. - // Preserve the exception boundary of load_pk_index_and_bf: DorisCallOnce can rethrow - // an initialization exception on a later call as well as on the first call. - RETURN_IF_CATCH_EXCEPTION({ - RETURN_IF_ERROR(_load_pk_bloom_filter(stats, io_ctx)); - if (!_pk_index_reader->check_present(key_without_seq)) { - return (Status::Error("")); - } - RETURN_IF_ERROR(load_index(stats, io_ctx)); - }); + DCHECK(_pk_index_reader != nullptr); + if (!_pk_index_reader->check_present(key_without_seq)) { + return Status::Error(""); + } bool exact_match = false; std::unique_ptr index_iterator; RETURN_IF_ERROR(_pk_index_reader->new_iterator(&index_iterator, stats, io_ctx)); diff --git a/be/src/storage/segment/segment.h b/be/src/storage/segment/segment.h index 92c168f0e4144d..31715425599ac0 100644 --- a/be/src/storage/segment/segment.h +++ b/be/src/storage/segment/segment.h @@ -325,7 +325,7 @@ class Segment : public std::enable_shared_from_this, public MetadataAdd // map column unique id ---> it's inner data type std::map> _file_column_types; - // used to guarantee that the short key or primary key index is loaded at most once + // used to guarantee that short key index will be loaded at most once in a thread-safe way DorisCallOnce _load_index_once; // used to guarantee that primary key bloom filter will be loaded at most once in a thread-safe way DorisCallOnce _load_pk_bf_once; @@ -342,7 +342,7 @@ class Segment : public std::enable_shared_from_this, public MetadataAdd // short key index decoder // all content is in memory std::unique_ptr _sk_index_decoder; - // Created in _open before the segment is shared; PK index and BF are loaded independently. + // primary key index reader std::unique_ptr _pk_index_reader; std::mutex _open_lock; // inverted index file reader diff --git a/be/src/storage/tablet/base_tablet.cpp b/be/src/storage/tablet/base_tablet.cpp index e40c815452a07f..c7ad7a6c639e93 100644 --- a/be/src/storage/tablet/base_tablet.cpp +++ b/be/src/storage/tablet/base_tablet.cpp @@ -530,11 +530,8 @@ Status BaseTablet::lookup_row_key(const Slice& encoded_key, TabletSchema* latest if (UNLIKELY(segment_caches[i] == nullptr)) { segment_caches[i] = std::make_unique(); - // Keep segment handles for reuse, but load PK indexes and bloom filters only - // when lookup_row_key visits a picked segment. Eagerly loading the whole rowset - // can retain large PK index pages even for segments excluded by key bounds. RETURN_IF_ERROR(SegmentLoader::instance()->load_segments( - std::static_pointer_cast(rs), segment_caches[i].get(), true, false, + std::static_pointer_cast(rs), segment_caches[i].get(), true, true, stats, io_ctx)); } auto& segments = segment_caches[i]->get_segments(); diff --git a/be/test/storage/segment/segment_primary_key_lookup_test.cpp b/be/test/storage/segment/segment_primary_key_cache_test.cpp similarity index 54% rename from be/test/storage/segment/segment_primary_key_lookup_test.cpp rename to be/test/storage/segment/segment_primary_key_cache_test.cpp index a346ddb0e20abf..977036d2c5f85e 100644 --- a/be/test/storage/segment/segment_primary_key_lookup_test.cpp +++ b/be/test/storage/segment/segment_primary_key_cache_test.cpp @@ -18,7 +18,6 @@ #include #include -#include #include #include #include @@ -41,7 +40,7 @@ void build_segment(SegmentWriterOptions opts, TabletSchemaSPtr build_schema, siz TabletSchemaSPtr query_schema, size_t nrows, Generator generator, std::shared_ptr* res, std::string segment_dir); -class SegmentPrimaryKeyLookupTest : public testing::Test { +class SegmentPrimaryKeyCacheTest : public testing::Test { protected: void SetUp() override { auto fs = io::global_local_filesystem(); @@ -66,7 +65,7 @@ class SegmentPrimaryKeyLookupTest : public testing::Test { }; build_segment(opts, _schema, 0, _schema, 8, generator, &_segment, _dir); ASSERT_NE(_segment, nullptr); - ASSERT_NE(_segment->_pk_index_reader, nullptr); + ASSERT_EQ(_segment->_pk_index_reader, nullptr); ASSERT_FALSE(_segment->_load_index_once.has_called()); ASSERT_FALSE(_segment->_load_pk_bf_once.has_called()); ASSERT_FALSE(_segment->_pk_index_meta->primary_key_index() @@ -99,14 +98,14 @@ class SegmentPrimaryKeyLookupTest : public testing::Test { return _segment->lookup_row_key(Slice(key), _schema.get(), false, false, location, nullptr); } - const std::string _dir = "./ut_dir/segment_primary_key_lookup_test"; + const std::string _dir = "./ut_dir/segment_primary_key_cache_test"; TabletSchemaSPtr _schema; std::shared_ptr _segment; std::string _present_key; std::string _missing_key; }; -TEST_F(SegmentPrimaryKeyLookupTest, CachedRootPagesIncreaseChargeOnlyOnce) { +TEST_F(SegmentPrimaryKeyCacheTest, CachedRootPagesIncreaseChargeOnlyOnce) { SegmentCache cache(1024 * 1024 * 1024, 10000); SegmentCacheHandle handle; const SegmentCache::CacheKey key(_segment->rowset_id(), _segment->id()); @@ -115,17 +114,17 @@ TEST_F(SegmentPrimaryKeyLookupTest, CachedRootPagesIncreaseChargeOnlyOnce) { const auto tracked = cache.mem_consumption(); RowLocation location; ASSERT_TRUE(lookup(_missing_key, &location).is()); - EXPECT_EQ(cache.get_usage(), usage); - ASSERT_TRUE(lookup(_present_key, &location).ok()); + // Keep the original index-first lookup: even a definite BF miss loads the roots. const auto roots = _segment->_pk_index_reader->get_memory_size(); EXPECT_GT(roots, 300000); EXPECT_EQ(cache.get_usage(), usage + roots); EXPECT_EQ(cache.mem_consumption(), tracked); + ASSERT_TRUE(lookup(_present_key, &location).ok()); ASSERT_TRUE(_segment->load_index(nullptr).ok()); EXPECT_EQ(cache.get_usage(), usage + roots); } -TEST_F(SegmentPrimaryKeyLookupTest, LazyRootLoadingEvictsOversizedSegment) { +TEST_F(SegmentPrimaryKeyCacheTest, LazyRootLoadingEvictsOversizedSegment) { // SegmentCache has 64 shards. Metadata fits; the large PK roots do not. SegmentCache cache((_segment->cache_charge() + 65536) * 64, 10000); SegmentCacheHandle handle; @@ -141,7 +140,7 @@ TEST_F(SegmentPrimaryKeyLookupTest, LazyRootLoadingEvictsOversizedSegment) { EXPECT_EQ(location.row_id, 0); } -TEST_F(SegmentPrimaryKeyLookupTest, EagerRootLoadingChargesOnInsert) { +TEST_F(SegmentPrimaryKeyCacheTest, EagerRootLoadingChargesOnInsert) { const auto capacity = (_segment->cache_charge() + 65536) * 64; ASSERT_TRUE(_segment->load_index(nullptr).ok()); SegmentCache cache(capacity, 10000); @@ -153,7 +152,7 @@ TEST_F(SegmentPrimaryKeyLookupTest, EagerRootLoadingChargesOnInsert) { EXPECT_EQ(handle.get_segments().front(), _segment); } -TEST_F(SegmentPrimaryKeyLookupTest, OldSegmentDoesNotChargeItsReplacement) { +TEST_F(SegmentPrimaryKeyCacheTest, OldSegmentDoesNotChargeItsReplacement) { SegmentCache cache(1024 * 1024 * 1024, 10000); const SegmentCache::CacheKey key(_segment->rowset_id(), _segment->id()); SegmentCacheHandle old_handle, new_handle; @@ -167,7 +166,7 @@ TEST_F(SegmentPrimaryKeyLookupTest, OldSegmentDoesNotChargeItsReplacement) { EXPECT_EQ(cache.get_element_count(), 1); } -TEST_F(SegmentPrimaryKeyLookupTest, SegmentCanOutliveCache) { +TEST_F(SegmentPrimaryKeyCacheTest, SegmentCanOutliveCache) { { SegmentCache cache(1024 * 1024 * 1024, 10000); SegmentCacheHandle handle; @@ -179,7 +178,7 @@ TEST_F(SegmentPrimaryKeyLookupTest, SegmentCanOutliveCache) { ASSERT_TRUE(lookup(_present_key, &location).ok()); } -TEST_F(SegmentPrimaryKeyLookupTest, FailedIndexLoadDoesNotPinPartialRoots) { +TEST_F(SegmentPrimaryKeyCacheTest, FailedIndexLoadDoesNotPinPartialRoots) { auto meta = *_segment->_pk_index_meta; meta.mutable_primary_key_index()->mutable_value_index_meta()->mutable_root_page()->set_offset( uint64_t {1} << 50); @@ -188,134 +187,4 @@ TEST_F(SegmentPrimaryKeyLookupTest, FailedIndexLoadDoesNotPinPartialRoots) { EXPECT_EQ(reader._index_reader, nullptr); } -TEST_F(SegmentPrimaryKeyLookupTest, BloomFilterMissDoesNotLoadIndex) { - RowLocation location; - for (int i = 0; i < 3; ++i) { - auto st = lookup(_missing_key, &location); - EXPECT_TRUE(st.is()) << st; - EXPECT_TRUE(_segment->_load_pk_bf_once.has_called()); - EXPECT_FALSE(_segment->_load_index_once.has_called()); - EXPECT_EQ(_segment->_pk_index_reader->_index_reader, nullptr); - EXPECT_TRUE(_segment->healthy_status().ok()); - } -} - -TEST_F(SegmentPrimaryKeyLookupTest, HitAfterMissPreservesBloomFilter) { - RowLocation location; - ASSERT_TRUE(lookup(_missing_key, &location).is()); - auto* reader = _segment->_pk_index_reader.get(); - auto* bf = reader->_bf.get(); - ASSERT_FALSE(_segment->_load_index_once.has_called()); - - ASSERT_TRUE(lookup(_present_key, &location).ok()); - EXPECT_EQ(location.segment_id, _segment->id()); - EXPECT_EQ(location.row_id, 0); - EXPECT_TRUE(_segment->_load_index_once.has_called()); - EXPECT_EQ(_segment->_pk_index_reader.get(), reader); - EXPECT_EQ(reader->_bf.get(), bf); - auto* index = reader->_index_reader.get(); - ASSERT_NE(index, nullptr); - - ASSERT_TRUE(lookup(_present_key, &location).ok()); - EXPECT_EQ(reader->_index_reader.get(), index); - EXPECT_EQ(reader->_bf.get(), bf); - EXPECT_TRUE(lookup(_missing_key, &location).is()); -} - -TEST_F(SegmentPrimaryKeyLookupTest, BloomFilterPositiveStillChecksExactKey) { - ASSERT_TRUE(_segment->_load_pk_bloom_filter(nullptr).ok()); - // Deliberately make the BF positive for an absent key to exercise the false-positive path. - _segment->_pk_index_reader->_bf->add_bytes(_missing_key.data(), _missing_key.size()); - ASSERT_TRUE(_segment->_pk_index_reader->check_present(Slice(_missing_key))); - RowLocation location; - auto st = lookup(_missing_key, &location); - EXPECT_TRUE(st.is()) << st; - EXPECT_TRUE(_segment->_load_index_once.has_called()); -} - -TEST_F(SegmentPrimaryKeyLookupTest, EagerLoadThenLookupReusesReader) { - auto* reader = _segment->_pk_index_reader.get(); - ASSERT_TRUE(_segment->load_pk_index_and_bf(nullptr).ok()); - auto* index = reader->_index_reader.get(); - auto* bf = reader->_bf.get(); - RowLocation location; - ASSERT_TRUE(lookup(_present_key, &location).ok()); - EXPECT_EQ(location.row_id, 0); - EXPECT_TRUE(lookup(_missing_key, &location).is()); - EXPECT_EQ(_segment->_pk_index_reader.get(), reader); - EXPECT_EQ(reader->_index_reader.get(), index); - EXPECT_EQ(reader->_bf.get(), bf); -} - -TEST_F(SegmentPrimaryKeyLookupTest, AddedSequenceColumnUsesUnsuffixedBloomKey) { - TabletSchema latest_schema; - latest_schema.copy_from(*_schema); - latest_schema._sequence_col_idx = 1; - std::string sequence_suffix(latest_schema.column(1).length() + 1, '\0'); - RowLocation location; - auto missing = _missing_key + sequence_suffix; - auto st = _segment->lookup_row_key(Slice(missing), &latest_schema, true, false, &location, - nullptr); - ASSERT_TRUE(st.is()) << st; - EXPECT_FALSE(_segment->_load_index_once.has_called()); - - auto present = _present_key + sequence_suffix; - std::string encoded_sequence = "not cleared"; - st = _segment->lookup_row_key(Slice(present), &latest_schema, true, false, &location, nullptr, - &encoded_sequence); - ASSERT_TRUE(st.ok()) << st; - EXPECT_EQ(location.row_id, 0); - EXPECT_TRUE(encoded_sequence.empty()); // The original segment has no sequence column. -} - -TEST_F(SegmentPrimaryKeyLookupTest, BloomMissStripsSequenceAndRowIdSuffixes) { - TabletSchema latest_schema; - latest_schema.copy_from(*_schema); - latest_schema._sequence_col_idx = 1; - latest_schema._cluster_key_uids = {0}; - auto missing = _missing_key + std::string(latest_schema.column(1).length() + 1 + - PrimaryKeyIndexReader::ROW_ID_LENGTH, - '\0'); - RowLocation location; - auto st = _segment->lookup_row_key(Slice(missing), &latest_schema, true, true, &location, - nullptr); - EXPECT_TRUE(st.is()) << st; - EXPECT_FALSE(_segment->_load_index_once.has_called()); -} - -TEST_F(SegmentPrimaryKeyLookupTest, ConcurrentIndexAndBloomFilterInitialization) { - std::promise start; - auto ready = start.get_future().share(); - std::vector> tasks; - for (int i = 0; i < 12; ++i) { - tasks.emplace_back(std::async(std::launch::async, [&, ready, i] { - ready.wait(); - if (i % 3 == 0) { - // This path need not initialize BF, and can race with BF-only lookup misses. - return _segment->load_index(nullptr); - } - RowLocation location; - auto st = lookup(i % 3 == 1 ? _present_key : _missing_key, &location); - if (i % 3 == 2) { - if (st.is()) { - return Status::OK(); - } - return st.ok() ? Status::InternalError("Expected a bloom-filter miss") : st; - } - if (st.ok() && location.row_id != 0) { - return Status::InternalError("Unexpected primary key row id"); - } - return st; - })); - } - start.set_value(); - for (auto& task : tasks) { - auto st = task.get(); - EXPECT_TRUE(st.ok()) << st; - } - EXPECT_TRUE(_segment->_load_index_once.has_called()); - EXPECT_TRUE(_segment->_load_pk_bf_once.has_called()); - EXPECT_TRUE(_segment->healthy_status().ok()); -} - } // namespace doris From 15ae743ff01fa4c7e31e44285456dfaa9b953314 Mon Sep 17 00:00:00 2001 From: laihui <1353307710@qq.com> Date: Mon, 21 Sep 2026 21:17:10 +0800 Subject: [PATCH 6/7] [fix](mow) Preserve cached segments when PK roots exceed capacity ### What problem does this PR solve? Related PR: #68304 A segment whose PK roots grow beyond its cache shard capacity evicts every other idle entry before being evicted itself. Remove cache ownership of the oversized entry immediately, preserving live handles and unrelated entries. Keep normal eviction when the growing entry can still fit in the shard. Update the Segment cache tests to the existing VerticalSegmentWriterOptions interface and matching build_segment helper signature. ### Release note Avoid unnecessary segment cache eviction when loaded PK roots exceed shard capacity. ### Check List (For Author) - Test: Added/updated BE unit tests for oversized entries, normal eviction, timestamp eviction, outstanding handles, and physical memory accounting. Extracted five low-level test bodies and the original LRU implementation into an external-dependency-stub harness: ASan/UBSan passed after the fix; the baseline failed the oversized-entry regression assertion. Changed-file clang-format 16, build hygiene and git diff --check passed. Full BE UT was attempted with CacheTest.UpdateCharge*:SegmentPrimaryKeyCacheTest.*; configuration failed before compilation because protoc, Boost and Snappy dependencies are missing. Full BE UT and clang-tidy remain pending. - Behavior changed: Yes; oversized growing entries are removed without evicting unrelated entries, while live handles retain ownership. - Does this need documentation: No --- be/src/util/lru_cache.cpp | 11 ++- be/test/storage/cache/lru_cache_test.cpp | 71 +++++++++++++++---- .../segment_primary_key_cache_test.cpp | 10 +-- 3 files changed, 73 insertions(+), 19 deletions(-) diff --git a/be/src/util/lru_cache.cpp b/be/src/util/lru_cache.cpp index 9bc5b9914746b7..daa3184067ab75 100644 --- a/be/src/util/lru_cache.cpp +++ b/be/src/util/lru_cache.cpp @@ -349,6 +349,15 @@ void LRUCache::update_charge(Cache::Handle* handle, size_t charge) { e->charge = charge; e->total_size += delta; _usage += delta; + if (e->total_size > _capacity) { + // This entry cannot fit even after evicting every other entry. Drop + // cache ownership now, while outstanding handles keep it alive. + bool removed = _table.remove(e); + DCHECK(removed); + e->in_cache = false; + _unref(e); + _usage -= e->total_size; + } if (_cache_value_check_timestamp) { _evict_from_lru_with_time(0, &to_remove_head); } else { @@ -361,8 +370,6 @@ void LRUCache::update_charge(Cache::Handle* handle, size_t charge) { to_remove_head->free(); to_remove_head = next; } - // If the updated entry itself exceeds capacity, release() evicts it after the - // last outstanding cache handle is released. } void LRUCache::release(Cache::Handle* handle) { diff --git a/be/test/storage/cache/lru_cache_test.cpp b/be/test/storage/cache/lru_cache_test.cpp index 76c19eb03461d4..76a45b54e65bf6 100644 --- a/be/test/storage/cache/lru_cache_test.cpp +++ b/be/test/storage/cache/lru_cache_test.cpp @@ -311,28 +311,65 @@ TEST_F(CacheTest, UpdateChargeDoesNotDoubleTrackMemory) { cache()->_cache->update_charge(handle, 500); cache()->_cache->update_charge(handle, 200); EXPECT_EQ(cache()->get_usage(), usage + 400); + cache()->_cache->update_charge(handle, 2 * 1024 * 1024); + EXPECT_EQ(cache()->get_usage(), 0); + // Detaching an oversized entry does not release its live handle's memory. + EXPECT_EQ(cache()->mem_consumption(), tracked); cache()->release(handle); cache()->prune_all(true); EXPECT_EQ(cache()->get_usage(), 0); EXPECT_EQ(cache()->mem_consumption(), 0); } -TEST_F(CacheTest, UpdateChargeEvictsIdleEntriesAndReleasesOversizedEntry) { +TEST_F(CacheTest, UpdateChargeEvictsIdleEntriesWhenEntryStillFits) { LRUCache cache(LRUCacheType::SIZE); cache.set_capacity(1024); - CacheKey idle("idle"), growing("growing"); + CacheKey idle("idle"), durable("durable"), growing("growing"); cache.release(cache.insert(idle, 1, new CacheValue(EncodeValue(1)), 100)); + cache.release( + cache.insert(durable, 3, new CacheValue(EncodeValue(3)), 100, CachePriority::DURABLE)); auto* handle = cache.insert(growing, 2, new CacheValue(EncodeValue(2)), 100); - cache.update_charge(handle, 2048); + cache.update_charge(handle, 700); EXPECT_EQ(cache.lookup(idle, 1), nullptr); - EXPECT_GT(cache.get_usage(), 2048); - // The caller's live handle remains usable even while over capacity. - EXPECT_EQ(DecodeValue( - static_cast(reinterpret_cast(handle)->value)->value), - 2); + EXPECT_LE(cache.get_usage(), 1024); + auto* retained = cache.lookup(durable, 3); + ASSERT_NE(retained, nullptr); + cache.release(retained); cache.release(handle); - EXPECT_EQ(cache.get_usage(), 0); - EXPECT_EQ(cache.lookup(growing, 2), nullptr); + EXPECT_EQ(cache.get_element_count(), 2); +} + +TEST_F(CacheTest, UpdateChargeOversizedEntryPreservesOtherEntries) { + for (bool check_timestamp : {false, true}) { + LRUCache cache(LRUCacheType::SIZE, true); + cache.set_capacity(1024); + cache.set_cache_value_time_extractor([](const void* value) -> int64_t { + return DecodeValue(static_cast(value)->value); + }); + cache.set_cache_value_check_timestamp(check_timestamp); + CacheKey idle("idle"), durable("durable"), growing("growing"); + cache.release(cache.insert(idle, 1, new CacheValue(EncodeValue(1)), 100)); + cache.release(cache.insert(durable, 3, new CacheValue(EncodeValue(3)), 100, + CachePriority::DURABLE)); + const auto usage = cache.get_usage(); + auto* handle = cache.insert(growing, 2, new CacheValue(EncodeValue(2)), 100); + cache.update_charge(handle, 2048); + EXPECT_EQ(cache.get_usage(), usage); + EXPECT_EQ(cache.get_element_count(), 2); + EXPECT_EQ(cache.lookup(growing, 2), nullptr); + // The detached entry remains usable through the caller's live handle. + EXPECT_EQ(DecodeValue(static_cast(reinterpret_cast(handle)->value) + ->value), + 2); + cache.release(handle); + auto* idle_handle = cache.lookup(idle, 1); + ASSERT_NE(idle_handle, nullptr); + cache.release(idle_handle); + auto* durable_handle = cache.lookup(durable, 3); + ASSERT_NE(durable_handle, nullptr); + cache.release(durable_handle); + EXPECT_EQ(cache.get_usage(), usage); + } } TEST_F(CacheTest, UpdateChargeIgnoresReplacedEntry) { @@ -375,13 +412,23 @@ TEST_F(CacheTest, UpdateChargePreservesOtherOutstandingHandles) { LRUCache cache(LRUCacheType::SIZE); cache.set_capacity(1024); CacheKey key("shared"); - auto* first = cache.insert(key, 1, new CacheValue(EncodeValue(1)), 100); + auto* first = cache.insert(key, 1, new CacheValueWithKey(1, EncodeValue(1)), 100); auto* second = cache.lookup(key, 1); ASSERT_NE(second, nullptr); cache.update_charge(first, 2048); + EXPECT_EQ(cache.get_usage(), 0); + EXPECT_EQ(cache.lookup(key, 1), nullptr); cache.release(first); - EXPECT_GT(cache.get_usage(), 2048); + EXPECT_TRUE(_deleted_keys.empty()); + EXPECT_EQ( + DecodeValue(static_cast(reinterpret_cast(second)->value) + ->value), + 1); + cache.update_charge(second, 4096); + EXPECT_EQ(cache.get_usage(), 0); cache.release(second); + ASSERT_EQ(_deleted_keys.size(), 1); + EXPECT_EQ(_deleted_keys.front(), 1); EXPECT_EQ(cache.get_usage(), 0); } diff --git a/be/test/storage/segment/segment_primary_key_cache_test.cpp b/be/test/storage/segment/segment_primary_key_cache_test.cpp index 977036d2c5f85e..7de4aa4625b27f 100644 --- a/be/test/storage/segment/segment_primary_key_cache_test.cpp +++ b/be/test/storage/segment/segment_primary_key_cache_test.cpp @@ -28,7 +28,7 @@ #include "storage/index/primary_key_index.h" #include "storage/segment/segment.h" #include "storage/segment/segment_loader.h" -#include "storage/segment/segment_writer.h" +#include "storage/segment/vertical_segment_writer.h" #include "storage/storage_engine.h" #include "storage/tablet/tablet_schema_helper.h" @@ -36,9 +36,9 @@ namespace doris { TabletSchemaSPtr create_schema(const std::vector& columns, KeysType keys_type); using Generator = std::function; -void build_segment(SegmentWriterOptions opts, TabletSchemaSPtr build_schema, size_t segment_id, - TabletSchemaSPtr query_schema, size_t nrows, Generator generator, - std::shared_ptr* res, std::string segment_dir); +void build_segment(VerticalSegmentWriterOptions opts, TabletSchemaSPtr build_schema, + size_t segment_id, TabletSchemaSPtr query_schema, size_t nrows, + Generator generator, std::shared_ptr* res, std::string segment_dir); class SegmentPrimaryKeyCacheTest : public testing::Test { protected: @@ -53,7 +53,7 @@ class SegmentPrimaryKeyCacheTest : public testing::Test { create_int_value(1, FieldAggregationMethod::OLAP_FIELD_AGGREGATION_REPLACE, false)}, UNIQUE_KEYS); - SegmentWriterOptions opts; + VerticalSegmentWriterOptions opts; opts.enable_unique_key_merge_on_write = true; auto generator = [](size_t rid, int cid, Field& field) { if (cid == 0) { From a24b0f38274b68a9b5946c2902d689a8b3694d80 Mon Sep 17 00:00:00 2001 From: laihui <1353307710@qq.com> Date: Mon, 21 Sep 2026 21:41:12 +0800 Subject: [PATCH 7/7] [fix](cache) Preserve entries when charge grows at the element limit ### What problem does this PR solve? Related PR: #68304 The eviction helpers use count >= limit to reserve a slot for insertion. Reusing them unconditionally for charge growth evicts an unrelated entry when the element limit is reached even if byte capacity is sufficient. Only invoke eviction from update_charge when usage exceeds capacity. ### Release note Avoid unnecessary segment cache eviction when updating entry size at the cache element-count limit. ### Check List (For Author) - Test: Added a BE regression test covering below, exactly at and above byte capacity while at the element-count limit, with both normal and timestamp eviction. Extracted production LRU code and six test bodies passed an ASan/UBSan harness with external dependency stubs; the new test fails on the preceding commit. Changed-file clang-format 16, build hygiene and git diff --check passed. Full BE UT and clang-tidy remain unavailable due to the previously confirmed missing protoc, Boost and Snappy dependencies. - Behavior changed: Yes; charge growth only initiates eviction when usage exceeds capacity, without reserving an additional element slot. - Does this need documentation: No --- be/src/util/lru_cache.cpp | 13 +++++--- be/test/storage/cache/lru_cache_test.cpp | 40 ++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 4 deletions(-) diff --git a/be/src/util/lru_cache.cpp b/be/src/util/lru_cache.cpp index daa3184067ab75..5d44236c9a592e 100644 --- a/be/src/util/lru_cache.cpp +++ b/be/src/util/lru_cache.cpp @@ -358,10 +358,15 @@ void LRUCache::update_charge(Cache::Handle* handle, size_t charge) { _unref(e); _usage -= e->total_size; } - if (_cache_value_check_timestamp) { - _evict_from_lru_with_time(0, &to_remove_head); - } else { - _evict_from_lru(0, &to_remove_head); + // Growing an existing entry does not need another element slot. The + // insertion eviction helpers also check count >= limit, so only call + // them when the updated usage exceeds the byte capacity. + if (_usage > _capacity) { + if (_cache_value_check_timestamp) { + _evict_from_lru_with_time(0, &to_remove_head); + } else { + _evict_from_lru(0, &to_remove_head); + } } } // Cache values can release other caches; never destroy them under the shard lock. diff --git a/be/test/storage/cache/lru_cache_test.cpp b/be/test/storage/cache/lru_cache_test.cpp index 76a45b54e65bf6..5c248838ad65e7 100644 --- a/be/test/storage/cache/lru_cache_test.cpp +++ b/be/test/storage/cache/lru_cache_test.cpp @@ -339,6 +339,46 @@ TEST_F(CacheTest, UpdateChargeEvictsIdleEntriesWhenEntryStillFits) { EXPECT_EQ(cache.get_element_count(), 2); } +TEST_F(CacheTest, UpdateChargeAtElementLimitOnlyEvictsAboveByteCapacity) { + for (bool check_timestamp : {false, true}) { + LRUCache cache(LRUCacheType::SIZE, true); + cache.set_capacity(1024); + cache.set_element_count_capacity(2); + cache.set_cache_value_time_extractor([](const void* value) -> int64_t { + return DecodeValue(static_cast(value)->value); + }); + cache.set_cache_value_check_timestamp(check_timestamp); + CacheKey idle("idle"), growing("growing"); + cache.release(cache.insert(idle, 1, new CacheValue(EncodeValue(1)), 100)); + auto* handle = cache.insert(growing, 2, new CacheValue(EncodeValue(2)), 100); + ASSERT_EQ(cache.get_element_count(), 2); + const auto usage = cache.get_usage(); + + cache.update_charge(handle, 200); + EXPECT_EQ(cache.get_usage(), usage + 100); + EXPECT_EQ(cache.get_element_count(), 2); + auto* retained = cache.lookup(idle, 1); + ASSERT_NE(retained, nullptr); + cache.release(retained); + + // Exactly filling byte capacity must also preserve both entries. + const auto full_charge = 100 + cache.get_capacity() - usage; + cache.update_charge(handle, full_charge); + EXPECT_EQ(cache.get_usage(), cache.get_capacity()); + EXPECT_EQ(cache.get_element_count(), 2); + + // Crossing byte capacity still evicts the idle entry normally. + cache.update_charge(handle, full_charge + 1); + EXPECT_EQ(cache.lookup(idle, 1), nullptr); + EXPECT_EQ(cache.get_element_count(), 1); + EXPECT_LE(cache.get_usage(), cache.get_capacity()); + cache.release(handle); + auto* growing_handle = cache.lookup(growing, 2); + ASSERT_NE(growing_handle, nullptr); + cache.release(growing_handle); + } +} + TEST_F(CacheTest, UpdateChargeOversizedEntryPreservesOtherEntries) { for (bool check_timestamp : {false, true}) { LRUCache cache(LRUCacheType::SIZE, true);