CreatePartition API执行流程_milvus源码解析

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

CreatePartition API执行流程源码解析

milvus版本:v2.3.2

整体架构:

architecture.png

CreatePartition 的数据流向:

create_partition数据流向.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")
AI 代码解读

客户端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 {
   
   
        ......
    }

    ......
}
AI 代码解读

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
}
AI 代码解读

从代码可以看出调用了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 {
   
   
        ......
    }

    ......
}
AI 代码解读

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)
}
AI 代码解读

创建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
}
AI 代码解读

在这里重点研究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()
AI 代码解读

create_partition堆栈.jpg

在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,
}
AI 代码解读

partitionInfo2.jpg

可以看出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)
    }

    ......
}
AI 代码解读

跟踪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:

partition.jpg

总结:

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

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

  • 前缀/root-coord/partitions/{collectionID}/{partitionID}
  • 前缀/snapshots/root-coord/partitions/{collectionID}/{partitionID}_ts{时间戳}
目录
打赏
0
0
0
0
25
分享
相关文章
「ximagine」业余爱好者的非专业显示器测试流程规范,同时也是本账号输出内容的数据来源!如何测试显示器?荒岛整理总结出多种测试方法和注意事项,以及粗浅的原理解析!
本期内容为「ximagine」频道《显示器测试流程》的规范及标准,我们主要使用Calman、DisplayCAL、i1Profiler等软件及CA410、Spyder X、i1Pro 2等设备,是我们目前制作内容数据的重要来源,我们深知所做的仍是比较表面的活儿,和工程师、科研人员相比有着不小的差距,测试并不复杂,但是相当繁琐,收集整理测试无不花费大量时间精力,内容不完善或者有错误的地方,希望大佬指出我们好改进!
78 16
「ximagine」业余爱好者的非专业显示器测试流程规范,同时也是本账号输出内容的数据来源!如何测试显示器?荒岛整理总结出多种测试方法和注意事项,以及粗浅的原理解析!
详细介绍SpringBoot启动流程及配置类解析原理
通过对 Spring Boot 启动流程及配置类解析原理的深入分析,我们可以看到 Spring Boot 在启动时的灵活性和可扩展性。理解这些机制不仅有助于开发者更好地使用 Spring Boot 进行应用开发,还能够在面对问题时,迅速定位和解决问题。希望本文能为您在 Spring Boot 开发过程中提供有效的指导和帮助。
46 12
企业级API集成方案:基于阿里云函数计算调用DeepSeek全解析
DeepSeek R1 是一款先进的大规模深度学习模型,专为自然语言处理等复杂任务设计。它具备高效的架构、强大的泛化能力和优化的参数管理,适用于文本生成、智能问答、代码生成和数据分析等领域。阿里云平台提供了高性能计算资源、合规与数据安全、低延迟覆盖和成本效益等优势,支持用户便捷部署和调用 DeepSeek R1 模型,确保快速响应和稳定服务。通过阿里云百炼模型服务,用户可以轻松体验满血版 DeepSeek R1,并享受免费试用和灵活的API调用方式。
125 12
小红书笔记详情 API 接口:获取、应用与收益全解析
小红书(RED)是国内领先的生活方式分享平台,汇聚大量用户生成内容(UGC),尤以“种草”笔记闻名。小红书笔记详情API接口为开发者提供了获取笔记详细信息的强大工具,包括标题、内容、图片、点赞数等。通过注册开放平台账号、申请API权限并调用接口,开发者可构建内容分析工具、笔记推荐系统、数据爬虫等应用,提升用户体验和运营效率,创造新的商业模式。本文将详细介绍该API的获取、应用及潜在收益,并附上代码示例。
163 13
使用 API 接口获取京东商品详情全解析
京东作为头部电商平台,其商品数据极具价值。开发者可通过API接口获取商品详情、订单数据等信息,满足各种业务需求。使用前需注册账号并创建应用获取App Key和App Secret。调用流程包括认证授权、构建请求、发送请求及处理响应。注意事项包括遵守平台规则、控制调用频率和确保数据时效性。通过这些步骤,可为电商数据分析提供有力支持。
新手上云教程参考:阿里云服务器租用、域名注册、备案及域名解析流程图文教程
对于想要在阿里云上搭建网站或应用的用户来说,购买阿里云服务器和注册域名,绑定以及备案的流程至关重要。本文将以图文形式为您介绍阿里云服务器购买、域名注册、备案及绑定的全流程,以供参考,帮助用户轻松上手。
速卖通商品详情接口(速卖通API系列)
速卖通(AliExpress)是阿里巴巴旗下的跨境电商平台,提供丰富的商品数据。通过速卖通开放平台(AliExpress Open API),开发者可获取商品详情、订单管理等数据。主要功能包括商品搜索、商品详情、订单管理和数据报告。商品详情接口aliexpress.affiliate.productdetail.get用于获取商品标题、价格、图片等详细信息。开发者需注册账号并创建应用以获取App Key和App Secret,使用PHP等语言调用API。该接口支持多种请求参数和返回字段,方便集成到各类电商应用中。
以项目登录接口为例-大前端之开发postman请求接口带token的请求测试-前端开发必学之一-如果要学会联调接口而不是纯写静态前端页面-这个是必学-本文以优雅草蜻蜓Q系统API为实践来演示我们如何带token请求接口-优雅草卓伊凡
以项目登录接口为例-大前端之开发postman请求接口带token的请求测试-前端开发必学之一-如果要学会联调接口而不是纯写静态前端页面-这个是必学-本文以优雅草蜻蜓Q系统API为实践来演示我们如何带token请求接口-优雅草卓伊凡
49 5
以项目登录接口为例-大前端之开发postman请求接口带token的请求测试-前端开发必学之一-如果要学会联调接口而不是纯写静态前端页面-这个是必学-本文以优雅草蜻蜓Q系统API为实践来演示我们如何带token请求接口-优雅草卓伊凡
亚马逊商品详情接口(亚马逊 API 系列)
亚马逊作为全球最大的电商平台之一,提供了丰富的商品资源。开发者和电商从业者可通过亚马逊商品详情接口获取商品的描述、价格、评论、排名等数据,对市场分析、竞品研究、价格监控及业务优化具有重要价值。接口基于MWS服务,支持HTTP/HTTPS协议,需注册并获得API权限。Python示例展示了如何使用mws库调用接口获取商品详情。应用场景包括价格监控、市场调研、智能选品、用户推荐和库存管理等,助力电商运营和决策。
80 23

推荐镜像

更多
AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等