Skip to content
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

Check if the type is an entry while iterating the stream subkeys #2112

Merged
merged 4 commits into from
Feb 23, 2024
Merged

Check if the type is an entry while iterating the stream subkeys #2112

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1f60e4a
fix Create stream consumer group will make the xread command error #2109
jihuayu Feb 22, 2024
bcd3497
Merge branch 'apache:unstable' into f/ok
jihuayu Feb 22, 2024
3d11159
fix test
jihuayu Feb 22, 2024
88f637b
Merge branch 'f/ok' of https://github.com/jihuayu/incubator-kvrocks i…
jihuayu Feb 22, 2024
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
9 changes: 7 additions & 2 deletions src/types/redis_stream.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -250,7 +250,7 @@ StreamConsumerMetadata Stream::decodeStreamConsumerMetadataValue(const std::stri
return consumer_metadata;
}

StreamSubkeyType Stream::identifySubkeyType(const rocksdb::Slice &key) {
StreamSubkeyType Stream::identifySubkeyType(const rocksdb::Slice &key) const {
InternalKey ikey(key, storage_->IsSlotIdEncoded());
Slice subkey = ikey.GetSubKey();
const size_t entry_id_size = sizeof(StreamEntryID);
Expand Down Expand Up @@ -618,7 +618,9 @@ rocksdb::Status Stream::Len(const Slice &stream_name, const StreamLenOptions &op
}

for (; iter->Valid(); options.to_first ? iter->Prev() : iter->Next()) {
*size += 1;
if (identifySubkeyType(iter->key()) == StreamSubkeyType::StreamEntry) {
*size += 1;
}
}

return rocksdb::Status::OK();
Expand Down Expand Up @@ -674,6 +676,9 @@ rocksdb::Status Stream::range(const std::string &ns_key, const StreamMetadata &m

for (; iter->Valid() && (options.reverse ? iter->key().ToString() >= end_key : iter->key().ToString() <= end_key);
options.reverse ? iter->Prev() : iter->Next()) {
if (identifySubkeyType(iter->key()) != StreamSubkeyType::StreamEntry) {
continue;
}
if (options.exclude_start && iter->key().ToString() == start_key) {
continue;
}
Expand Down
2 changes: 1 addition & 1 deletion src/types/redis_stream.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,7 @@ class Stream : public SubKeyScanner {
std::string consumerNameFromInternalKey(rocksdb::Slice key) const;
static std::string encodeStreamConsumerMetadataValue(const StreamConsumerMetadata &consumer_metadata);
static StreamConsumerMetadata decodeStreamConsumerMetadataValue(const std::string &value);
StreamSubkeyType identifySubkeyType(const rocksdb::Slice &key);
StreamSubkeyType identifySubkeyType(const rocksdb::Slice &key) const;
};

} // namespace redis
15 changes: 15 additions & 0 deletions tests/gocase/unit/type/stream/stream_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -985,6 +985,21 @@ func TestStreamOffset(t *testing.T) {
r1 = rdb.XInfoConsumers(ctx, streamName, group2).Val()
require.Equal(t, consumer3, r1[0].Name)
})

t.Run("XREAD After XGroupCreate and XGroupCreateConsumer, for issue #2109", func(t *testing.T) {
streamName := "test-stream"
group := "group"
require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{
Stream: streamName,
ID: "*",
Values: []string{"data1", "b"},
}).Err())
require.NoError(t, rdb.XGroupCreate(ctx, streamName, group, "0").Err())
require.NoError(t, rdb.XGroupCreateConsumer(ctx, streamName, group, "consumer").Err())
require.NoError(t, rdb.XRead(ctx, &redis.XReadArgs{
Streams: []string{streamName, "0"},
}).Err())
})
}

func parseStreamEntryID(id string) (ts int64, seqNum int64) {
Expand Down
Loading