Talks about the design of consumer group in stream

[Yangsx-1]

Discussion about issue #1586

I'm now working on adding consumer group in stream.
The relevant structures are listed below, referring to redis:

struct StreamConsumer {
  time_t seen_time;                                   // Last time this consumer was active.
  std::string name;                                   // Consumer name.
  std::unordered_map<StreamEntryID, StreamNACK> pel;  // Consumer specific pending entries list
};

struct StreamNACK {
  time_t delivery_time;      // Last time this message was delivered.
  uint64_t delivery_count;   // Number of times this message was delivered.
  StreamConsumer *consumer;  // The consumer this message was delivered to in the last delivery.
};

struct StreamCG {
  StreamEntryID last_id;  // Last delivered (not acknowledged) ID for this group.
  int64_t entries_read;
  std::unordered_map<StreamEntryID, StreamNACK> pel;
  std::vector<StreamConsumer> consumers;
};

Here are some questions:
1.Is it proper to add StreamCG in StreamMetadata?
2.If it's ok to add it to StreamMetadata, how can i encode the metadata? Is there any compatibility issues in encoding complex data structure?
3.If not proper, i can try to store consumer group separately. I'm not sure if there's an encoding issue here.

[PragmaTwice]

I think you can add them to StreamMetadata, but to maintain the data compatiblity, you need to check the length of stream metadata, e.g.

StreamMetadata::Decode() {
  ... // original decode process
  if (length > original_len) {
    group_enable = true;
    DecodeRestPart();
  } else {
    group_enable = false;
  }
}

So that the original user data of Stream type (without these new fields) will be read and written normally.

[Yangsx-1]

Is it ok to encode vector and unordered_map in metadata? Their size is variable.

[git-hulk]

I'm uncertain about how much space is going to be used by consumer groups, it may bring a lot of read/write amplification if it's too large. Putting them together with the stream metadata would be better if they are small enough(for example < 100bytes). Otherwise, it'd be better to store them in a separate key, so that it only affects the consume operation. To see if you guys have any other comments. @torwig @PragmaTwice

[PragmaTwice]

Agree with @git-hulk. If it is small, less-frequent to access and has less elements, I think you can encode it directly on the metadata. Otherwise you need multiple key-value to store them and fetch them via rocksdb interface.

[torwig]

I was thinking about the implementation some time ago. I'll write my opinion later. However, I think that the read/write patterns of the information about consumer groups should also be taken into account.

[torwig]

@Yangsx-1 vector/unordered_map/other_container is just a C++ representation. I'm not sure you will be operating with exactly those containers. I think the more important thing is the choice of how to store consumer groups, consumers and related data in RocksDB (since we will not keep all the data in memory like Redis; we will read/write a sufficient amount of data when executing commands).

Since each consumer group (CG) has its metadata/state and even every consumer has its metadata/state and most commands that operate on CG are changing the state of CG, I tend to store the CG's state separately from the stream's metadata.
The XINFO STREAM command has a response that includes the number of CGs. For this case, we can have the counter of the stream's CGs as a field of the stream's metadata and update it on creating/destroying a CG (luckily, these commands will be relatively rare).

The PEL (pending entry list) of each CG should be stored in a way that gives us the ability to quickly:

• remove entry from the PEL on XACK
• change the owner of the entry
• track each entry's last delivery time and the number of deliveries.

One question is whether we need to lock the stream when executing write operations on one of its CGs?
I also was wondering whether we need a separate column family for CGs?
And how to delete CGs, consumers and related data on the stream deletion?

[Yangsx-1]

I'll try to implement CGs in a separate way. And the answers of these questions may be clear when i do these tasks.

[Yangsx-1]

Here are my designs for storing consumer group, consumer and PEL.
The consumer group and PEL should be stored separately.

For consumer group:
Key:
stream name + consumer group name(something like streamname_CGname, e.g., using underscore to combine the two name as a single key)
Metadata:
StreamEntryID last_id; // Last delivered (not acknowledged) ID for this group.
int64_t entries_read = 0;
Subkey:
consumer name
Value:
seen_time /* Last time this consumer was active. */

For PEL:
Key:
stream name + consumer group name + StreamEntryID
Metadata:
delivery_time; /* Last time this message was delivered. */
delivery_count; /* Number of times this message was delivered.*/
Subkey:
consumer name

Is this design ok? @git-hulk @PragmaTwice @torwig

[Yangsx-1]

You mean it's not ok to combine the two name together, because it may conflict with the key just the same as the combination?

[git-hulk]

That's right, maybe we can put the consumer group inside the stream column family as well. What do you think? @torwig @PragmaTwice

[torwig]

@git-hulk My first thought is: even if we move consumer groups to the stream column family, we still can face the same key name conflict. Maybe there is a way to avoid it.

[git-hulk]

Is it possible to avoid conflicts by limiting the consumer group name symbol, like only alpha, and some special chars are allowed?

[PragmaTwice]

I am not sure about the detail of key encoding: is consumer group name and fields behind it encoded as subkey (refer to https://kvrocks.apache.org/community/data-structure-on-rocksdb#user-key-encoding)?

[Yangsx-1]

I am not sure about the detail of key encoding: is consumer group name and fields behind it encoded as subkey (refer to https://kvrocks.apache.org/community/data-structure-on-rocksdb#user-key-encoding)?

We can do like that, but the subkey maybe a little long.

If we store them together, the subkeys and values may be as follows:

1. CGname + metadata --> last_id, entries_read
2. CGname + consumer name --> seen_time(a variable of uint64_t)
3. CGname + StreamEntryID + metadata --> delivery_time, delivery_count
4. CGname + StreamEntryID + consumer --> consumer name

May be 3 and 4 can be encoded together, like CGname + StreamEntryID --> delivery_time, delivery_count, consumer name

Is this ok? @git-hulk @PragmaTwice @torwig

[torwig]

Since the entry ID is a subkey for a full stream record key (stream_name + entry_id) and if we disallow starting a consumer group's name and a consumer's name with a number then maybe it will allow us to distinguish a key of a regular stream element from other keys related to CG implementation.

I think that all the keys related to a CG should start with a stream name because, in the case of a stream deletion, we need to delete all his CGs with their consumers and PEL.

Generally, I think that the provided approach may work if each case is clearly distinct from others.

1. Stream_Name + CG_Name + Some_Way_Of_Metadata_Definition -> CG's Metadata.
2. Stream_Name + CG_Name + Consumer_Name [ + Some_Way_Of-Metadata_Definition ] -> Consumer's Metadata (more than just last_seen timestamp, see XINFO CONSUMERS).
3. PEL entries: Stream_Name + CG_Name + Stream_Entry_ID -> (owner, delivery time, delivery count). Perhaps, we need to implement it differently to efficiently change the owner of delivered entries with the XCLAIM command (we need to answer the query "Give me elements from PEL, delivered to consumer X which satisfy some conditions related to idle time and delivery count ).

[Yangsx-1]

I agree with you @torwig .
So when encoding the subkey, is it better to add a length field before each name? such as :
sizeof(Stream_Name) + Stream_Name + sizeof(CG_Name) + CG_Name + sizeof(Consumer name) + consumer name + others

[torwig]

@Yangsx-1 Yes, I believe it's the way we do right now when encoding metadata and namespace (which is a string value) inside it.

[Yangsx-1]

Here is a question for the implementation of pending list.

The PEL entries need to be created, deleted and claimed quickly.

So a good way is to use Stream_Name + CG_Name + Stream_Entry_ID -> (owner, delivery time, delivery count) to store the information, which is discussed above.

However when i implement XGROUP DELCONSUMER, the PEL of the deleted consumer will also be deleted, so this implementation of PEL makes it difficult to delete all pending entries belongs to the deleted consumer. We need to scan the whole key space and find out the owner same as the deleted consumer which may be expensive because of frequent decoding the value.

I'm not sure if it's ok, is there any idea? Maybe DELCONSUMER is not a frequent operation and we can bear the cost. @git-hulk @PragmaTwice @torwig

[git-hulk]

So a good way is to use Stream_Name + CG_Name + Stream_Entry_ID -> (owner, delivery time, delivery count) to store the information, which is discussed above.

Looks good to me.

I'm not sure if it's ok, is there any idea?

I'm fine about this since I have no better idea.

[uds5501]

Hi good engineers, @Yangsx-1 @git-hulk @torwig I tried to make sense of this discussion but following it and reading the encoding article here, can someone explain / point me to a resource which could elaborate on

1. what exactly is key metadata and sub-key ?
2. how is sub-key different from key?

Thank you

[Yangsx-1]

I think you can refer to patch #1704 as an example, it's a little hard to fully explain it. Here are my understanding:
Actually in my opinion, the way they store are almost the same, they are all key-value pairs in rocksdb, but the way they encoding are very different.
What's more, in the implementation of stream, the key's metadata and subkey-value are stored separately, key's metadata is stored in metadata column family and its subkey-value are stored in stream column family.
Maybe you can see the code of Hash or other to understand it.
See if others have some supplement and correct me.

[PragmaTwice]

what exactly is key metadata and sub-key ?

Metadata is just how the datatype (especially properties of the datatype) is encoded in the rocksdb key-value interface.
so it is about reduction or encoding.

Usually metadata refers to how the value (of key-value) is encoded.

For encoding a key, like a user key x, kvrocks add more information to the key for some kvrocks features, like x -> <namespace> <key length: 1> x.

And if x is typed hash, the key <namespace> <key length: 1> x is mapped to the value (the metadata) <hash flag> <expire time> <version> <size>.

Then if we add some elements to the hash x, some more key-value will be added. For example if we add a -> 1 to hash x, the underlying rocksdb key-value that need to be added is <namespace> <key length: 1> x <version> <subkey: a> -> 1.

So the subkey is just to distinguish with the user key. And then when we delete a key x, all of keys prefixed <namespace> <key length: 1> x will be deleted, including all its subkeys. If we do not have subkeys, the deletion operation will be hard.

[uds5501]

okay that makes so much sense, thank you! @PragmaTwice
Also, I will take a look at the implementation @Yangsx-1 .