From 9a3d375ca010b1c8f24d3dcdc96ba4f0f8d4572e Mon Sep 17 00:00:00 2001 From: Jerry Zhao Date: Mon, 19 Oct 2026 21:52:41 +0800 Subject: [PATCH] refactor: separate encoded ZSet page loading and update Bycorf --- bycorf | 2 +- src/storage/engine/grouped/ordered_io.cpp | 48 ++++++++++++++--------- src/storage/engine/grouped/zset.cpp | 18 ++++++--- src/storage/engine/impl.h | 14 +++++-- 4 files changed, 54 insertions(+), 28 deletions(-) diff --git a/bycorf b/bycorf index 587a9e776..08fa44dad 160000 --- a/bycorf +++ b/bycorf @@ -1 +1 @@ -Subproject commit 587a9e776f9f64729456d6772596d53597f4a983 +Subproject commit 08fa44dad5c695f6b2bfedc47e37708c1eeaffae diff --git a/src/storage/engine/grouped/ordered_io.cpp b/src/storage/engine/grouped/ordered_io.cpp index e4edf6f03..95ce92179 100644 --- a/src/storage/engine/grouped/ordered_io.cpp +++ b/src/storage/engine/grouped/ordered_io.cpp @@ -147,28 +147,17 @@ StorageEngine::Impl::LoadOrderedGroupSnapshot( WorkerStore& store, WorkerStore::PartitionStore& partition, std::uint8_t db_id, std::string_view key, const Digest& digest, GroupedObject::Handle object, std::uint64_t page_id, bool pinned, - GroupReadLimiter* limiter, SortedSetMemberValidation validation, - bool encoded_sorted_set) { + GroupReadLimiter* limiter, SortedSetMemberValidation validation) { return LoadOrderedGroup( store, partition, db_id, key, digest, std::move(object), page_id, pinned, - [validation, encoded_sorted_set]( - std::string_view payload, std::uint64_t sequence, - const OrderedGroupEntry& route, - LoadedValue&&) -> absl::StatusOr { - auto decoded = encoded_sorted_set - ? DecodeEncodedSortedSetGroup(payload) - : DecodeOrderedGroup(payload, validation); + [validation](std::string_view payload, std::uint64_t sequence, + const OrderedGroupEntry& route, + LoadedValue&&) -> absl::StatusOr { + auto decoded = DecodeOrderedGroup(payload, validation); if (!decoded.ok()) return decoded.status(); if (decoded->kind_ == OrderedCollectionKind::kSortedSet) { - const auto min_score = - decoded->encoded_sorted_set_ - ? decoded->encoded_sorted_set_->entries().front().score_ - : decoded->entries_.front().score_; - const auto max_score = - decoded->encoded_sorted_set_ - ? decoded->encoded_sorted_set_->entries().back().score_ - : decoded->entries_.back().score_; - if (min_score != route.min_score_ || max_score != route.max_score_) + if (decoded->entries_.front().score_ != route.min_score_ || + decoded->entries_.back().score_ != route.max_score_) return absl::DataLossError("ordered page score bounds mismatch"); } return LoadedOrderedGroup{.sequence_ = sequence, @@ -177,6 +166,29 @@ StorageEngine::Impl::LoadOrderedGroupSnapshot( limiter); } +Task> +StorageEngine::Impl::LoadEncodedSortedSetGroupSnapshot( + WorkerStore& store, WorkerStore::PartitionStore& partition, + std::uint8_t db_id, std::string_view key, const Digest& digest, + GroupedObject::Handle object, std::uint64_t page_id, + GroupReadLimiter* limiter) { + return LoadOrderedGroup( + store, partition, db_id, key, digest, std::move(object), page_id, false, + [](std::string_view payload, std::uint64_t sequence, + const OrderedGroupEntry& route, + LoadedValue&&) -> absl::StatusOr { + auto decoded = DecodeEncodedSortedSetGroup(payload); + if (!decoded.ok()) return decoded.status(); + const auto entries = decoded->encoded_sorted_set_->entries(); + if (entries.front().score_ != route.min_score_ || + entries.back().score_ != route.max_score_) + return absl::DataLossError("ordered page score bounds mismatch"); + return LoadedOrderedGroup{.sequence_ = sequence, + .snapshot_ = std::move(*decoded)}; + }, + limiter); +} + Task> StorageEngine::Impl::LoadOrderedListRange( WorkerStore& store, WorkerStore::PartitionStore& partition, std::uint8_t db_id, std::string_view key, const Digest& digest, diff --git a/src/storage/engine/grouped/zset.cpp b/src/storage/engine/grouped/zset.cpp index 860cd3cef..d80e644c7 100644 --- a/src/storage/engine/grouped/zset.cpp +++ b/src/storage/engine/grouped/zset.cpp @@ -1251,10 +1251,15 @@ StorageEngine::Impl::ExecuteGroupedSortedSetLocked( else if constexpr (std::is_same_v) return LoadPackedSortedSetPage(store, partition, db_id, key, digest, object, metadata[i].id_, limiter); - else + else { + if (point_write) + return LoadEncodedSortedSetGroupSnapshot(store, partition, db_id, key, + digest, object, + metadata[i].id_, limiter); return LoadOrderedGroupSnapshot(store, partition, db_id, key, digest, object, metadata[i].id_, false, limiter, - validation, point_write); + validation); + } }; auto page = co_await load(); if (!page.ok()) co_return page.status(); @@ -2107,9 +2112,12 @@ StorageEngine::Impl::ExecuteGroupedSortedSetLocked( co_return absl::CancelledError( "Sorted Set plan cancelled by shutdown"); } - auto page = co_await LoadOrderedGroupSnapshot( - store, partition, db_id, key, digest, object, metadata[i].id_, false, - nullptr, SortedSetMemberValidation::kVerifyUnique, point_write); + auto page = co_await (point_write ? LoadEncodedSortedSetGroupSnapshot( + store, partition, db_id, key, + digest, object, metadata[i].id_) + : LoadOrderedGroupSnapshot( + store, partition, db_id, key, + digest, object, metadata[i].id_)); if (!page.ok()) co_return page.status(); loaded.emplace(i, std::move(page->snapshot_)); } diff --git a/src/storage/engine/impl.h b/src/storage/engine/impl.h index 792ca088b..90b5c7ed8 100644 --- a/src/storage/engine/impl.h +++ b/src/storage/engine/impl.h @@ -4440,16 +4440,22 @@ class StorageEngine::Impl { // caller-owned limiter accounts against the refreshed physical owner until // decoding finishes; it must outlive this task, including cancellation. // Only read-only consumers may trust stored Sorted Set member uniqueness. - // Point writers may retain a checked encoded block instead of member strings; - // that representation always checks uniqueness and also releases the lease. Task> LoadOrderedGroupSnapshot( WorkerStore& store, WorkerStore::PartitionStore& partition, std::uint8_t db_id, std::string_view key, const Digest& digest, GroupedObject::Handle object, std::uint64_t id, bool pinned = false, GroupReadLimiter* limiter = nullptr, SortedSetMemberValidation validation = - SortedSetMemberValidation::kVerifyUnique, - bool encoded_sorted_set = false); + SortedSetMemberValidation::kVerifyUnique); + // Returns a fully validated encoded_sorted_set_ with entries_ empty, after + // releasing the physical lease. The snapshot and any copies remain on the + // key owner worker; a supplied limiter must outlive the task, including + // cancellation. + Task> LoadEncodedSortedSetGroupSnapshot( + WorkerStore& store, WorkerStore::PartitionStore& partition, + std::uint8_t db_id, std::string_view key, const Digest& digest, + GroupedObject::Handle object, std::uint64_t id, + GroupReadLimiter* limiter = nullptr); Task> LoadSortedSetPage( WorkerStore& store, WorkerStore::PartitionStore& partition, std::uint8_t db_id, std::string_view key, const Digest& digest,