编者按：本文具体介绍 Milvus2.0 如何对查问节点的数据进行治理，以及如何提供查问能力
内容纲要：
- 疾速回顾 Milvus 进行数据插入与长久化存储相干的流程及机制；
- 如何将数据加载进查问节点（Query Node）以进行查问操作；
- Milvus 上实现实时查问的相干操作和流程。
疾速回顾 Milvus 进行数据插入与长久化相干的流程与机制
Milvus 架构疾速回顾
如下图所示，Milvus 向量数据库的整体架构能够分为 coordinator service、worker node、message storage 和 object storage 这几大部分。
Coordinator services 承当的次要工作是协调各个 worker node 的工作，其中的各个模块与 work node 是一一对应的关系，并协调治理各个 node 之间的工作。如架构图中所示，query coordinator 对应并协调 query node，data coordinator 对应并协调 data node，index coordinator 对应并协调 index node。
Data node 负责数据的长久化存储，基本上是一个 I/O 密集型的工作负载，负责把数据从 log broker 中写到最终的 object storage 当中。而 index node 负责实现向量索引的构建，最初由 query node 来承当整个 Milvus 的查问工作，这两类 node 是数据计算密集型的节点，
除此之外，零碎架构中还有两个比拟重要的局部：message storage 和 object storage。
Message storage 相当于一个 WAL 的货色，当数据插入到这个中央之后，零碎会保障数据不会有失落。其中的 log broker 会默认将数据寄存 7 天，这期间即便上面的 work node 呈现了局部宕机的状况，零碎也能够从 log broker 中复原一些数据及状态。Object storage 负责实现数据长久化存储，log broker 外面的数据最终都会长久化到 object storage 外面，以进行数据的长期保留。
总体来说这个架构相当于一个存储与计算拆散的一个零碎，data 这边负责数据存储，而后 query 这边负责查问计算。
数据插入流程
第一步：Insert Message 从 SDK 发到 proxy 之后，proxy 把这个 insert message 插到相应的 log broker 中，插入到 log broker 中的每条音讯都有惟一的主键和一个工夫戳；
第二步：插入到 log broker 之后，数据会被 data node 生产；
第三步：Data node 会把数据写入进长久化存储当中，最终数据在长久化存储中是基于 segment 的粒度来组织的，也就是说这个音讯除了中主键和工夫戳，还会被额定赋予一个 segment ID，以标识出这条数据最终会属于哪个 segment。Data note 在收到这些信息之后，会把相应的信息写入相应的 segment 中，并最终写入到长久化存储中去。
第四、五步：在数据被长久化之后，如果说基于这些数据间接做查问的话，查问速度会比较慢，因而个别状况下会思考去构建一些索引去以减速查问速度。这时 index node 就会把信息从长久化存储里拉进去并构建索引，而构建的索引文件又会被回写进长久化存储中（S3 或 Minio 等等）。有时咱们会须要构建多个索引，以从当选挑选出其中查问速度最快的一个，这样的操作也能够在 index node 中实现。
Log broker 和 object storage 也是 Milvus 架构中保障数据可靠性很重要的两局部，在零碎设计中这两局部也能够别离抉择一些第三方组件，来保障不同状况下的可靠性。
一种常见的状况，是在查问的同时也进行数据插入，这时一部分数据处在 log broker 中，而一部分数据处于 object storage 外面。咱们把这两部数据别离做了定义，在 object storage 外面的数据为 批数据，而在 log broker 外面的是流数据。不言而喻，在做实时查问的场景下，如果想遍历所有曾经插入的数据，则必须要在流数据和批数据里同时做查问，能力返回正确的实时查问数据。
数据组织机制
接下来看一下数据存储的相干机制，数据分两局部存储。一部分是在 object storage；一部分是在 log broker。
首先看一下在 log broker 外面，数据的组织模式是怎么的呢？
能够看参考下图，数据可分成这几局部：惟一的 collection ID、惟一的 partiton ID、惟一 的 segment ID。
每个 collection 在零碎外面都会调配指定数量的 channel，能够了解成是相似 Kafka 中的 topic，或相似传统数据库外面的 shard 的概念。
在图示中，如果咱们对 collection 调配了三个 channel，假如咱们要插入 100 条数据，那么这 100 条数据会均匀的分到这三个 channel 中，而后在三个 channel 外面，数据又是以 segment 为粒度进行拆分。目前每个 segment 的容量是有下限的，零碎默认最大到 512M。在继续的数据插入过程中，会优先继续往一个 segment 中写入，但如果容量超过 512M，零碎会新调配一个 segment ID 持续数据插入。所以在实在的场景中，每个 channel 外面都会蕴含很多个 segment。总结来说，数据在 log broker 中，能够拆分成，collection、partition 和 segment，最终咱们存储在零碎外面，实际上是很多个小的 segment。
接下来，咱们再看一下在 object storage 中的数据组织形式。
与 log broker 一样，data node 在收到 insert message 之后，也是依照 segment 进行组织的。当一个 segment 达到 512M 的默认下限时，或者用户间接强制进行对这个 segment 插入数据，这时 segment 会被长久化存储进 object storage 当中。在长久化存储中，每个 segment 中的存储格局是一个一个更小的 log snapshot，而且是分成多列的。具体的这个列数是和待插入的 collection 的 schema 无关。如果 collection 的 schema 有 4 列，数据插入 segment 中也会有 4 列。所以，最终在 object storage 中，数据存储的模式是很多个 log snapshot。
如何将数据加载进查问节点 query node
数据加载流程详解
在明确了数据的组织形式后，接下来咱们看看数据进行查问加载的具体流程。
在 query node 中，把 log broker 中的流数据称为 streaming，把 object storage 中的批数据称为 historical。流数据和批量数据的加载流程如下：
首先，query coord 会询问 data coord。Data coord 因为始终在负责继续的插入数据，它能够反馈给 query coord 两种信息：一种是曾经长久化存储了哪些 segment，另一种是这些曾经长久化的 segment 所对应 checkpoint 信息，依据 checkpoint 能够晓得从 log broker 中取得这些 segment 所生产到的最初地位。
接着，在收到这两局部信息后，query coord 会输入肯定的调配策略。这些策略也分成两局部：依照 segment 进行调配（如图示 segment allocator），或依照 channel 进行调配（如图示 channel allocator）。
Segment allocator 会把长久化存储 - 也就是批数据 - 中的不同的 segment 调配给不同的 query node 进行解决，如图将 S1、S3 调配给 query node 1，将 S2、S4 调配给 query node 2。Channel allocator 会把 log broker 中不同的 channel 调配给不通的 query node 进行监听，如图 querynode 1 监听 Ch 1，query node 2 监听 Ch 2。
这些调配策略下发到各个 query node 之后，query node 就会依照策略进行相应的 load 和 watch 操作。如图示 query node 1 中，historical（批数据）局部会将调配给它的 S1、S3 数据从长久化存储中加载进来，而 streaming 局部会订阅 log broker 中的 Ch1，将这部分流数据接入。
因为 Ch1 能够继续一直的插入数据（流数据），而由这部分接入 query node 中的数据咱们定义为 growing segment，因为会继续一直的增长，是增量数据，如图示的 G5。绝对应的，histroical 中的 segment 定义 sealed segment，是动态的存量数据。
数据管理与保护
对于 sealed segment 的的治理，零碎的设计次要思考负载平衡和宕机的状况。
如图示，如果 query node 4 下面有很多这个 sealed segment，然而其余节点比拟少，在这种状况下 query node 4 的查问可能是整个查问外面的一个瓶颈。所以这时，零碎就要思考说把这些 sealed segment 负载平衡到到其余节点下来。
另一种状况，如果某一个节点忽然挂掉了，这个时候它下面的负载也可能疾速的迁徙到其余失常节点上，以保障查问到的后果是正确的。
对增量数据来讲，方才提到说 query node 监听相应的 dmchannel 之后，这些增量数据都就会进入到 query node 里。但具体是怎么进入的呢？这里咱们用到了一个 flow graph 模型，一种状态驱动的模型，整个 flowGraph 包含 input node, filter node, insert node 和 service time 四局部。首先，input node 负责从流外面收到 Insert 音讯，而后 filter node 对音讯进行过滤。为什么须要过滤呢？因为用户可能仅须要加载 collection 下的某一个 partition 数据。过滤完之后，insert node 把这些数据插到底层的 growing sagment 中。在这当前 server time node 负责更新查问的服务工夫
最开始咱们回顾数据 insert 流程时提到，每一条 insert message 中都有调配了一个工夫戳。
大家能够参看图示左侧的例子，如果说数据从左到右只顺次插入，那么第一条音讯插入的工夫戳是 1，第二条音讯插入的工夫戳是 2，第三条音讯操作工夫戳是 6，第四条这里为什么标红呢？这是零碎插入的 timetick message，它代表的不是 insert message。Timticker 示意 timestamp 小于这个 timetick 的插入数据都曾经在 log broker 中了。换句话说，在这个 timetick 5 之后呈现的 insert message 它们所对应的工夫戳不会小于 5，能够看到前面的几条工夫戳别离是 7、8、9、10，工夫戳都是大于 5 的，也就是说工夫戳小于 5 的 insert message 音讯必定都会呈现在左侧。换句话说，当 query node 收到 timetick = 5 的音讯时，能够确定说工夫戳在 5 之前的所有音讯都曾经进入到 qurey node 中，从而来确认查问的正确。那么这里的 server time node 就是在从 insert node 接管到 timetiker 后，比方图示的 5 或 9，会更新一个 tsafe，相当于一个示意平安的工夫戳，只有 tsafe 到了 5，那么 5 之前的数据都是能够查的。
有了这些铺垫，上面开始讲如何真正的做 query 的这部分。
Milvus 上实现实时查问的相干操作和流程
首先讲一下查问申请（query message）是如何定义的。
Query message 同样由 proxy 插入到 log broker, 在之后 query node 会通过监听 log broker 中的 query channel，来获取到 query message。
Query message 具体长什么样呢？
- Message ID，对这个查问零碎调配的一个全局调配的 ID；
- Collection ID：query 申请对应的 collection ID，如果说 query 是制订在 collection 中查问，那么它要指定对应的 collection ID。当然在 SDK 那边，其实这个中央指定的是 collection name, 在零碎内会对 name 和 ID 做一对一的映射。
- execPlan：执行数，对应 SDK 那边的操作，相当于在 SDK 做查问的时候指定了表达式，也就是一个 PR。对于向量查问来讲，次要是做属性过滤的，如果说某一个属性大于 10 或者是等于 10 做一些应用过滤。
- Service timestamp：上文提到的 tsafe 更新之后，service timestamp 也会相应更新，用来阐明当初服务的工夫到哪个点了，在此之前插入的数据都能够进行查问。
- Travel timestamp：如果须要对对某一个时间段之前的数据进行查问，能够通过（services timestamp – travle timestamp）来标定新的工夫戳和数据范畴；
- Guarantee timestmap：如果须要对某一个时间段之后的在进行数据查问，只有当 services timestam 大于等于 guarantee timestamp 这个条件满足时，查问工作才会开始。
当初看一下具体的查问操作流程：
收到 query message 之后，零碎会先去做一个判断，如果 service time 大于 query message 中的 guarantee timestamp，那么就会执行这个查问。查问分成两个同时并行的局部，一部分是长久化存储的 historical data，另一部分是 log broker 中的 streaming data。最初会做一个 local reduce。之前也讲过 historical 和 streaming 两头因为种种原因是可能会呈现一些数据的反复的，那么这里最初就须要先做一个 reduce。
以上是比较顺利的流程。而如果说在第一步判断工夫戳是，可服务工夫还没能推动到 guarantee timestamp，那么这个查问会放进 unsolved meessage, 始终期待，直到满足条件能够进行查问。
最终后果会被推送到 result channel，由 proxy 来承受。当然 proxy 会从很多 query node 下面承受后果，也会在做一轮 global reduce。到此整个查问流程结束。
但这里还有一个问题，就是 proxy 在向 SDK 返回最终后果之前，如何去确定曾经收到了全副的查问后果。为此咱们做了一个策略：在返回的 result message 中，也会记录下，哪些 sealed segments 被查问过（searched sealed segments），以及哪些 dmChannel 被查问过（dmchannels searched），以及在 querynode 上有哪些 segment（global sealed segments）。如果所有 query node 的 search result 里 searched sealed segments 的并集大于 global sealed segments，而且这个 collection 的所有 dmchannel 对应的增量数据都被查问过，就认为所有的查问后果都收到了，proxy 就能够进行 reduce 操作，并将后果最终返回给 SDK。