feat:Stream base commands

[KKorpse]

issue：#1584
实现文档：#1717

PR 概要

Redis 中的 stream 分为两部分指令，分别对应两个单测文件：

1. stream.tcl：基本指令集
2. stream-cgroups.tcl：消费组指令集

第一部分完全独立于第二部分的指令，拥有消息队列增删改查等基本功能，第二部分是第一部分基础上的更高级功能。

鉴于工作量巨大，我建议先把第一部分的指令，以及基础框架合并，第二部分中的指令作为单独的 issue 分批次补充（或者直接在迁移到 4.0 后补充）。

此 pr 为第一部分功能的实现，包含了第一部分的所有指令，以及第二部分所依赖的元数据，工具函数等。（代码是从原始pr中抽离出来的，原始完整 pr 见：#1673）
测试：单测使用的是 go 语言，内容在 stream.go 中，依附于 integration 已有的测试框架。其中的所有测试样例都是从 redis 官方的 stream.tcl 中翻译而来，不支持的单测在后续有解释。

该 PR 支持的指令：Stream 基本指令集

• XADD
• XDEL
• XTRIM
• XRANGE
• XREVRANGE
• XREAD
• XLEN
• XINFO
• DEL：Stream 的删除

暂不支持的指令：Stream 消费组指令集

（实现代码未在当前 pr 中）

暂不支持的单测及理由：

AOF文件相关

Pika 无 AOF，无需支持

XADD，XTRIM： “~” 以及 “limit” 参数

~ 和 limit 是 redis 用于模糊裁剪以提高裁剪性能的参数。redis stream 的消息是存储在 radix tree 中，每个树节指向了一个 Listpack，内可容纳多条消息。而模糊裁剪是为了性能，以树节点为单位进行裁剪。

Pika 每条消息都是以独立 kv 对存储的，不需要模糊裁剪功能。对于 ~ 参数，pika 当作 = 参数使用。对于 limit 参数，pika 会返回参数错误。

XSETID

管理员指令，无实用意义，优先级在 xgroup 之后。

XREAD：block

block 参数功能和 blpop，brpop 类似，可以仿照其逻辑。但考虑到实现与 Pika 的线程模型有关，建议迁移到 Pika 4.0 以后再实现。

后续流程建议

1. 迁移到 4.0
2. 实现 XREAD block 功能
3. XGROUP CREATE，XREADGROUP
4. XPENDING，XACK
5. XCLAIM
6. 剩余简单指令

[wanghenshui]

你这个实现有没有可能hash key和xstream key冲突？

[KKorpse]

我突然发现你这个没log复制啊

是指 Cmd::DoBinlog 么，好像非阻塞的指令直接用基类的实现就可以了，不需要重载？
我看执行命令的时候会自动调 DoBinLog

测过stream的主从复制么

简单测了一下，OK 的。 主节点写，从节点查，数据都在。

pika_instance_ip = '127.0.0.1'
pika_instance_port = '9221'
pika_slave_ip = '127.0.0.1'
pika_slave_port = '9231'

def some_test(db_):
    master = redis.Redis(host=pika_instance_ip, port=int(pika_instance_port), db=db_)
    slave = redis.Redis(host=pika_slave_ip, port=int(pika_slave_port), db=db_)
    slave.slaveof(pika_instance_ip, pika_instance_port)

    time.sleep(2)

    stream_name = 'mystream'

    # 在 master 中写入 20 条消息
    for i in range(20):
        message_data = {'key': f'value{i}'}
        master.xadd(stream_name, message_data)

    # 等待一秒，确保数据同步到 slave
    time.sleep(1)

    # 在 slave 中读取所有消息
    messages = slave.xrange(stream_name, min='-', max='+')
    assert(len(messages) == 20)
    for message in messages:
        message_id, message_data = message
        print(f"Message ID: {message_id}, Message Data: {message_data}")

    master.flushdb()

some_test(0)

改成 Go 测试代码，放到 Go 测试里面？

好的，我加到 replication_test 里面

[KKorpse]

你这个实现有没有可能hash key和xstream key冲突？

有两个前缀会冲突，我在 hash 中禁止掉了这两个前缀

  // pika_hash.cc
  // check the conflict of stream used prefix, see details in defination of STREAM_TREE_PREFIX
  if (key_.compare(0, STERAM_TREE_PREFIX.size(), STERAM_TREE_PREFIX) == 0 ||
      key_.compare(0, STREAM_DATA_HASH_PREFIX.size(), STREAM_DATA_HASH_PREFIX) == 0) {
    res_.SetRes(CmdRes::kErrOther, "hash key can't start with " + STERAM_TREE_PREFIX + " or " + STREAM_META_HASH_KEY);
    return;
  }

//pika_stream_base.h
// each abstracted tree has a prefix,
// prefix + treeID will be the key of the hash to store the tree,
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STERAM_TREE_PREFIX = "STR_TREE";

// key of the hash to store stream meta,
// notice: we need to ban the use of this key when using "HSET".
static const std::string STREAM_META_HASH_KEY = "STREAM";

// each stream's data (messages) are stored in a hash,
// to avoid stream key confilict with hash key, we add a prefix to each key for stream data,
// it's ok to use the same string as STREAM_META_HASH_KEY to be the prefix.
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STREAM_DATA_HASH_PREFIX = STREAM_META_HASH_KEY;

[wanghenshui]

你这个实现有没有可能hash key和xstream key冲突？

有两个前缀会冲突，我在 hash 中禁止掉了这两个前缀

  // pika_hash.cc
  // check the conflict of stream used prefix, see details in defination of STREAM_TREE_PREFIX
  if (key_.compare(0, STERAM_TREE_PREFIX.size(), STERAM_TREE_PREFIX) == 0 ||
      key_.compare(0, STREAM_DATA_HASH_PREFIX.size(), STREAM_DATA_HASH_PREFIX) == 0) {
    res_.SetRes(CmdRes::kErrOther, "hash key can't start with " + STERAM_TREE_PREFIX + " or " + STREAM_META_HASH_KEY);
    return;
  }

//pika_stream_base.h
// each abstracted tree has a prefix,
// prefix + treeID will be the key of the hash to store the tree,
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STERAM_TREE_PREFIX = "STR_TREE";

// key of the hash to store stream meta,
// notice: we need to ban the use of this key when using "HSET".
static const std::string STREAM_META_HASH_KEY = "STREAM";

// each stream's data (messages) are stored in a hash,
// to avoid stream key confilict with hash key, we add a prefix to each key for stream data,
// it's ok to use the same string as STREAM_META_HASH_KEY to be the prefix.
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STREAM_DATA_HASH_PREFIX = STREAM_META_HASH_KEY;

这样搞也行吧。为什么当初不选择单独整个db类型来支持stream

[lqxhub]

你这个实现有没有可能hash key和xstream key冲突？

有两个前缀会冲突，我在 hash 中禁止掉了这两个前缀

  // pika_hash.cc
  // check the conflict of stream used prefix, see details in defination of STREAM_TREE_PREFIX
  if (key_.compare(0, STERAM_TREE_PREFIX.size(), STERAM_TREE_PREFIX) == 0 ||
      key_.compare(0, STREAM_DATA_HASH_PREFIX.size(), STREAM_DATA_HASH_PREFIX) == 0) {
    res_.SetRes(CmdRes::kErrOther, "hash key can't start with " + STERAM_TREE_PREFIX + " or " + STREAM_META_HASH_KEY);
    return;
  }

//pika_stream_base.h
// each abstracted tree has a prefix,
// prefix + treeID will be the key of the hash to store the tree,
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STERAM_TREE_PREFIX = "STR_TREE";

// key of the hash to store stream meta,
// notice: we need to ban the use of this key when using "HSET".
static const std::string STREAM_META_HASH_KEY = "STREAM";

// each stream's data (messages) are stored in a hash,
// to avoid stream key confilict with hash key, we add a prefix to each key for stream data,
// it's ok to use the same string as STREAM_META_HASH_KEY to be the prefix.
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STREAM_DATA_HASH_PREFIX = STREAM_META_HASH_KEY;

这样搞也行吧。为什么当初不选择单独整个db类型来支持stream

那有个疑问, stream的数据会不会 也能被 hash操作了, 比如通过 hash相关的命令 增删改查, 还有 在scan的时候 会不会也把 Steam的数据也 拿到了

[KKorpse]

你这个实现有没有可能hash key和xstream key冲突？

有两个前缀会冲突，我在 hash 中禁止掉了这两个前缀

  // pika_hash.cc
  // check the conflict of stream used prefix, see details in defination of STREAM_TREE_PREFIX
  if (key_.compare(0, STERAM_TREE_PREFIX.size(), STERAM_TREE_PREFIX) == 0 ||
      key_.compare(0, STREAM_DATA_HASH_PREFIX.size(), STREAM_DATA_HASH_PREFIX) == 0) {
    res_.SetRes(CmdRes::kErrOther, "hash key can't start with " + STERAM_TREE_PREFIX + " or " + STREAM_META_HASH_KEY);
    return;
  }

//pika_stream_base.h
// each abstracted tree has a prefix,
// prefix + treeID will be the key of the hash to store the tree,
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STERAM_TREE_PREFIX = "STR_TREE";

// key of the hash to store stream meta,
// notice: we need to ban the use of this key when using "HSET".
static const std::string STREAM_META_HASH_KEY = "STREAM";

// each stream's data (messages) are stored in a hash,
// to avoid stream key confilict with hash key, we add a prefix to each key for stream data,
// it's ok to use the same string as STREAM_META_HASH_KEY to be the prefix.
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STREAM_DATA_HASH_PREFIX = STREAM_META_HASH_KEY;

这样搞也行吧。为什么当初不选择单独整个db类型来支持stream

一开始是 db 的方案，后来被换掉了，因为blackwidow本身有一些缺陷可能要改，如果往里面加类型会增大后续改 blackwidow 的工作量。

[KKorpse]

你这个实现有没有可能hash key和xstream key冲突？

有两个前缀会冲突，我在 hash 中禁止掉了这两个前缀

  // pika_hash.cc
  // check the conflict of stream used prefix, see details in defination of STREAM_TREE_PREFIX
  if (key_.compare(0, STERAM_TREE_PREFIX.size(), STERAM_TREE_PREFIX) == 0 ||
      key_.compare(0, STREAM_DATA_HASH_PREFIX.size(), STREAM_DATA_HASH_PREFIX) == 0) {
    res_.SetRes(CmdRes::kErrOther, "hash key can't start with " + STERAM_TREE_PREFIX + " or " + STREAM_META_HASH_KEY);
    return;
  }

//pika_stream_base.h
// each abstracted tree has a prefix,
// prefix + treeID will be the key of the hash to store the tree,
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STERAM_TREE_PREFIX = "STR_TREE";

// key of the hash to store stream meta,
// notice: we need to ban the use of this key when using "HSET".
static const std::string STREAM_META_HASH_KEY = "STREAM";

// each stream's data (messages) are stored in a hash,
// to avoid stream key confilict with hash key, we add a prefix to each key for stream data,
// it's ok to use the same string as STREAM_META_HASH_KEY to be the prefix.
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STREAM_DATA_HASH_PREFIX = STREAM_META_HASH_KEY;

这样搞也行吧。为什么当初不选择单独整个db类型来支持stream

那有个疑问, stream的数据会不会 也能被 hash操作了, 比如通过 hash相关的命令 增删改查, 还有 在scan的时候 会不会也把 Steam的数据也 拿到了

我想了下，理论上如果 hset 不能调用成功，db 里面的 hash 就没有对应 key 的元数据，hash 的其它操作也不会执行成功。
我有空再进一步验证下。

[lqxhub]

你这个实现有没有可能hash key和xstream key冲突？

有两个前缀会冲突，我在 hash 中禁止掉了这两个前缀

  // pika_hash.cc
  // check the conflict of stream used prefix, see details in defination of STREAM_TREE_PREFIX
  if (key_.compare(0, STERAM_TREE_PREFIX.size(), STERAM_TREE_PREFIX) == 0 ||
      key_.compare(0, STREAM_DATA_HASH_PREFIX.size(), STREAM_DATA_HASH_PREFIX) == 0) {
    res_.SetRes(CmdRes::kErrOther, "hash key can't start with " + STERAM_TREE_PREFIX + " or " + STREAM_META_HASH_KEY);
    return;
  }

//pika_stream_base.h
// each abstracted tree has a prefix,
// prefix + treeID will be the key of the hash to store the tree,
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STERAM_TREE_PREFIX = "STR_TREE";

// key of the hash to store stream meta,
// notice: we need to ban the use of this key when using "HSET".
static const std::string STREAM_META_HASH_KEY = "STREAM";

// each stream's data (messages) are stored in a hash,
// to avoid stream key confilict with hash key, we add a prefix to each key for stream data,
// it's ok to use the same string as STREAM_META_HASH_KEY to be the prefix.
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STREAM_DATA_HASH_PREFIX = STREAM_META_HASH_KEY;

这样搞也行吧。为什么当初不选择单独整个db类型来支持stream

那有个疑问, stream的数据会不会 也能被 hash操作了, 比如通过 hash相关的命令 增删改查, 还有 在scan的时候 会不会也把 Steam的数据也 拿到了

我想了下，理论上如果 hset 不能调用成功，db 里面的 hash 就没有对应 key 的元数据，hash 的其它操作也不会执行成功。 我有空再进一步验证下。

突然想到了, 能不能给 hash的db 再加一个 column families 用来存stream的数据, 这样就能做到 和 hash的数据 硬隔离了

[KKorpse]

你这个实现有没有可能hash key和xstream key冲突？

有两个前缀会冲突，我在 hash 中禁止掉了这两个前缀

  // pika_hash.cc
  // check the conflict of stream used prefix, see details in defination of STREAM_TREE_PREFIX
  if (key_.compare(0, STERAM_TREE_PREFIX.size(), STERAM_TREE_PREFIX) == 0 ||
      key_.compare(0, STREAM_DATA_HASH_PREFIX.size(), STREAM_DATA_HASH_PREFIX) == 0) {
    res_.SetRes(CmdRes::kErrOther, "hash key can't start with " + STERAM_TREE_PREFIX + " or " + STREAM_META_HASH_KEY);
    return;
  }

//pika_stream_base.h
// each abstracted tree has a prefix,
// prefix + treeID will be the key of the hash to store the tree,
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STERAM_TREE_PREFIX = "STR_TREE";

// key of the hash to store stream meta,
// notice: we need to ban the use of this key when using "HSET".
static const std::string STREAM_META_HASH_KEY = "STREAM";

// each stream's data (messages) are stored in a hash,
// to avoid stream key confilict with hash key, we add a prefix to each key for stream data,
// it's ok to use the same string as STREAM_META_HASH_KEY to be the prefix.
// notice: we need to ban the use of this prefix when using "HSET".
static const std::string STREAM_DATA_HASH_PREFIX = STREAM_META_HASH_KEY;

这样搞也行吧。为什么当初不选择单独整个db类型来支持stream

那有个疑问, stream的数据会不会 也能被 hash操作了, 比如通过 hash相关的命令 增删改查, 还有 在scan的时候 会不会也把 Steam的数据也 拿到了

我想了下，理论上如果 hset 不能调用成功，db 里面的 hash 就没有对应 key 的元数据，hash 的其它操作也不会执行成功。 我有空再进一步验证下。

突然想到了, 能不能给 hash的db 再加一个 column families 用来存stream的数据, 这样就能做到 和 hash的数据 硬隔离了

是可以的，但要搞个方法判断是不是要存入这个特定的 column family，要么新增接口，要么搞个 flag，感觉都不太优雅。我建议既然选择了不在 blackwidow 做单独的类型，就尽量不侵入 blackwidow 层。

[wanghenshui]

我觉得还是单独搞个db 设计cf编码合理，你和hash捏在一块，可能影响hash结构的性能

当然这么搞我也没啥意见，得雨哥谦祥排一下

[AlexStocks]

我觉得还是单独搞个db 设计cf编码合理，你和hash捏在一块，可能影响hash结构的性能

当然这么搞我也没啥意见，得雨哥谦祥排一下

这个方案最早是我的意见，可以先实现一个版本。然后等后续再改进一版，在 blackwidow 层面进行升级。还是希望 @KKorpse 在这个 PR 合并完毕后，能进一步完成整体的工作。

[lqxhub]

我觉得还是单独搞个db 设计cf编码合理，你和hash捏在一块，可能影响hash结构的性能
当然这么搞我也没啥意见，得雨哥谦祥排一下

这个方案最早是我的意见，可以先实现一个版本。然后等后续再改进一版，在 blackwidow 层面进行升级。还是希望 @KKorpse 在这个 PR 合并完毕后，能进一步完成整体的工作。

good idea, 可以现有功能, 可以 后续在 blackwidow 或者 Floyd 里再优化存储

[KKorpse]

我觉得还是单独搞个db 设计cf编码合理，你和hash捏在一块，可能影响hash结构的性能
当然这么搞我也没啥意见，得雨哥谦祥排一下

这个方案最早是我的意见，可以先实现一个版本。然后等后续再改进一版，在 blackwidow 层面进行升级。还是希望 @KKorpse 在这个 PR 合并完毕后，能进一步完成整体的工作。

好的，以后可以根据存储层的具体方案来优化，blackwidow或者floyd都可以。

[AlexStocks]

@KKorpse 如果用户没使用 stream 但是用 hashtable，当前的实现不会影响 hashtable 的性能吧？
@lqxhub 谦祥你给瞧下这个问题

[KKorpse]

@KKorpse 如果用户没使用 stream 但是用 hashtable，当前的实现不会影响 hashtable 的性能吧？ @lqxhub 谦祥你给瞧下这个问题

应该不会有影响，不用 stream 就没有 stream 的任何数据，hash 本身的逻辑也没有做改动。

[lqxhub]

@KKorpse 如果用户没使用 stream 但是用 hashtable，当前的实现不会影响 hashtable 的性能吧？ @lqxhub 谦祥你给瞧下这个问题

好的, 有时间我尝试分析一下这个问题

[cheniujh]

在单机测试部分，建议加上针对同一个stream的，多个线程生产，多个线程消费同时进行的并发测试，这也是实际使用时经常出现的场景，同时也能对锁的正确性进行一个基础检验。
（如果我没有漏看的话，似乎没有这部分测试？）