diff --git a/cmd/distributed/components/master_service.go b/cmd/distributed/components/master_service.go index b059a84631d0c13314387cf4730e993d6bff068a..9d90c7e139dc2ded1737d9c77d45203ad9540042 100644 --- a/cmd/distributed/components/master_service.go +++ b/cmd/distributed/components/master_service.go @@ -103,6 +103,9 @@ func NewMasterService(ctx context.Context) (*MasterService, error) { is.Params.Init() log.Printf("index service address : %s", is.Params.Address) indexService := isc.NewClient(is.Params.Address) + if err = indexService.Init(); err != nil { + return nil, err + } if err = svr.SetIndexService(indexService); err != nil { return nil, err diff --git a/internal/datanode/flow_graph_insert_buffer_node.go b/internal/datanode/flow_graph_insert_buffer_node.go index a9ff30b1e5579ef3da280d109194c42648ca388a..eac2177dfbef01b71f4d2832ae7cfffe552c6b57 100644 --- a/internal/datanode/flow_graph_insert_buffer_node.go +++ b/internal/datanode/flow_graph_insert_buffer_node.go @@ -456,6 +456,11 @@ func (ibNode *insertBufferNode) Operate(in []*Msg) []*Msg { continue } } + err := ibNode.completeFlush(currentSegID) + if err != nil { + log.Println(err) + } + log.Println("Flush completed") } } diff --git a/internal/indexservice/indexservice.go b/internal/indexservice/indexservice.go index 27730955f66a924fcdfba41710d031fc51fe4041..085908bf3aed0b6d9fa18828ae1ee42bb4a6c5c7 100644 --- a/internal/indexservice/indexservice.go +++ b/internal/indexservice/indexservice.go @@ -2,6 +2,7 @@ package indexservice import ( "context" + "fmt" "log" "sync" "time" @@ -177,6 +178,7 @@ func (i *ServiceImpl) GetStatisticsChannel() (string, error) { } func (i *ServiceImpl) BuildIndex(req *indexpb.BuildIndexRequest) (*indexpb.BuildIndexResponse, error) { + fmt.Println("builder building index ..., indexName = ", req.IndexName, "indexID = ", req.IndexID, "dataPath = ", req.DataPaths) ret := &indexpb.BuildIndexResponse{ Status: &commonpb.Status{ ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR, @@ -244,10 +246,13 @@ func (i *ServiceImpl) GetIndexStates(req *indexpb.IndexStatesRequest) (*indexpb. } func (i *ServiceImpl) GetIndexFilePaths(req *indexpb.IndexFilePathsRequest) (*indexpb.IndexFilePathsResponse, error) { - var indexPaths []*indexpb.IndexFilePathInfo + var indexPaths []*indexpb.IndexFilePathInfo = nil for _, indexID := range req.IndexBuildIDs { - indexPathInfo, _ := i.metaTable.GetIndexFilePathInfo(indexID) + indexPathInfo, err := i.metaTable.GetIndexFilePathInfo(indexID) + if err != nil { + return nil, err + } indexPaths = append(indexPaths, indexPathInfo) } diff --git a/internal/indexservice/node_mgr.go b/internal/indexservice/node_mgr.go index c29b2437c2673816b25a731ba4e7536b90bf7fff..76f8d7077f2ccd3296d805e6441a2b9c0fd5c86d 100644 --- a/internal/indexservice/node_mgr.go +++ b/internal/indexservice/node_mgr.go @@ -30,6 +30,10 @@ func (i *ServiceImpl) addNode(nodeID UniqueID, req *indexpb.RegisterNodeRequest) if err != nil { return err } + err = nodeClient.Init() + if err != nil { + return err + } item := &PQItem{ value: nodeClient, key: nodeID, diff --git a/internal/masterservice/task.go b/internal/masterservice/task.go index f144fefe03eb028aa40242d9def4e166f7bb1280..e7200ca7d75b988f3e6bd8a1867df2f93063e397 100644 --- a/internal/masterservice/task.go +++ b/internal/masterservice/task.go @@ -1,6 +1,7 @@ package masterservice import ( + "fmt" "log" "github.com/golang/protobuf/proto" @@ -597,6 +598,7 @@ func (t *CreateIndexReqTask) Execute() error { indexParams: t.Req.ExtraParams, } t.core.indexTaskQueue <- &task + fmt.Println("create index task enqueue, segID = ", seg) } return nil } diff --git a/internal/querynode/index_loader.go b/internal/querynode/index_loader.go index 08d867ec7fe096a6f59eccee7b37f74481915a5a..37e24abd9054d088b3d8022d238f5c01f8faf10d 100644 --- a/internal/querynode/index_loader.go +++ b/internal/querynode/index_loader.go @@ -323,7 +323,6 @@ func (loader *indexLoader) getIndexPaths(indexBuildID UniqueID) ([]string, error } indexFilePathRequest := &indexpb.IndexFilePathsRequest{ - // TODO: rename indexIDs to buildIDs IndexBuildIDs: []UniqueID{indexBuildID}, } pathResponse, err := loader.indexClient.GetIndexFilePaths(indexFilePathRequest) diff --git a/internal/querynode/load_service.go b/internal/querynode/load_service.go index 6540011370ad539958cc8d83e9e3c48366af88bf..b06f2c5037fe83ba3916bc7d8ed51c4665afc087 100644 --- a/internal/querynode/load_service.go +++ b/internal/querynode/load_service.go @@ -10,7 +10,7 @@ import ( "github.com/zilliztech/milvus-distributed/internal/msgstream" ) -const indexCheckInterval = 1 +const indexCheckInterval = 3 type loadService struct { ctx context.Context @@ -115,6 +115,7 @@ func (s *loadService) loadSegmentInternal(collectionID UniqueID, partitionID Uni return err } if errIndex == nil { + fmt.Println("loading index...") indexPaths, err := s.segLoader.indexLoader.getIndexPaths(buildID) if err != nil { return err