讨论区：流处理引擎中基于共享状态索引的 Delta Join

[skyzh]

原文链接：https://www.skyzh.dev/posts/articles/2022-05-29-shared-state-in-risingwave/

[mapleFU]

迟老师迟老师，首先恭贺毕业。

其次有一个问题，这个是相当于引入了一个 Epoch，让系统有根据 Epoch 的一致的视图？然后我看这个是发起 Checkpoint，然后从表开始一步步推进这个状态。那么这个时候，如果某个上游的请求过来，它会带 epoch吗？如果会的话，假如这个算子没有到这个 epoch，它会等待查询的节点到这个 epoch 吗？如果某个流，在某段时间内没有更新，但是来了个 epoch 更高的请求，它会怎么样呢？（我不是很熟悉流，这一段用词可能不合理，见谅）

[skyzh]

其次有一个问题，这个是相当于引入了一个 Epoch，让系统有根据 Epoch 的一致的视图？

是的

那么这个时候，如果某个上游的请求过来，它会带 epoch吗？

上游的请求是指用户的批查询？批查询走批算子，和流算子互不影响。批处理查询会首先向 meta service 询问最新的已经 checkpoint 完成的 epoch，然后直接从共享存储上以这个 epoch 读数据。

[mapleFU]

@skyzh 非常感谢解惑！所以它不会请求一个「正在 checkpoint」的数据，不会有这种需要等待的问题是吗？

[skyzh]

所以它不会请求一个「正在 checkpoint」的数据，不会有这种需要等待的问题是吗？

是的 🤪

[SaintBacchus]

@skyzh 请问一下, Delta Join支持Delete类型的Delta吗? Insert类型的Delta比较好理解, 但是如果是更新模式的话, 是不是删除部分是不是和新增部署是不是要单独切分出来计算呢

[skyzh]

支持，不用。

[SaintBacchus]

那请问一下, Delete类型和Update类型的数据, 在join算子里面是怎么处理的呢? 是不是这行记录要特殊标记一下? 因为后续要做Unoin,而Delete语义应该是做差集吧

[skyzh]

Union 算子本身不做并集差集的计算，只负责把收到的消息按顺序传给下游

[skyzh]

insert delete 标记在信息上，insert 经过 join 产生 insert 消息；delete 经过 join 产生 delete 消息

[SaintBacchus]

了解了. 还请教一下, rising wave 里面如何处理状态和表存储两者的关系呢? 例如简单的join sql

select a.f1,b.f2 from a,b where a.id =b.id

在这例子中, 按照文档中的设计a和b的数据在状态里面完整的存储了一份, 而且在表的存储数据中也同时存储了一份, 那存储成本是不是原来的2倍了呢

[skyzh]

是的

[skyzh]

通常 join 用到的列是原表的一部分，列裁剪后比原表小很多

[SaintBacchus]

通常 join 用到的列是原表的一部分，列裁剪后比原表小很多

这个是会小很多. Flink里面一般都会设置一个TTL时间防止状态过大, 在Rising Wave中应该不需要TTL设置吧, 那他是怎么防止状态慢慢变大的问题呢? 是对状态做的分区吗? 另外请教一下这部分状态管理和join实现是在这里吗

[skyzh]

1. 还在探索如何在提供正确的 SQL 语义的同时减少状态大小。直接对状态设置 TTL 会导致一些问题，不是好的用法。
2. RisingWave 有两套 join。主要可以看 hash join 而不是 delta join。

[SaintBacchus]

1. 还在探索如何在提供正确的 SQL 语义的同时减少状态大小。直接对状态设置 TTL 会导致一些问题，不是好的用法。
2. RisingWave 有两套 join。主要可以看 hash join 而不是 delta join。

Got, Thx~