RFC: Support The Binlog of Vertex And Edge

[zhaojunnana]

    背景:
    生产使用中很多场景需要用到图库变更日志，作为下游业务的触发点。如用户特征的生产流程中，图库作为中游数据关系聚合使用，当用户的某些属性发生更新后，才会触发下游特征的重新读库、更新和构建。
    类似业务注重两个点：一是触发频率，必须是真正的发生变化才需执行；二是数据落库的时间，触发之后马上去读，库中必须已经更新。这两点就意味着再数据库的外部作操作会有很大的难度，最直接的方法就是数据库自己吐出本身的变更状态。

    方案:
    nebula的落库是在raft阶段，主分片先记wal然后发AppendLogs到从节点，等收到大多数副本成功的响应再CommitLog。记BinLog的时机应该在leader part的CommitLog中，真正提交日志之前。
    由于rocksdb的原因，想确定用户的insert、update和delete是否真的对数据有影响，必须先做一次get比较value值。因此对库的性能会有些影响，但因为是get操作，整体来说影响是可控并且比较轻微的。
    上述所描述的BinLog面向的是业务使用，只是起到触发点的作用，和类似MySQL的依赖BinLog做数据同步有一定区别，不考虑打印详细的变更信息(不解析value值，以免性能影响过大)及索引信息。

    最终结果如下:

I20230228 21:04:27.854123 3216573 Part.cpp:466] _BINLOG:{"spaceId":1, "tagId":2, "type":"VERTEX", "logID":26999, "termID":42, "vid": "8", "spaceType":"INT", "operate":"UPDATE"}

I20230228 21:04:38.709828 3216573 Part.cpp:466] _BINLOG:{"spaceId":1, "tagId":2, "type":"VERTEX", "logID":27002, "termID":42, "vid": "8", "spaceType":"INT", "operate":"UPDATE"}

I20230228 21:05:03.793665 3216573 Part.cpp:466] _BINLOG:{"spaceId":1, "tagId":2, "type":"VERTEX", "logID":27007, "termID":42, "vid": "8", "spaceType":"INT", "operate":"UPDATE"}

I20230228 21:05:30.797564 3216573 Part.cpp:466] _BINLOG:{"spaceId":1, "tagId":2, "type":"VERTEX", "logID":27033, "termID":46, "vid": "18", "spaceType":"INT", "operate":"INSERT"}

I20230228 21:06:03.608285 3216575 Part.cpp:466] _BINLOG:{"spaceId":1, "tagId":2, "type":"VERTEX", "logID":27037, "termID":46, "vid": "18", "spaceType":"INT", "operate":"REMOVE"}

I20230228 21:06:26.705806 3216575 Part.cpp:466] _BINLOG:{"spaceId":1, "tagId":2, "type":"VERTEX", "logID":27018, "termID":42, "vid": "8", "spaceType":"INT", "operate":"REMOVE"}

I20230228 21:06:50.605970 3216575 Part.cpp:488] _BINLOG:{"spaceId":1, "edgeType":-4, "type":"EDGE", "logID":27021, "termID":42, "source": "2", "rank":3, "target": "1", "spaceType":"INT", "operate":"REMOVE"}

I20230228 21:06:50.606010 3216574 Part.cpp:488] _BINLOG:{"spaceId":1, "edgeType":4, "type":"EDGE", "logID":27038, "termID":44, "source": "1", "rank":3, "target": "2", "spaceType":"INT", "operate":"REMOVE"}

I20230228 21:06:50.606045 3216575 Part.cpp:488] _BINLOG:{"spaceId":1, "edgeType":-4, "type":"EDGE", "logID":27021, "termID":42, "source": "2", "rank":4, "target": "1", "spaceType":"INT", "operate":"REMOVE"}

I20230228 21:06:50.606078 3216574 Part.cpp:488] _BINLOG:{"spaceId":1, "edgeType":4, "type":"EDGE", "logID":27038, "termID":44, "source": "1", "rank":4, "target": "2", "spaceType":"INT", "operate":"REMOVE"}

I20230228 21:06:50.606063 3216575 Part.cpp:488] _BINLOG:{"spaceId":1, "edgeType":-4, "type":"EDGE", "logID":27021, "termID":42, "source": "2", "rank":5, "target": "1", "spaceType":"INT", "operate":"REMOVE"}

I20230228 21:06:50.606107 3216574 Part.cpp:488] _BINLOG:{"spaceId":1, "edgeType":4, "type":"EDGE", "logID":27038, "termID":44, "source": "1", "rank":5, "target": "2", "spaceType":"INT", "operate":"REMOVE"}

I20230228 21:06:50.606120 3216575 Part.cpp:488] _BINLOG:{"spaceId":1, "edgeType":-4, "type":"EDGE", "logID":27021, "termID":42, "source": "2", "rank":6, "target": "1", "spaceType":"INT", "operate":"REMOVE"}

I20230228 21:06:50.606159 3216575 Part.cpp:488] _BINLOG:{"spaceId":1, "edgeType":-4, "type":"EDGE", "logID":27021, "termID":42, "source": "2", "rank":10, "target": "1", "spaceType":"INT", "operate":"REMOVE"}

I20230228 21:06:50.606139 3216574 Part.cpp:488] _BINLOG:{"spaceId":1, "edgeType":4, "type":"EDGE", "logID":27038, "termID":44, "source": "1", "rank":6, "target": "2", "spaceType":"INT", "operate":"REMOVE"}

I20230228 21:06:50.606220 3216574 Part.cpp:488] _BINLOG:{"spaceId":1, "edgeType":4, "type":"EDGE", "logID":27038, "termID":44, "source": "1", "rank":10, "target": "2", "spaceType":"INT", "operate":"REMOVE"}

I20230228 21:06:50.607018 3216575 Part.cpp:466] _BINLOG:{"spaceId":1, "tagId":2, "type":"VERTEX", "logID":27022, "termID":42, "vid": "2", "spaceType":"INT", "operate":"REMOVE"}

    问题:
    使用现在的日志打印方式，BinLog日志会和运行日志耦合在一起，且日志保存时间、占用磁盘不可控，期待更好的优化方案。

[HarrisChu]

额外提两个点，可能还是会有影响：

1. ttl，ttl 真正删数据，是在 compact 的时候，这部分不走 raft，CDC 如何能感知到。
2. alter schema，新增一个带有 default value 的属性。这部分 nebula 是不刷数据，从 storage 读出数据后，拼上默认值，如果这种情况下，从 nebula 读的数据，可能会和下游不一致。

[HarrisChu]

不过这些更多是本身 nebula 设计不太好的地方，不影响 CDC 的主流程。

[zhaojunnana]

暂时还没考虑到静态数据这块，binlog目前想到的是只能针对有动态写入的情况，例如有一条insert进来，去判断下这个insert是真正的插入了一条数据，还是是修改了某些属性，或者没做任何操作。只有有动态变更，才会生成动态日志。

[zhaojunnana]

ttl和alter schema这种确实很难实现，考虑到就算有办法获取感知，但可能触发大量的数据，性能应该感觉也会很大

[cangfengzhs]

两个问题：

1. binlog发送到远端还是本地文件？一般来说应该是kafka一类的吧
2. commitLog时如果binlog写入失败，那么显然是不能commit的。这时数据库的所有写入实际上都被block了，我觉得binlog不应该影响主流程

是不是可以这样搞：

1. 在AppendLog时，除了原本的数据之外，我们再落盘一个binlog。
2. Storage内部去消费binlog发送到远端。消费完成后再删除binlog。
   这样保证了数据至少发送一次

[zhaojunnana]

1. binlog我感觉还是存储在本地吧，不再额外耦合其他东西，本地存文件，剩下由使用方去决定
2. commitLog时写入binlog，考虑到了只有part leader commitLog时才记一份binlog日志。再commitLog写日志面临的问题其实主要是怕影响整体的commit和确保binlog成功写入，我当前的实现只是用日志框架做了个日志打印，应该不会存在这两个问题。但是我不太清楚这个日志框架能否将日志分离开，这确实是个问题。

大佬上面提的确实是优秀的实现方案

[cangfengzhs]

存本地确实简单，也不用考虑外部系统不可用导致binlog刷不出去。

emmmm，怎么说呢，假如消费binlog的那一方访问不了机器（比如没有权限之类的），那就没的玩了。

看你需求吧，binlog刷出去可能比较符合主流做法，但是它确实会难一点。

[wey-gu]

降低影响的话可以考虑只在一个 listener 里做 cdc 的事情哈？

[zhaojunnana]

好的 感谢启发，我尝试下。 但是磁盘上确实会多一次get操作，感觉不需要这个功能的还是慎开。

[wey-gu]

嗯嗯，能够（无痛）开启 watch/even-driven 模式真的打开一扇大门，超级期待！！

[AndyShen105]

现在CDC的方案是打算通过Listener来实现吗 各位大佬