Skip to content

fix(stream): bound range reads by metadata entries #3595

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wengsht wants to merge 1 commit into
base: unstable
Choose a base branch
from
+112 −14
Draft

fix(stream): bound range reads by metadata entries #3595

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
61 changes: 47 additions & 14 deletions src/types/redis_stream.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1151,11 +1151,47 @@ rocksdb::Status Stream::Len(engine::Context &ctx, const Slice &stream_name, cons

rocksdb::Status Stream::range(engine::Context &ctx, const std::string &ns_key, const StreamMetadata &metadata,
const StreamRangeOptions &options, std::vector<StreamEntry> *entries) const {
std::string start_key = internalKeyFromEntryID(ns_key, metadata, options.start);
std::string end_key = internalKeyFromEntryID(ns_key, metadata, options.end);
if (metadata.size == 0) {
return rocksdb::Status::OK();
}

if ((!options.reverse && options.end < options.start) || (options.reverse && options.start < options.end)) {
return rocksdb::Status::OK();
}

// Metadata is the logical stream frontier; raw keys outside it may remain after an interrupted migration.
StreamRangeOptions range_options = options;
if (options.reverse) {
if (options.start < metadata.first_entry_id || options.end > metadata.last_entry_id) {
return rocksdb::Status::OK();
}
if (options.start > metadata.last_entry_id) {
range_options.start = metadata.last_entry_id;
range_options.exclude_start = false;
}
if (options.end < metadata.first_entry_id) {
range_options.end = metadata.first_entry_id;
range_options.exclude_end = false;
}
} else {
if (options.end < metadata.first_entry_id || options.start > metadata.last_entry_id) {
return rocksdb::Status::OK();
}
if (options.start < metadata.first_entry_id) {
range_options.start = metadata.first_entry_id;
range_options.exclude_start = false;
}
if (options.end > metadata.last_entry_id) {
range_options.end = metadata.last_entry_id;
range_options.exclude_end = false;
}
}

std::string start_key = internalKeyFromEntryID(ns_key, metadata, range_options.start);
std::string end_key = internalKeyFromEntryID(ns_key, metadata, range_options.end);

if (start_key == end_key) {
if (options.exclude_start || options.exclude_end) {
if (range_options.exclude_start || range_options.exclude_end) {
return rocksdb::Status::OK();
}

Expand All @@ -1171,11 +1207,7 @@ rocksdb::Status Stream::range(engine::Context &ctx, const std::string &ns_key, c
return rocksdb::Status::InvalidArgument(rv.Msg());
}

entries->emplace_back(options.start.ToString(), std::move(values));
return rocksdb::Status::OK();
}

if ((!options.reverse && options.end < options.start) || (options.reverse && options.start < options.end)) {
entries->emplace_back(range_options.start.ToString(), std::move(values));
return rocksdb::Status::OK();
}

Expand All @@ -1191,20 +1223,21 @@ rocksdb::Status Stream::range(engine::Context &ctx, const std::string &ns_key, c

auto iter = util::UniqueIterator(ctx, read_options, stream_cf_handle_);
iter->Seek(start_key);
if (options.reverse && (!iter->Valid() || iter->key().ToString() != start_key)) {
if (range_options.reverse && (!iter->Valid() || iter->key().ToString() != start_key)) {
iter->SeekForPrev(start_key);
}

for (; iter->Valid() && (options.reverse ? iter->key().ToString() >= end_key : iter->key().ToString() <= end_key);
options.reverse ? iter->Prev() : iter->Next()) {
for (;
iter->Valid() && (range_options.reverse ? iter->key().ToString() >= end_key : iter->key().ToString() <= end_key);
range_options.reverse ? iter->Prev() : iter->Next()) {
if (identifySubkeyType(iter->key()) != StreamSubkeyType::StreamEntry) {
continue;
}
if (options.exclude_start && iter->key().ToString() == start_key) {
if (range_options.exclude_start && iter->key().ToString() == start_key) {
continue;
}

if (options.exclude_end && iter->key().ToString() == end_key) {
if (range_options.exclude_end && iter->key().ToString() == end_key) {
break;
}

Expand All @@ -1216,7 +1249,7 @@ rocksdb::Status Stream::range(engine::Context &ctx, const std::string &ns_key, c

entries->emplace_back(entryIDFromInternalKey(iter->key()).ToString(), std::move(values));

if (options.with_count && entries->size() == options.count) {
if (range_options.with_count && entries->size() == range_options.count) {
break;
}
}
Expand Down
65 changes: 65 additions & 0 deletions tests/cppunit/types/stream_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,22 @@ class RedisStreamTest : public TestBase { // NOLINT
void SetUp() override { auto s = stream_->Del(*ctx_, name_); }
void TearDown() override { auto s = stream_->Del(*ctx_, name_); }

rocksdb::Status PutRawEntry(const redis::StreamEntryID &id, const std::vector<std::string> &values) {
std::string ns_key = stream_->AppendNamespacePrefix(name_);
StreamMetadata metadata(false);
auto s = stream_->GetMetadata(*ctx_, ns_key, &metadata);
if (!s.ok()) return s;

std::string sub_key;
PutFixed64(&sub_key, id.ms);
PutFixed64(&sub_key, id.seq);
std::string entry_key = InternalKey(ns_key, sub_key, metadata.version, storage_->IsSlotIdEncoded()).Encode();
auto batch = storage_->GetWriteBatchBase();
s = batch->Put(storage_->GetCFHandle(ColumnFamilyID::Stream), entry_key, redis::EncodeStreamEntryValue(values));
if (!s.ok()) return s;
return storage_->Write(*ctx_, storage_->DefaultWriteOptions(), batch->GetWriteBatch());
}

std::string name_;
redis::Stream *stream_;
};
Expand Down Expand Up @@ -649,6 +665,55 @@ TEST_F(RedisStreamTest, RangeFromMinimumToMaximum) {
CheckStreamEntryValues(entries[3].values, values4);
}

TEST_F(RedisStreamTest, RangeIgnoresEntriesOutsideMetadataBounds) {
redis::StreamAddOptions add_options;
redis::StreamEntryID id;
for (const char *entry_id : {"1-0", "2-0", "3-0"}) {
add_options.next_id_strategy = *ParseNextStreamEntryIDStrategy(entry_id);
auto s = stream_->Add(*ctx_, name_, add_options, {"key", entry_id}, &id);
ASSERT_TRUE(s.ok());
}

redis::StreamTrimOptions trim_options;
trim_options.strategy = redis::StreamTrimStrategy::MinID;
trim_options.min_id = redis::StreamEntryID{2, 0};
uint64_t trimmed = 0;
auto s = stream_->Trim(*ctx_, name_, trim_options, &trimmed);
ASSERT_TRUE(s.ok());
ASSERT_EQ(trimmed, 1);

ASSERT_TRUE(PutRawEntry({1, 0}, {"stale", "before"}).ok());
ASSERT_TRUE(PutRawEntry({4, 0}, {"stale", "after"}).ok());

redis::StreamRangeOptions range_options;
range_options.start = redis::StreamEntryID::Minimum();
range_options.end = redis::StreamEntryID::Maximum();
std::vector<redis::StreamEntry> entries;
s = stream_->Range(*ctx_, name_, range_options, &entries);
ASSERT_TRUE(s.ok());
ASSERT_EQ(entries.size(), 2);
EXPECT_EQ(entries[0].key, "2-0");
EXPECT_EQ(entries[1].key, "3-0");

range_options.reverse = true;
range_options.start = redis::StreamEntryID::Maximum();
range_options.end = redis::StreamEntryID::Minimum();
entries.clear();
s = stream_->Range(*ctx_, name_, range_options, &entries);
ASSERT_TRUE(s.ok());
ASSERT_EQ(entries.size(), 2);
EXPECT_EQ(entries[0].key, "3-0");
EXPECT_EQ(entries[1].key, "2-0");

range_options.reverse = false;
range_options.start = {1, 0};
range_options.end = {1, 0};
entries.clear();
s = stream_->Range(*ctx_, name_, range_options, &entries);
ASSERT_TRUE(s.ok());
EXPECT_TRUE(entries.empty());
}

TEST_F(RedisStreamTest, RangeFromMinimumToMinimum) {
redis::StreamAddOptions add_options;
add_options.next_id_strategy = *ParseNextStreamEntryIDStrategy(fmt::format("{}-{}", 123456, 1));
Expand Down