CreateIndex API执行流程_milvus源码解析2

简介: CreateIndex API执行流程_milvus源码解析2

CreateIndex API执行流程源码解析2

milvus版本:v2.3.2

上一篇介绍了CreateIndex对etcd元数据的操作,这里介绍另一个操作。

整体架构:

architecture.png

CreateIndex 的数据流向:

create_index数据流向3.jpg

1.dataCoord执行CreateIndex。

func (s *Server) CreateIndex(ctx context.Context, req *indexpb.CreateIndexRequest) (*commonpb.Status, error) {
   
   
    ......
    // 分配indexID,indexID=0
    indexID, err := s.meta.CanCreateIndex(req)
    ......

    err = s.meta.CreateIndex(index)
    ......

    // 将collectionID发送到channel,其它的goroutine进行消费。
    select {
   
   
    case s.notifyIndexChan <- req.GetCollectionID():
    default:
    }

    ......
}

上一篇已经分析过s.meta.CreateIndex(),这里重点分析s.notifyIndexChan。将collectionID发送到channel,其它的goroutine进行消费。

2.消费notifyIndexChan也在dataCoord。

代码路径:internal\datacoord\index_service.go

启动dataCoord服务的时候会调用此方法。

func (s *Server) createIndexForSegmentLoop(ctx context.Context) {
   
   
    log.Info("start create index for segment loop...")
    defer s.serverLoopWg.Done()

    ticker := time.NewTicker(time.Minute)
    defer ticker.Stop()
    for {
   
   
        select {
   
   
        case <-ctx.Done():
            log.Warn("DataCoord context done, exit...")
            return
        case <-ticker.C:
            segments := s.meta.GetHasUnindexTaskSegments()
            for _, segment := range segments {
   
   
                if err := s.createIndexesForSegment(segment); err != nil {
   
   
                    log.Warn("create index for segment fail, wait for retry", zap.Int64("segmentID", segment.ID))
                    continue
                }
            }
        case collectionID := <-s.notifyIndexChan:
            log.Info("receive create index notify", zap.Int64("collectionID", collectionID))
            // 获取collection的segment信息
            segments := s.meta.SelectSegments(func(info *SegmentInfo) bool {
   
   
                return isFlush(info) && collectionID == info.CollectionID
            })
            for _, segment := range segments {
   
   
                if err := s.createIndexesForSegment(segment); err != nil {
   
   
                    log.Warn("create index for segment fail, wait for retry", zap.Int64("segmentID", segment.ID))
                    continue
                }
            }
        case segID := <-s.buildIndexCh:
            log.Info("receive new flushed segment", zap.Int64("segmentID", segID))
            segment := s.meta.GetSegment(segID)
            if segment == nil {
   
   
                log.Warn("segment is not exist, no need to build index", zap.Int64("segmentID", segID))
                continue
            }
            if err := s.createIndexesForSegment(segment); err != nil {
   
   
                log.Warn("create index for segment fail, wait for retry", zap.Int64("segmentID", segment.ID))
                continue
            }
        }
    }
}

走case collectionID := <-s.notifyIndexChan这个分支。

3.进入s.createIndexesForSegment()

func (s *Server) createIndexesForSegment(segment *SegmentInfo) error {
   
   
    // 要创建的索引的信息:索引名称,维度等信息
    indexes := s.meta.GetIndexesForCollection(segment.CollectionID, "")
    for _, index := range indexes {
   
   
        if _, ok := segment.segmentIndexes[index.IndexID]; !ok {
   
   
            if err := s.createIndexForSegment(segment, index.IndexID); err != nil {
   
   
                log.Warn("create index for segment fail", zap.Int64("segmentID", segment.ID),
                    zap.Int64("indexID", index.IndexID))
                return err
            }
        }
    }
    return nil
}

4.进入s.createIndexForSegment()

在segment上建立索引。

func (s *Server) createIndexForSegment(segment *SegmentInfo, indexID UniqueID) error {
   
   
    log.Info("create index for segment", zap.Int64("segmentID", segment.ID), zap.Int64("indexID", indexID))
    buildID, err := s.allocator.allocID(context.Background())
    if err != nil {
   
   
        return err
    }
    segIndex := &model.SegmentIndex{
   
   
        SegmentID:    segment.ID,
        CollectionID: segment.CollectionID,
        PartitionID:  segment.PartitionID,
        NumRows:      segment.NumOfRows,
        IndexID:      indexID,
        BuildID:      buildID,
        CreateTime:   uint64(segment.ID),
        WriteHandoff: false,
    }
    // 添加segment-index
    // 前缀/segment-index/{collectionID}/{partitionID}/{segmentID}/{buildID}
    if err = s.meta.AddSegmentIndex(segIndex); err != nil {
   
   
        return err
    }
    // 入队,通知IndexNode执行索引创建任务
    s.indexBuilder.enqueue(buildID)
    return nil
}

segIndex变量:

segIndex.jpg

s.meta.AddSegmentIndex(segIndex)在etcd添加segment-index。

==前缀/segment-index/{collectionID}/{partitionID}/{segmentID}/{buildID}==

func (kc *Catalog) CreateSegmentIndex(ctx context.Context, segIdx *model.SegmentIndex) error {
   
   
    // key规则
    key := BuildSegmentIndexKey(segIdx.CollectionID, segIdx.PartitionID, segIdx.SegmentID, segIdx.BuildID)

    value, err := proto.Marshal(model.MarshalSegmentIndexModel(segIdx))
    if err != nil {
   
   
        return err
    }
    // 写入etcd
    err = kc.MetaKv.Save(key, string(value))
    if err != nil {
   
   
        log.Error("failed to save segment index meta in etcd", zap.Int64("buildID", segIdx.BuildID),
            zap.Int64("segmentID", segIdx.SegmentID), zap.Error(err))
        return err
    }
    return nil
}

5.进入s.indexBuilder.enqueue(buildID)

此函数作用:入队,通知IndexNode执行索引创建任务。

create_index数据流向2.jpg

6.进入process()

func (ib *indexBuilder) process(buildID UniqueID) bool {
   
   
    ......
    // 有一个定时器,默认1秒,配置项indexCoord.scheduler.interval
    switch state {
   
   
    case indexTaskInit:
        segment := ib.meta.GetSegment(meta.SegmentID)
        if !isSegmentHealthy(segment) || !ib.meta.IsIndexExist(meta.CollectionID, meta.IndexID) {
   
   
            log.Ctx(ib.ctx).Info("task is no need to build index, remove it", zap.Int64("buildID", buildID))
            if err := ib.meta.DeleteTask(buildID); err != nil {
   
   
                log.Ctx(ib.ctx).Warn("IndexCoord delete index failed", zap.Int64("buildID", buildID), zap.Error(err))
                return false
            }
            deleteFunc(buildID)
            return true
        }
        indexParams := ib.meta.GetIndexParams(meta.CollectionID, meta.IndexID)
        if isFlatIndex(getIndexType(indexParams)) || meta.NumRows < Params.DataCoordCfg.MinSegmentNumRowsToEnableIndex.GetAsInt64() {
   
   
            // 数量小于最低数量无需建立索引,默认为1024
            log.Ctx(ib.ctx).Info("segment does not need index really", zap.Int64("buildID", buildID),
                zap.Int64("segmentID", meta.SegmentID), zap.Int64("num rows", meta.NumRows))
            if err := ib.meta.FinishTask(&indexpb.IndexTaskInfo{
   
   
                BuildID:        buildID,
                State:          commonpb.IndexState_Finished,
                IndexFileKeys:  nil,
                SerializedSize: 0,
                FailReason:     "",
            }); err != nil {
   
   
                log.Ctx(ib.ctx).Warn("IndexCoord update index state fail", zap.Int64("buildID", buildID), zap.Error(err))
                return false
            }
            updateStateFunc(buildID, indexTaskDone)
            return true
        }
        // peek client
        // if all IndexNodes are executing task, wait for one of them to finish the task.
        nodeID, client := ib.nodeManager.PeekClient(meta)
        if client == nil {
   
   
            log.Ctx(ib.ctx).WithRateGroup("dc.indexBuilder", 1, 60).RatedInfo(5, "index builder peek client error, there is no available")
            return false
        }
        // update version and set nodeID
        if err := ib.meta.UpdateVersion(buildID, nodeID); err != nil {
   
   
            log.Ctx(ib.ctx).Warn("index builder update index version failed", zap.Int64("build", buildID), zap.Error(err))
            return false
        }
        // binLogs为向量数据文件的位置信息
        // files/insert_log/444517122896489678/444517122896489679/444517122896489694/102/444517122896236031
        // files/insert_log/{collectionID}/{partitionID}/{segmentID}/{fieldID}/444517122896236031
        binLogs := make([]string, 0)
        fieldID := ib.meta.GetFieldIDByIndexID(meta.CollectionID, meta.IndexID)
        for _, fieldBinLog := range segment.GetBinlogs() {
   
   
            if fieldBinLog.GetFieldID() == fieldID {
   
   
                for _, binLog := range fieldBinLog.GetBinlogs() {
   
   
                    binLogs = append(binLogs, binLog.LogPath)
                }
                break
            }
        }

        typeParams := ib.meta.GetTypeParams(meta.CollectionID, meta.IndexID)

        var storageConfig *indexpb.StorageConfig
        // 获取元数据存储类型:minio
        if Params.CommonCfg.StorageType.GetValue() == "local" {
   
   
            storageConfig = &indexpb.StorageConfig{
   
   
                RootPath:    Params.LocalStorageCfg.Path.GetValue(),
                StorageType: Params.CommonCfg.StorageType.GetValue(),
            }
        } else {
   
   
            storageConfig = &indexpb.StorageConfig{
   
   
                Address:          Params.MinioCfg.Address.GetValue(),
                AccessKeyID:      Params.MinioCfg.AccessKeyID.GetValue(),
                SecretAccessKey:  Params.MinioCfg.SecretAccessKey.GetValue(),
                UseSSL:           Params.MinioCfg.UseSSL.GetAsBool(),
                BucketName:       Params.MinioCfg.BucketName.GetValue(),
                RootPath:         Params.MinioCfg.RootPath.GetValue(),
                UseIAM:           Params.MinioCfg.UseIAM.GetAsBool(),
                IAMEndpoint:      Params.MinioCfg.IAMEndpoint.GetValue(),
                StorageType:      Params.CommonCfg.StorageType.GetValue(),
                Region:           Params.MinioCfg.Region.GetValue(),
                UseVirtualHost:   Params.MinioCfg.UseVirtualHost.GetAsBool(),
                CloudProvider:    Params.MinioCfg.CloudProvider.GetValue(),
                RequestTimeoutMs: Params.MinioCfg.RequestTimeoutMs.GetAsInt64(),
            }
        }
        req := &indexpb.CreateJobRequest{
   
   
            ClusterID:           Params.CommonCfg.ClusterPrefix.GetValue(),
            IndexFilePrefix:     path.Join(ib.chunkManager.RootPath(), common.SegmentIndexPath),
            BuildID:             buildID,
            DataPaths:           binLogs,
            IndexVersion:        meta.IndexVersion + 1,
            StorageConfig:       storageConfig,
            IndexParams:         indexParams,
            TypeParams:          typeParams,
            NumRows:             meta.NumRows,
            CurrentIndexVersion: ib.indexEngineVersionManager.GetCurrentIndexEngineVersion(),
        }
        // 通知IndexNode创建索引
        if err := ib.assignTask(client, req); err != nil {
   
   
            ......
        }
        ......
        // update index meta state to InProgress
        // 更新索引状态
        if err := ib.meta.BuildIndex(buildID); err != nil {
   
   
            ......
        }
        updateStateFunc(buildID, indexTaskInProgress)

    case indexTaskDone:
        ......

    default:
        ......
    }
    return true
}

ib.assignTask()通知indexNode创建索引。

7.进入ib.assignTask()

// assignTask sends the index task to the IndexNode, it has a timeout interval, if the IndexNode doesn't respond within
// the interval, it is considered that the task sending failed.
func (ib *indexBuilder) assignTask(builderClient types.IndexNodeClient, req *indexpb.CreateJobRequest) error {
   
   
    ......
    // rpc调用indexNode
    resp, err := builderClient.CreateJob(ctx, req)
    ......
}

8.进入builderClient.CreateJob()

代码路径:internal\indexnode\indexnode_service.go

func (i *IndexNode) CreateJob(ctx context.Context, req *indexpb.CreateJobRequest) (*commonpb.Status, error) {
   
   
    ......
    task := &indexBuildTask{
   
   
        ident:          fmt.Sprintf("%s/%d", req.ClusterID, req.BuildID),
        ctx:            taskCtx,
        cancel:         taskCancel,
        BuildID:        req.GetBuildID(),
        ClusterID:      req.GetClusterID(),
        node:           i,
        req:            req,
        cm:             cm,
        nodeID:         i.GetNodeID(),
        tr:             timerecord.NewTimeRecorder(fmt.Sprintf("IndexBuildID: %d, ClusterID: %s", req.BuildID, req.ClusterID)),
        serializedSize: 0,
    }
    ret := merr.Success()
    if err := i.sched.IndexBuildQueue.Enqueue(task); err != nil {
   
   
        ......
    }
    ......
}

task执行逻辑:

pipelines := []func(context.Context) error{
   
   t.Prepare, t.BuildIndex, t.SaveIndexFiles}

依次执行task的Prepare()、BuildIndex()、SaveIndexFiles()方法。

9.进入BuildIndex()

func (it *indexBuildTask) BuildIndex(ctx context.Context) error {
   
   
    ......
    var buildIndexInfo *indexcgowrapper.BuildIndexInfo
    buildIndexInfo, err = indexcgowrapper.NewBuildIndexInfo(it.req.GetStorageConfig())
    ......
    err = buildIndexInfo.AppendFieldMetaInfo(it.collectionID, it.partitionID, it.segmentID, it.fieldID, it.fieldType)
    ......

    err = buildIndexInfo.AppendIndexMetaInfo(it.req.IndexID, it.req.BuildID, it.req.IndexVersion)
    ......

    err = buildIndexInfo.AppendBuildIndexParam(it.newIndexParams)
    ......

    jsonIndexParams, err := json.Marshal(it.newIndexParams)
    ......

    err = buildIndexInfo.AppendBuildTypeParam(it.newTypeParams)
    ......

    for _, path := range it.req.GetDataPaths() {
   
   
        err = buildIndexInfo.AppendInsertFile(path)
        if err != nil {
   
   
            log.Ctx(ctx).Warn("append insert binlog path failed", zap.Error(err))
            return err
        }
    }

    it.currentIndexVersion = getCurrentIndexVersion(it.req.GetCurrentIndexVersion())
    if err := buildIndexInfo.AppendIndexEngineVersion(it.currentIndexVersion); err != nil {
   
   
        log.Ctx(ctx).Warn("append index engine version failed", zap.Error(err))
        return err
    }
    // cgo层调用CreateIndex()
    // it.index有个Upload()方法写入s3(SaveIndexFiles方法里调用it.index.Upload()写入s3)
    it.index, err = indexcgowrapper.CreateIndex(ctx, buildIndexInfo)
    if err != nil {
   
   
        ......
    }

    ......
}

indexcgowrapper.CreateIndex()调用C层创建索引。

10.进入SaveIndexFiles()

func (it *indexBuildTask) SaveIndexFiles(ctx context.Context) error {
   
   
    ......
    // c++层上传索引文件到s3
    indexFilePath2Size, err := it.index.UpLoad()
    ......

}

总结:

  • CreateIndex由proxy传递给协调器dataCoord操作etcd。
    • 前缀/field-index/{collectionID}/{IndexID}
    • 前缀/segment-index/{collectionID}/{partitionID}/{segmentID}/{buildID}
  • dataCoord通知indexNode创建索引,indexNode通过cgo调用C++层创建索引,并上传至s3。
目录
相关文章
|
8天前
|
C语言
内核源码中遇到不会解析的宏怎么办?
内核源码中遇到不会解析的宏怎么办?
189 1
|
10天前
|
Java 程序员 API
Java并发基础:concurrent Flow API全面解析
java.util.concurrent.Flow定义了响应式编程的核心接口,促进了Java在异步数据处理和背压机制方面的标准化,这使得第三方库如Reactor和RxJava能够基于这些接口提供丰富的实现和功能,同时简化了响应式编程在Java中的使用,Flow API增强了Java在并发编程领域的灵活性,使得处理异步数据流变得更加自然和高效。
Java并发基础:concurrent Flow API全面解析
|
21小时前
|
移动开发 API HTML5
HTML5响应式自动采集API壁纸系统源码自适应手机端
HTML5响应式自动采集API壁纸系统源码自适应手机端
23 11
HTML5响应式自动采集API壁纸系统源码自适应手机端
|
1天前
|
Java 关系型数据库 数据库连接
Spring源码解析--深入Spring事务原理
本文将带领大家领略Spring事务的风采,Spring事务是我们在日常开发中经常会遇到的,也是各种大小面试中的高频题,希望通过本文,能让大家对Spring事务有个深入的了解,无论开发还是面试,都不会让Spring事务成为拦路虎。
11 1
|
2天前
|
存储 Oracle Java
Java 包和 API 深度解析:组织代码,避免命名冲突
Java 中的包 用于将相关的类分组在一起。可以将其视为文件目录中的一个文件夹。我们使用包来避免名称冲突,并编写更易于维护的代码。 包分为两类: 内置包(来自 Java API 的包) 用户定义的包(创建自己的包)
174 2
|
2天前
|
API 开发工具 数据安全/隐私保护
API接口的对接流程和注意事项
随着互联网技术的发展和应用的普及,API接口已经成为不同系统、不同应用之间进行交互和数据交换的重要方式。API接口使得不同的系统能够互相调用对方的功能,提高了系统的灵活性和扩展性。但是,在进行API接口对接的过程中,需要注意一些流程和事项,以确保对接的顺利进行和系统的稳定运行。
|
3天前
|
数据挖掘 API 数据处理
获取商品详情信息API接口:1688开放平台功能解析
首先,要获取商品详情信息,开发者需要向1688开放平台申请相应的权限,并遵循其调用规则。在调用商品详情信息API接口时,需要传入相应的请求参数,如商品ID、调用时间戳、密钥等。这些参数对于确保接口的正常运行至关重要。
|
4天前
|
数据采集 安全 API
如何实时获取小红书笔记详情的API使用与解析
小红书是一个以分享消费经验、生活方式为主的社交平台,拥有大量的用户和内容。为了更好地了解用户在小红书上的行为和内容,许多开发者选择使用小红书开放平台提供的API接口。本文将介绍如何通过小红书笔记详情API实现实时数据获取,并给出相应的代码示例。
|
8天前
|
JSON 前端开发 JavaScript
从前端到后端——Web开发的全流程解析
【2月更文挑战第2天】Web开发涉及多个方面,从前端设计到后端实现,需要开发者具备一定的技术能力和知识储备。本文将以一个简单的Web应用为例,详细介绍Web开发的全流程。
|
2月前
ChatGLM2 源码解析:`GLMTransformer`
ChatGLM2 源码解析:`GLMTransformer`
162 0

相关产品

  • 云迁移中心
  • 推荐镜像

    更多