milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix bug for get index state (#22396) 

Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
pull/22455/head
cai.zhang 2023-02-28 10:29:50 +08:00 committed by GitHub
parent 1779d25e9c
commit eefa29a5e8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 77 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
internal/datacoord/index_service.go
View File

@ -52,7 +52,7 @@ func (s *Server) createIndexForSegment(segment *SegmentInfo, indexID UniqueID) e
NumRows: segment.NumOfRows,
IndexID: indexID,
BuildID: buildID,
CreateTime: segment.LastExpireTime,
CreateTime: uint64(segment.ID),
WriteHandoff: false,
}
if err = s.meta.AddSegmentIndex(segIndex); err != nil {
@ -255,7 +255,7 @@ func (s *Server) GetIndexState(ctx context.Context, req *indexpb.GetIndexStateRe
IndexStateFailReason: "",
}
s.completeIndexInfo(indexInfo, indexes[0], s.meta.SelectSegments(func(info *SegmentInfo) bool {
return isSegmentHealthy(info) && info.CollectionID == req.GetCollectionID()
return isFlush(info) && info.CollectionID == req.GetCollectionID()
}))
ret.State = indexInfo.State
ret.FailReason = indexInfo.IndexStateFailReason
@ -341,6 +341,7 @@ func (s *Server) completeIndexInfo(indexInfo *indexpb.IndexInfo, index *model.In
switch segIdx.IndexState {
case commonpb.IndexState_IndexStateNone:
// can't to here
log.Warn("receive unexpected index state: IndexStateNone", zap.Int64("segmentID", segIdx.SegmentID))
cntNone++
case commonpb.IndexState_Unissued:
cntUnissued++
@ -355,17 +356,18 @@ func (s *Server) completeIndexInfo(indexInfo *indexpb.IndexInfo, index *model.In
}
}
allCnt := len(segments)
indexInfo.TotalRows = totalRows
indexInfo.IndexedRows = indexedRows
switch {
case cntFailed > 0:
indexInfo.State = commonpb.IndexState_Failed
indexInfo.IndexStateFailReason = failReason
case cntFinished == allCnt:
indexInfo.State = commonpb.IndexState_Finished
default:
case cntInProgress > 0 || cntUnissued > 0:
indexInfo.State = commonpb.IndexState_InProgress
case cntNone > 0:
indexInfo.State = commonpb.IndexState_IndexStateNone
default:
indexInfo.State = commonpb.IndexState_Finished
}
log.Info("completeIndexInfo success", zap.Int64("collID", index.CollectionID), zap.Int64("indexID", index.IndexID),
@ -418,7 +420,7 @@ func (s *Server) GetIndexBuildProgress(ctx context.Context, req *indexpb.GetInde
State: 0,
}
s.completeIndexInfo(indexInfo, indexes[0], s.meta.SelectSegments(func(info *SegmentInfo) bool {
return isSegmentHealthy(info) && info.CollectionID == req.GetCollectionID()
return isFlush(info) && info.CollectionID == req.GetCollectionID()
}))
log.Info("GetIndexBuildProgress success", zap.Int64("collectionID", req.GetCollectionID()),
zap.String("indexName", req.GetIndexName()))

68
internal/datacoord/index_service_test.go
View File

@ -238,6 +238,74 @@ func TestServer_GetIndexState(t *testing.T) {
resp, err := s.GetIndexState(ctx, req)
assert.NoError(t, err)
assert.Equal(t, commonpb.ErrorCode_Success, resp.GetStatus().GetErrorCode())
assert.Equal(t, commonpb.IndexState_InProgress, resp.GetState())
})
s.meta = &meta{
catalog: &datacoord.Catalog{MetaKv: mocks.NewMetaKv(t)},
indexes: map[UniqueID]map[UniqueID]*model.Index{
collID: {
indexID: {
TenantID: "",
CollectionID: collID,
FieldID: fieldID,
IndexID: indexID,
IndexName: indexName,
IsDeleted: false,
CreateTime: createTS,
TypeParams: typeParams,
IndexParams: indexParams,
IsAutoIndex: false,
UserIndexParams: nil,
},
},
},
segments: &SegmentsInfo{map[UniqueID]*SegmentInfo{
segID: {
SegmentInfo: &datapb.SegmentInfo{
ID: segID,
CollectionID: collID,
PartitionID: partID,
InsertChannel: "",
NumOfRows: 10250,
State: commonpb.SegmentState_Flushed,
MaxRowNum: 65536,
LastExpireTime: createTS - 1,
},
segmentIndexes: map[UniqueID]*model.SegmentIndex{
indexID: {
SegmentID: segID,
CollectionID: collID,
PartitionID: partID,
NumRows: 3000,
IndexID: indexID,
BuildID: buildID,
NodeID: 0,
IndexVersion: 1,
IndexState: commonpb.IndexState_IndexStateNone,
FailReason: "",
IsDeleted: false,
CreateTime: 0,
IndexFileKeys: nil,
IndexSize: 0,
WriteHandoff: false,
},
},
currRows: 0,
allocations: nil,
lastFlushTime: time.Time{},
isCompacting: false,
size: 0,
lastWrittenTime: time.Time{},
},
}},
}
t.Run("index state is node", func(t *testing.T) {
resp, err := s.GetIndexState(ctx, req)
assert.NoError(t, err)
assert.Equal(t, commonpb.ErrorCode_Success, resp.GetStatus().GetErrorCode())
assert.Equal(t, commonpb.IndexState_IndexStateNone, resp.GetState())
})
t.Run("ambiguous index name", func(t *testing.T) {