Added support to expire specfic member of a sorted set #1000
Conversation
| @@ -4837,6 +4837,31 @@ | |||
| } | |||
| ] | |||
| }, | |||
| { | |||
There was a problem hiding this comment.
Shouldn't these Garnet-specific commands be in playground/CommandInfoUpdater/GarnetCommandsInfo.json?
There was a problem hiding this comment.
I remember it should be in both otherwise there was an issue with test cases. I need to check once again its been sometime
| @@ -1523,6 +1523,241 @@ Integer reply: the number of members in the resulting sorted set at destination. | |||
|
|
|||
| --- | |||
|
|
|||
| ### ZEXPIRE | |||
There was a problem hiding this comment.
These commands are not in RESP - maybe we need a section on this page for Garnet-specific Z commands.
There was a problem hiding this comment.
I have a plan to reorg the documentation for all commands. Here is my plan
- Create a new page just to show all commands in Garnet (Combine commands from Redis and Garnet specific). It will contain a table with columns
category,command,Is garnet specific,description, andnote. This page will not contain any commands that are yet to be implemented. - Create a separate page for each category. Each data structure is its own page, there won't be any separation of garnet-specific commands, there will be just a field saying if it's a garnet-specific command.
- The Garnet-specific page that is there will be removed and those commands will be added as part of their own category's page.
Let me know if you are find with this change. I think this is need where we add more and more garnet-specific commands . I think we shouldn't stop treating special casing garnet-specific commands and just treat is as a normal command.
There was a problem hiding this comment.
This sounds okay, but we do need special pages to call out these:
- Highlight a list of Garnet-specific commands as existing RESP users will not be familiar with those and can learn about it from one place
- Users of other RESP systems might want to know which RESP commands are NYI in Garnet
There was a problem hiding this comment.
I have added this in Garnet-specific commands doc but I will be reorganizing it as part of a separate PR
|
@TalZaccai @badrishc With this PR, // Code to reproduce the issue
var header = new RespInputHeader(GarnetObjectType.SortedSet) { SortedSetOp = SortedSetOperation.ZCOLLECT };
Assert.AreEqual(SortedSetOperation.ZCOLLECT, header.SortedSetOp ) // Fails Actual value: ZCARD |
Hey @Vijay-Nirmal, thanks for reaching out! |
|
@TalZaccai Thought of the same thing as a temp fix, I will do it as part of a separate PR
So formal ;-) |
|
I am marking it as ready for review but there are 2 pending items before merging this PR
|
|
Here is the benchmark result, I don't see any much difference between main and this PR with no expiring items. I see very minor change in the item but it could be run to run variant. If you guys have any concern then let me know Main as of 046fcd8
This PR as of c236828
|
|
Are we supporting Resp format v2? If not why some of the commands has respProtocolVersion flag? |
We do indeed support RESP3, not fully yet though. |
| @@ -829,5 +829,255 @@ | |||
| "Flags": "RO" | |||
| } | |||
| ] | |||
| }, | |||
| { | |||
| "Command": "ZCOLLECT", | |||
There was a problem hiding this comment.
Please also add docs for these (GarnetCommandsDocs.json)
| @@ -403,7 +403,7 @@ private static unsafe bool TryGetNextSetObjects(byte[] key, SortedSetObject sort | |||
| { | |||
| result = default; | |||
|
|
|||
| if (sortedSetObj.Dictionary.Count == 0) return false; | |||
| if (sortedSetObj.Count() == 0) return false; | |||
There was a problem hiding this comment.
Each of these Count() calls is an O(n) op, and the count can't change since you're locking the key here, so you might want to cache that value. The caller of this method is also calling Count(), so you might also pass that along to the method.
| @@ -132,10 +138,15 @@ public enum SortedSetOrderOperation | |||
| /// <summary> | |||
| /// Sorted Set | |||
| /// </summary> | |||
| public partial class SortedSetObject : GarnetObjectBase | |||
| public unsafe partial class SortedSetObject : GarnetObjectBase | |||
| } | ||
| else | ||
| { | ||
| sortedSetDict.Add(item, score); |
There was a problem hiding this comment.
Can do add an addItem boolean to decide if we should add these (as these adds are duplicated)
| @@ -171,20 +203,41 @@ public SortedSetObject(BinaryReader reader) | |||
| /// <summary> | |||
| /// Copy constructor | |||
| /// </summary> | |||
| public SortedSetObject(SortedSet<(double, byte[])> sortedSet, Dictionary<byte[], double> sortedSetDict, long expiration, long size) | |||
| public SortedSetObject(SortedSet<(double, byte[])> sortedSet, Dictionary<byte[], double> sortedSetDict, Dictionary<byte[], long> expirationTimes, PriorityQueue<byte[], long> expirationQueue, long expiration, long size) | |||
There was a problem hiding this comment.
Why not implement a copy constructor that takes a SortedSetObject instance?
| @@ -223,6 +290,7 @@ public void Add(byte[] item, double score) | |||
| /// </summary> | |||
| public bool Equals(SortedSetObject other) | |||
| { | |||
| // TODO: Implement equals with expiration times | |||
There was a problem hiding this comment.
Is this not in the scope of this PR?
| dict1.Remove(item.Key); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| [MethodImpl(MethodImplOptions.AggressiveInlining)] | ||
| public bool TryGetScore(byte[] key, out double value) |
There was a problem hiding this comment.
Please add XML comments for public methods
|
|
||
| public int Count() | ||
| { | ||
| if (expirationTimes is null) |
| @@ -385,7 +395,8 @@ private void SortedSetIncrement(ref ObjectInput input, ref SpanByteAndMemory out | |||
|
|
|||
| if (sortedSetDict.TryGetValue(member, out var score)) | |||
| { | |||
| sortedSetDict[member] += incrValue; | |||
| score = IsExpired(member) ? 0 : score; | |||
| sortedSetDict[member] = score + incrValue; | |||
| sortedSet.Remove((score, member)); | |||
There was a problem hiding this comment.
You're changing the score if the member is expired, so you might not be removing the member from the sorted set here...
| @@ -641,6 +642,36 @@ private bool NetworkHCOLLECT<TGarnetApi>(ref TGarnetApi storageApi) | |||
| return true; | |||
| } | |||
|
|
|||
| private bool NetworkZCOLLECT<TGarnetApi>(ref TGarnetApi storageApi) | |||
There was a problem hiding this comment.
General question: why is this an admin command?
| var fieldOption = parseState.GetArgSliceByRef(currIdx++); | ||
| if (!fieldOption.ReadOnlySpan.EqualsUpperCaseSpanIgnoringCase(CmdStrings.MEMBERS)) | ||
| { | ||
| return AbortWithErrorMessage(Encoding.ASCII.GetBytes(string.Format(CmdStrings.GenericErrMandatoryMissing, "MEMBERS"))); |
There was a problem hiding this comment.
We should have an AbortWithErrorMessage that takes a (string format, params string[] args)... If you could add that that'd be cool
| @@ -1697,5 +1697,265 @@ private unsafe bool SortedSetBlockingMPop() | |||
|
|
|||
| return true; | |||
| } | |||
|
|
|||
| /// <summary> | |||
| /// Sets an expiration time for a member in the SortedSet stored at key. | |||
There was a problem hiding this comment.
If you could add the command(s) format on the XML comment that'd be super helpful (e.g. ZEXPIRE key seconds [NX | XX | GT | LT] MEMBERS member [member ... ])
Especially when these are not standard RESP commands.
| return AbortWithErrorMessage(Encoding.ASCII.GetBytes(string.Format(CmdStrings.GenericErrMandatoryMissing, "MEMBERS"))); | ||
| } | ||
|
|
||
| if (!parseState.TryGetInt(currIdx++, out var numMembers)) |
There was a problem hiding this comment.
Why do we need numMembers? Can't we just assume that all strings after the MEMBERS keyword are members?
| /// <param name="outputFooter">The output footer object to store the result.</param> | ||
| /// <param name="objectContext">The object context for the operation.</param> | ||
| /// <returns>The status of the operation.</returns> | ||
| public GarnetStatus SortedSetExpire<TObjectContext>(ArgSlice key, long expireAt, bool isMilliseconds, ExpireOption expireOption, ref ObjectInput input, ref GarnetObjectStoreOutput outputFooter, ref TObjectContext objectContext) |
There was a problem hiding this comment.
There should be 2 implementations for each of these methods - 1 for programmatic calls and 1 for network calls. The network call should have the ref ObjectInput input, ref GarnetObjectStoreOutput outputFooter parameters and shouldn't have to make any changes to the ObjectInput / parseState. The programmatic call should not have any of these parameters and should build the ObjectInput & parseState from scratch as well as process the output.
Refer to other methods for examples (e.g. SortedSetPop)
|
|
||
| static void ExecuteSortedSetCollect(ScratchBufferManager scratchBufferManager, StorageSession storageSession) | ||
| { | ||
| var header = new RespInputHeader(GarnetObjectType.SortedSet) { SortedSetOp = SortedSetOperation.ZCOLLECT }; |
There was a problem hiding this comment.
This code (as well as the contents of ExecuteHashCollect) doesn't belong here IMO, there should be a programmatic call to HashCollect / SortedSetCollect in SortedSetOps (see my comment about programmatic calls there).
Adding support for member-specific expiration in a sorted set. This PR adds ZEXPIRE, ZPEXPIRE, ZEXPIREAT, ZPEXPIREAT, ZTTL, ZPTTL, ZEXPIRETIME, ZPEXPIRETIME, ZPERSIST and ZCOLLECT commands to garnet to support this feature.
Since this is a garnet specific command, I used hashset expire API format as the base line. Below is the syntax for each command