CreatePartition API执行流程源码解析

milvus版本:v2.3.2

整体架构:

CreatePartition 的数据流向:

1.客户端sdk发出CreatePartition API请求。

from pymilvus import ( connections, Collection, Partition, ) print("start connecting to Milvus") connections.connect(db_name="default", host="192.168.230.71", port="19530") hello_milvus = Collection("hello_milvus") print("create a partition") partition = Partition(hello_milvus, name="part01", description="this is a partition") 

客户端SDK向proxy发送一个CreatePartition API请求，在hello_milvus这个collection下创建一个名为part01的partition。

2.客户端接受API请求，将request封装为createPartitionTask，并压入ddQueue队列。

代码路径:internal\proxy\impl.go

// CreatePartition create a partition in specific collection. func (node *Proxy) CreatePartition(ctx context.Context, request *milvuspb.CreatePartitionRequest) (*commonpb.Status, error) {  ...... // request封装为task cpt := &createPartitionTask{  ctx: ctx, Condition: NewTaskCondition(ctx), CreatePartitionRequest: request, rootCoord: node.rootCoord, result: nil, } ...... // 将task压入ddQueue队列 if err := node.sched.ddQueue.Enqueue(cpt); err != nil {  ...... } ...... // 等待cct执行完 if err := cpt.WaitToFinish(); err != nil {  ...... } ...... } 

3.执行createPartitionTask的3个方法PreExecute、Execute、PostExecute。

PreExecute()一般为参数校验等工作。

Execute()一般为真正执行逻辑。

代码路径:internal\proxy\task.go

func (t *createPartitionTask) Execute(ctx context.Context) (err error) {  t.result, err = t.rootCoord.CreatePartition(ctx, t.CreatePartitionRequest) if err != nil {  return err } if t.result.ErrorCode != commonpb.ErrorCode_Success {  return errors.New(t.result.Reason) } return err } 

从代码可以看出调用了rootCoord的CreatePartition接口。

4.进入rootCoord的CreatePartition接口。

代码路径:internal\rootcoord\root_coord.go

继续将请求封装为rootcoord里的createDatabaseTask

// CreatePartition create partition func (c *Core) CreatePartition(ctx context.Context, in *milvuspb.CreatePartitionRequest) (*commonpb.Status, error) {  ...... // 封装为createPartitionTask t := &createPartitionTask{  baseTask: newBaseTask(ctx, c), Req: in, } // 加入调度 if err := c.scheduler.AddTask(t); err != nil {  ...... } // 等待task完成 if err := t.WaitToFinish(); err != nil {  ...... } ...... } 

5.执行createPartitionTask的Prepare、Execute、NotifyDone方法。

Execute()为核心方法。

代码路径:internal\rootcoord\create_partition_task.go

func (t *createPartitionTask) Execute(ctx context.Context) error {  for _, partition := range t.collMeta.Partitions {  if partition.PartitionName == t.Req.GetPartitionName() {  log.Warn("add duplicate partition", zap.String("collection", t.Req.GetCollectionName()), zap.String("partition", t.Req.GetPartitionName()), zap.Uint64("ts", t.GetTs())) return nil } } cfgMaxPartitionNum := Params.RootCoordCfg.MaxPartitionNum.GetAsInt() if len(t.collMeta.Partitions) >= cfgMaxPartitionNum {  return fmt.Errorf("partition number (%d) exceeds max configuration (%d), collection: %s", len(t.collMeta.Partitions), cfgMaxPartitionNum, t.collMeta.Name) } // 分配partID partID, err := t.core.idAllocator.AllocOne() if err != nil {  return err } // 构建partition结构体 // 包含partID,partName,collectID等 partition := &model.Partition{  PartitionID: partID, PartitionName: t.Req.GetPartitionName(), PartitionCreatedTimestamp: t.GetTs(), Extra: nil, CollectionID: t.collMeta.CollectionID, State: pb.PartitionState_PartitionCreating, } undoTask := newBaseUndoTask(t.core.stepExecutor) // 分为多个step执行，每一个undoTask由todoStep和undoStep构成 // 执行todoStep,报错则执行undoStep undoTask.AddStep(&expireCacheStep{  baseStep: baseStep{ core: t.core}, dbName: t.Req.GetDbName(), collectionNames: []string{ t.collMeta.Name}, collectionID: t.collMeta.CollectionID, ts: t.GetTs(), }, &nullStep{ }) // 添加partition元数据 undoTask.AddStep(&addPartitionMetaStep{  baseStep: baseStep{ core: t.core}, partition: partition, }, &removePartitionMetaStep{  baseStep: baseStep{ core: t.core}, dbID: t.collMeta.DBID, collectionID: partition.CollectionID, partitionID: partition.PartitionID, ts: t.GetTs(), }) undoTask.AddStep(&nullStep{ }, &releasePartitionsStep{  baseStep: baseStep{ core: t.core}, collectionID: t.collMeta.CollectionID, partitionIDs: []int64{ partID}, }) undoTask.AddStep(&syncNewCreatedPartitionStep{  baseStep: baseStep{ core: t.core}, collectionID: t.collMeta.CollectionID, partitionID: partID, }, &nullStep{ }) undoTask.AddStep(&changePartitionStateStep{  baseStep: baseStep{ core: t.core}, collectionID: t.collMeta.CollectionID, partitionID: partID, state: pb.PartitionState_PartitionCreated, ts: t.GetTs(), }, &nullStep{ }) return undoTask.Execute(ctx) } 

创建partition涉及多个步骤，可以看出这里依次分为expireCacheStep、addPartitionMetaStep、syncNewCreatedPartitionStep、changePartitionStateStep这几个步骤，关于etcd元数据的操作，这里重点关注addPartitionMetaStep。其余step另用篇幅进行讲解。

6.进入addCollectionMetaStep，执行其Execute()方法。

代码路径:internal\rootcoord\step.go

func (s *addPartitionMetaStep) Execute(ctx context.Context) ([]nestedStep, error) {  err := s.core.meta.AddPartition(ctx, s.partition) return nil, err } 

在这里重点研究s.core.meta.AddPartition()这个方法做了什么事情。

调用栈如下:

s.core.meta.AddPartition() |--AddPartition()(internal\rootcoord\meta_table.go) |--mt.catalog.CreatePartition()(同上) |--CreatePartition()(internal\metastore\kv\rootcoord\kv_catalog.go) |--kc.Snapshot.Save() 

在etcd产生partition相关的key：

==root-coord/partitions/445762854989594797/445766775506484222==

value的值的结构为etcdpb.PartitionInfo，然后进行protobuf序列化后存入etcd。

因此etcd存储的是二进制数据。

&pb.PartitionInfo{  PartitionID: partition.PartitionID, PartitionName: partition.PartitionName, PartitionCreatedTimestamp: partition.PartitionCreatedTimestamp, CollectionId: partition.CollectionID, State: partition.State, } 

可以看出partition由partitionID、partitionName、collectionID等组成。

func (kc *Catalog) CreatePartition(ctx context.Context, dbID int64, partition *model.Partition, ts typeutil.Timestamp) error {  collMeta, err := kc.loadCollection(ctx, dbID, partition.CollectionID, ts) if err != nil {  return err } if partitionVersionAfter210(collMeta) {  // save to newly path. // 走这条路径 // 构建key的规则 k := BuildPartitionKey(partition.CollectionID, partition.PartitionID) partitionInfo := model.MarshalPartitionModel(partition) // 序列化 v, err := proto.Marshal(partitionInfo) if err != nil {  return err } // 写入etcd return kc.Snapshot.Save(k, string(v), ts) } ...... } 

跟踪BuildPartitionKey()函数，不难得出key的规则。整理如下:

key规则:

• 前缀/root-coord/partitions/{collectionID}/{partitionID}
• 前缀/snapshots/root-coord/partitions/{collectionID}/{partitionID}_ts{时间戳}

根据路径能够反映出partition属于哪个collection。一个collection可以包含多个partition。

默认partition名为:_default。

可配置(milvus.yml):common.defaultPartitionName

使用etcd-manager查看etcd：

总结:

1.CreatePartition由proxy传递给协调器rootCoord操作etcd。

2.CreatePartition最终会在etcd上写入一种类型的key

• 前缀/root-coord/partitions/{collectionID}/{partitionID}
• 前缀/snapshots/root-coord/partitions/{collectionID}/{partitionID}_ts{时间戳}