CreatePartition API执行流程_syncNewCreatedPartitionStep_milvus源码解析

本文涉及的产品
RDS PostgreSQL Serverless,0.5-4RCU 50GB 3个月
推荐场景:
对影评进行热评分析
阿里云百炼推荐规格 ADB PostgreSQL,4核16GB 100GB 1个月
简介: CreatePartition API执行流程_syncNewCreatedPartitionStep_milvus源码解析

CreatePartition API执行流程源码解析

milvus版本:v2.3.2

syncNewCreatedPartitionStep_milvus源码解析

整体架构:

architecture.png

CreatePartition 的数据流向:

create_partition数据流向2.jpg

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。

exam_partition.jpg

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},
    })
    // 同样也是添加partition元数据,侧重于load后的partition
    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元数据的操作,这里重点关注syncNewCreatedPartitionStep。

6.进入syncNewCreatedPartitionStep,执行其Execute()方法。

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

func (s *syncNewCreatedPartitionStep) Execute(ctx context.Context) ([]nestedStep, error) {
   
   
    err := s.core.broker.SyncNewCreatedPartition(ctx, s.collectionID, s.partitionID)
    return nil, err
}

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

调用栈如下(分2种情况):

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

collection未load:

s.core.broker.SyncNewCreatedPartition()
  |--SyncNewCreatedPartition()(internal\rootcoord\broker.go)
    |--b.s.queryCoord.SyncNewCreatedPartition(同上)
      |--SyncNewCreatedPartition()(internal\querycoordv2\services.go)
        |--job.NewSyncNewCreatedPartitionJob()(同上)
          |--Execute()(internal\querycoordv2\job\job_sync.go)

这种情况下,返回nil,不操作etcd。

collection已load:

s.core.broker.SyncNewCreatedPartition()
  |--SyncNewCreatedPartition()(internal\rootcoord\broker.go)
    |--b.s.queryCoord.SyncNewCreatedPartition(同上)
      |--SyncNewCreatedPartition()(internal\querycoordv2\services.go)
        |--job.NewSyncNewCreatedPartitionJob()(同上)
          |--Execute()(internal\querycoordv2\job\job_sync.go)
            |--job.meta.CollectionManager.PutPartition()(同上)
              |--m.putPartition()(internal\querycoordv2\meta\collection_manager.go)
                |--m.catalog.SavePartition(同上)
                  |--SavePartition()(internal\metastore\kv\querycoord\kv_catalog.go)
                    |--s.cli.Save()(同上)

这种情况下,会操作etcd。

create_partition堆栈2.jpg

在etcd产生partition相关的key:

==querycoord-partition-loadinfo/445762854989594797/445788143864261250==

value的值的结构为querypb.PartitionLoadInfo,然后进行protobuf序列化后存入etcd。

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

type PartitionLoadInfo struct {
   
   
    CollectionID         int64           
    PartitionID          int64           
    ReplicaNumber        int32           
    Status               LoadStatus      
    FieldIndexID         map[int64]int64 
    RecoverTimes         int32           
    XXX_NoUnkeyedLiteral struct{
   
   }        
    XXX_unrecognized     []byte          
    XXX_sizecache        int32           
}

partitionloadinfo.jpg

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

func (s Catalog) SavePartition(info ...*querypb.PartitionLoadInfo) error {
   
   
    for _, partition := range info {
   
   
        // 构建key规则
        // querycoord-partition-loadinfo/collectionID/partitionID
        k := EncodePartitionLoadInfoKey(partition.GetCollectionID(), partition.GetPartitionID())
        // 序列化
        v, err := proto.Marshal(partition)
        if err != nil {
   
   
            return err
        }
        // 写入etcd
        err = s.cli.Save(k, string(v))
        if err != nil {
   
   
            return err
        }
    }
    return nil
}

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

key规则:

  • 前缀/querycoord-partition-loadinfo/{collectionID}/{partitionID}

已load的partition会产生这种类型的key,即如果有这种类型的key,就可以判断哪些partition已经load。

使用etcd-manager查看etcd:

partitionloadinfo-key.jpg

总结:

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

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

  • 前缀/querycoord-partition-loadinfo/{collectionID}/{partitionID}
目录
相关文章
|
11天前
|
JSON API 数据安全/隐私保护
Python采集淘宝评论API接口及JSON数据返回全流程指南
Python采集淘宝评论API接口及JSON数据返回全流程指南
JSON 安全 API
46 0
JSON 监控 API
29 0
|
20天前
|
存储 域名解析 弹性计算
阿里云上云流程参考:云服务器+域名+备案+域名解析绑定,全流程图文详解
对于初次通过阿里云完成上云的企业和个人用户来说,很多用户不仅是需要选购云服务器,同时还需要注册域名以及完成备案和域名的解析相关流程,从而实现网站的上线。本文将以上云操作流程为核心,结合阿里云的活动政策与用户系统梳理云服务器选购、域名注册、备案申请及域名绑定四大关键环节,以供用户完成线上业务部署做出参考。
|
20天前
|
人工智能 API 开发者
图文教程:阿里云百炼API-KEY获取方法,亲测全流程
本文详细介绍了如何获取阿里云百炼API-KEY,包含完整流程与截图指引。需先开通百炼平台及大模型服务,再通过控制台创建并复制API-KEY。目前平台提供千万tokens免费额度,适合开发者快速上手使用。
401 5
|
1月前
|
JSON 缓存 供应链
API 接口驱动 1688 采购自动化:从商品获取到下单支付的全流程贯通
在B2B电商采购中,1688开放平台通过API实现商品筛选、比价、下单、支付及物流跟踪的全流程自动化,大幅提升采购效率,降低人工成本与错误率。企业可无缝对接ERP系统,实现数据驱动决策,显著优化采购周期、成本与风险管控,助力数字化转型。
|
1月前
|
JSON 自然语言处理 供应链
API接口赋能1688采购全流程:从商品获取到下单支付一键贯通
1688采购API助力企业实现全流程自动化,涵盖商品数据获取、智能比价、一键下单、支付及物流跟踪等环节,显著提升采购效率,降低成本与风险,推动B2B采购模式智能化升级。
|
1月前
|
缓存 Java API
Spring WebFlux 2025 实操指南详解高性能非阻塞 API 开发全流程核心技巧
本指南基于Spring WebFlux 2025最新技术栈,详解如何构建高性能非阻塞API。涵盖环境搭建、响应式数据访问、注解与函数式两种API开发模式、响应式客户端使用、测试方法及性能优化技巧,助你掌握Spring WebFlux全流程开发核心实践。
397 0
|
2月前
|
存储 JSON 关系型数据库
【干货满满】解密 API 数据解析:从 JSON 到数据库存储的完整流程
本文详解电商API开发中JSON数据解析与数据库存储的全流程,涵盖数据提取、清洗、转换及优化策略,结合Python实战代码与主流数据库方案,助开发者构建高效、可靠的数据处理管道。
|
2月前
|
存储 监控 API
零基础 3 天搞定京东 / 淘宝 API 开发,从注册到调通接口全流程拆解
本文详解京东/淘宝API开发入门,涵盖账号注册、应用创建、签名生成及实战项目,助零基础开发者3天掌握电商API调用,实现商品数据获取与价格监控。

推荐镜像

更多
  • DNS