Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 30 additions & 18 deletions src/storage/engine/grouped/ordered_io.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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<LoadedOrderedGroup>(
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<LoadedOrderedGroup> {
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<LoadedOrderedGroup> {
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,
Expand All @@ -177,6 +166,29 @@ StorageEngine::Impl::LoadOrderedGroupSnapshot(
limiter);
}

Task<absl::StatusOr<LoadedOrderedGroup>>
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<LoadedOrderedGroup>(
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<LoadedOrderedGroup> {
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<absl::StatusOr<ListResult>> StorageEngine::Impl::LoadOrderedListRange(
WorkerStore& store, WorkerStore::PartitionStore& partition,
std::uint8_t db_id, std::string_view key, const Digest& digest,
Expand Down
18 changes: 13 additions & 5 deletions src/storage/engine/grouped/zset.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1251,10 +1251,15 @@ StorageEngine::Impl::ExecuteGroupedSortedSetLocked(
else if constexpr (std::is_same_v<Page, PackedSortedSetValues>)
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();
Expand Down Expand Up @@ -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_));
}
Expand Down
14 changes: 10 additions & 4 deletions src/storage/engine/impl.h
Original file line number Diff line number Diff line change
Expand Up @@ -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<absl::StatusOr<LoadedOrderedGroup>> 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<absl::StatusOr<LoadedOrderedGroup>> 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<absl::StatusOr<LoadedSortedSetPage>> LoadSortedSetPage(
WorkerStore& store, WorkerStore::PartitionStore& partition,
std::uint8_t db_id, std::string_view key, const Digest& digest,
Expand Down