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

Improve the error message for getting all indexes of collection (#27389) 

Signed-off-by: yah01 <yah2er0ne@outlook.com>
pull/27540/head
yah01 2023-10-08 21:23:32 +08:00 committed by GitHub
parent e4f73cc805
commit 41495ed266
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 679 additions and 646 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
internal/datacoord/index_service.go
View File

@ -213,9 +213,9 @@ func (s *Server) CreateIndex(ctx context.Context, req *indexpb.CreateIndexReques
func (s *Server) GetIndexState(ctx context.Context, req *indexpb.GetIndexStateRequest) (*indexpb.GetIndexStateResponse, error) {
log := log.Ctx(ctx).With(
zap.Int64("collectionID", req.GetCollectionID()),
zap.String("indexName", req.GetIndexName()),
)
log.Info("receive GetIndexState request",
zap.String("indexName", req.GetIndexName()))
log.Info("receive GetIndexState request")
if err := merr.CheckHealthy(s.GetStateCode()); err != nil {
log.Warn(msgDataCoordIsUnhealthy(paramtable.GetNodeID()), zap.Error(err))
@ -227,8 +227,7 @@ func (s *Server) GetIndexState(ctx context.Context, req *indexpb.GetIndexStateRe
indexes := s.meta.GetIndexesForCollection(req.GetCollectionID(), req.GetIndexName())
if len(indexes) == 0 {
err := merr.WrapErrIndexNotFound(req.GetIndexName())
log.Warn("GetIndexState fail",
zap.String("indexName", req.GetIndexName()), zap.Error(err))
log.Warn("GetIndexState fail", zap.Error(err))
return &indexpb.GetIndexStateResponse{
Status: merr.Status(err),
}, nil
@ -245,12 +244,7 @@ func (s *Server) GetIndexState(ctx context.Context, req *indexpb.GetIndexStateRe
State: commonpb.IndexState_Finished,
}
indexInfo := &indexpb.IndexInfo{
IndexedRows: 0,
TotalRows: 0,
State: 0,
IndexStateFailReason: "",
}
indexInfo := &indexpb.IndexInfo{}
s.completeIndexInfo(indexInfo, indexes[0], s.meta.SelectSegments(func(info *SegmentInfo) bool {
return isFlush(info) && info.CollectionID == req.GetCollectionID()
}), false, indexes[0].CreateTime)
@ -258,7 +252,8 @@ func (s *Server) GetIndexState(ctx context.Context, req *indexpb.GetIndexStateRe
ret.FailReason = indexInfo.IndexStateFailReason
log.Info("GetIndexState success",
zap.String("IndexName", req.GetIndexName()), zap.String("state", ret.GetState().String()))
zap.String("state", ret.GetState().String()),
)
return ret, nil
}
@ -485,9 +480,11 @@ func (s *Server) GetIndexBuildProgress(ctx context.Context, req *indexpb.GetInde
func (s *Server) DescribeIndex(ctx context.Context, req *indexpb.DescribeIndexRequest) (*indexpb.DescribeIndexResponse, error) {
log := log.Ctx(ctx).With(
zap.Int64("collectionID", req.GetCollectionID()),
zap.String("indexName", req.GetIndexName()),
)
log.Info("receive DescribeIndex request",
zap.Uint64("timestamp", req.GetTimestamp()),
)
log.Info("receive DescribeIndex request", zap.String("indexName", req.GetIndexName()),
zap.Uint64("timestamp", req.GetTimestamp()))
if err := merr.CheckHealthy(s.GetStateCode()); err != nil {
log.Warn(msgDataCoordIsUnhealthy(paramtable.GetNodeID()), zap.Error(err))
@ -499,7 +496,7 @@ func (s *Server) DescribeIndex(ctx context.Context, req *indexpb.DescribeIndexRe
indexes := s.meta.GetIndexesForCollection(req.GetCollectionID(), req.GetIndexName())
if len(indexes) == 0 {
err := merr.WrapErrIndexNotFound(req.GetIndexName())
log.Warn("DescribeIndex fail", zap.String("indexName", req.GetIndexName()), zap.Error(err))
log.Warn("DescribeIndex fail", zap.Error(err))
return &indexpb.DescribeIndexResponse{
Status: merr.Status(err),
}, nil
@ -532,7 +529,7 @@ func (s *Server) DescribeIndex(ctx context.Context, req *indexpb.DescribeIndexRe
s.completeIndexInfo(indexInfo, index, segments, false, createTs)
indexInfos = append(indexInfos, indexInfo)
}
log.Info("DescribeIndex success", zap.String("indexName", req.GetIndexName()))
log.Info("DescribeIndex success")
return &indexpb.DescribeIndexResponse{
Status: merr.Status(nil),
IndexInfos: indexInfos,

51
internal/proxy/default_limit_reducer.go
View File

@ -3,7 +3,6 @@ package proxy
import (
"context"
"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
"github.com/milvus-io/milvus-proto/go-api/v2/milvuspb"
"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
"github.com/milvus-io/milvus/internal/proto/internalpb"
@ -39,42 +38,38 @@ func (r *defaultLimitReducer) afterReduce(result *milvuspb.QueryResults) error {
outputFieldsID := r.req.GetOutputFieldsId()
result.CollectionName = collectionName
var err error
if len(result.FieldsData) > 0 {
result.Status = merr.Status(nil)
} else {
result.Status = &commonpb.Status{
ErrorCode: commonpb.ErrorCode_EmptyCollection,
Reason: "empty collection", // TODO
}
return nil
}
for i := 0; i < len(result.FieldsData); i++ {
for i := 0; i < len(result.GetFieldsData()); i++ {
// drop ts column
if outputFieldsID[i] == common.TimeStampField {
result.FieldsData = append(result.FieldsData[:i], result.FieldsData[(i+1):]...)
outputFieldsID = append(outputFieldsID[:i], outputFieldsID[i+1:]...)
i--
continue
}
for _, field := range schema.Fields {
if field.FieldID == outputFieldsID[i] {
// deal with the situation that offset equal to or greater than the number of entities
if result.FieldsData[i] == nil {
var err error
result.FieldsData[i], err = typeutil.GenEmptyFieldData(field)
if err != nil {
return err
}
}
result.FieldsData[i].FieldName = field.Name
result.FieldsData[i].FieldId = field.FieldID
result.FieldsData[i].Type = field.DataType
result.FieldsData[i].IsDynamic = field.IsDynamic
}
field := typeutil.GetField(schema, outputFieldsID[i])
if field == nil {
err = merr.WrapErrFieldNotFound(outputFieldsID[i])
break
}
if result.FieldsData[i] == nil {
result.FieldsData[i], err = typeutil.GenEmptyFieldData(field)
if err != nil {
break
}
continue
}
result.FieldsData[i].FieldName = field.GetName()
result.FieldsData[i].FieldId = field.GetFieldID()
result.FieldsData[i].Type = field.GetDataType()
result.FieldsData[i].IsDynamic = field.GetIsDynamic()
}
return nil
result.Status = merr.Status(err)
return err
}
func newDefaultLimitReducer(ctx context.Context, params *queryParams, req *internalpb.RetrieveRequest, schema *schemapb.CollectionSchema, collectionName string) *defaultLimitReducer {

1184
internal/proxy/task.go
View File

File diff suppressed because it is too large Load Diff

12
internal/proxy/task_index.go
View File

@ -491,13 +491,19 @@ func (dit *describeIndexTask) Execute(ctx context.Context) error {
}
resp, err := dit.datacoord.DescribeIndex(ctx, &indexpb.DescribeIndexRequest{CollectionID: dit.collectionID, IndexName: dit.IndexName, Timestamp: dit.Timestamp})
if err != nil || resp == nil {
if err != nil {
return err
}
dit.result = &milvuspb.DescribeIndexResponse{}
dit.result.Status = resp.GetStatus()
if dit.result.GetStatus().GetErrorCode() != commonpb.ErrorCode_Success {
return merr.Error(dit.result.GetStatus())
err = merr.Error(resp.GetStatus())
if err != nil {
if errors.Is(err, merr.ErrIndexNotFound) && len(dit.GetIndexName()) == 0 {
err = merr.WrapErrIndexNotFoundForCollection(dit.GetCollectionName())
dit.result.Status = merr.Status(err)
}
return err
}
for _, indexInfo := range resp.IndexInfos {
field, err := schemaHelper.GetFieldFromID(indexInfo.FieldID)

3
pkg/util/merr/errors_test.go
View File

@ -109,6 +109,9 @@ func (s *ErrSuite) TestWrap() {
// Index related
s.ErrorIs(WrapErrIndexNotFound("failed to get Index"), ErrIndexNotFound)
s.ErrorIs(WrapErrIndexNotFoundForCollection("milvus_hello", "failed to get collection index"), ErrIndexNotFound)
s.ErrorIs(WrapErrIndexNotFoundForSegment(100, "failed to get collection index"), ErrIndexNotFound)
s.ErrorIs(WrapErrIndexNotSupported("wsnh", "failed to create index"), ErrIndexNotSupported)
// Node related
s.ErrorIs(WrapErrNodeNotFound(1, "failed to get node"), ErrNodeNotFound)

8
pkg/util/merr/utils.go
View File

@ -489,6 +489,14 @@ func WrapErrIndexNotFoundForSegment(segmentID int64, msg ...string) error {
return err
}
func WrapErrIndexNotFoundForCollection(collection string, msg ...string) error {
err := wrapWithField(ErrIndexNotFound, "collection", collection)
if len(msg) > 0 {
err = errors.Wrap(err, strings.Join(msg, "; "))
}
return err
}
func WrapErrIndexNotSupported(indexType string, msg ...string) error {
err := wrapWithField(ErrIndexNotSupported, "indexType", indexType)
if len(msg) > 0 {

33
pkg/util/typeutil/gen_empty_field_data.go
View File

@ -15,7 +15,8 @@ func genEmptyBoolFieldData(field *schemapb.FieldSchema) *schemapb.FieldData {
Data: &schemapb.ScalarField_BoolData{BoolData: &schemapb.BoolArray{Data: nil}},
},
},
FieldId: field.GetFieldID(),
FieldId: field.GetFieldID(),
IsDynamic: field.GetIsDynamic(),
}
}
@ -28,7 +29,8 @@ func genEmptyIntFieldData(field *schemapb.FieldSchema) *schemapb.FieldData {
Data: &schemapb.ScalarField_IntData{IntData: &schemapb.IntArray{Data: nil}},
},
},
FieldId: field.GetFieldID(),
FieldId: field.GetFieldID(),
IsDynamic: field.GetIsDynamic(),
}
}
@ -41,7 +43,8 @@ func genEmptyLongFieldData(field *schemapb.FieldSchema) *schemapb.FieldData {
Data: &schemapb.ScalarField_LongData{LongData: &schemapb.LongArray{Data: nil}},
},
},
FieldId: field.GetFieldID(),
FieldId: field.GetFieldID(),
IsDynamic: field.GetIsDynamic(),
}
}
@ -54,7 +57,8 @@ func genEmptyFloatFieldData(field *schemapb.FieldSchema) *schemapb.FieldData {
Data: &schemapb.ScalarField_FloatData{FloatData: &schemapb.FloatArray{Data: nil}},
},
},
FieldId: field.GetFieldID(),
FieldId: field.GetFieldID(),
IsDynamic: field.GetIsDynamic(),
}
}
@ -67,7 +71,8 @@ func genEmptyDoubleFieldData(field *schemapb.FieldSchema) *schemapb.FieldData {
Data: &schemapb.ScalarField_DoubleData{DoubleData: &schemapb.DoubleArray{Data: nil}},
},
},
FieldId: field.GetFieldID(),
FieldId: field.GetFieldID(),
IsDynamic: field.GetIsDynamic(),
}
}
@ -80,7 +85,8 @@ func genEmptyVarCharFieldData(field *schemapb.FieldSchema) *schemapb.FieldData {
Data: &schemapb.ScalarField_StringData{StringData: &schemapb.StringArray{Data: nil}},
},
},
FieldId: field.GetFieldID(),
FieldId: field.GetFieldID(),
IsDynamic: field.GetIsDynamic(),
}
}
@ -93,7 +99,8 @@ func genEmptyArrayFieldData(field *schemapb.FieldSchema) *schemapb.FieldData {
Data: &schemapb.ScalarField_ArrayData{ArrayData: &schemapb.ArrayArray{Data: nil}},
},
},
FieldId: field.GetFieldID(),
FieldId: field.GetFieldID(),
IsDynamic: field.GetIsDynamic(),
}
}
@ -106,7 +113,8 @@ func genEmptyJSONFieldData(field *schemapb.FieldSchema) *schemapb.FieldData {
Data: &schemapb.ScalarField_JsonData{JsonData: &schemapb.JSONArray{Data: nil}},
},
},
FieldId: field.GetFieldID(),
FieldId: field.GetFieldID(),
IsDynamic: field.GetIsDynamic(),
}
}
@ -126,7 +134,8 @@ func genEmptyBinaryVectorFieldData(field *schemapb.FieldSchema) (*schemapb.Field
},
},
},
FieldId: field.GetFieldID(),
FieldId: field.GetFieldID(),
IsDynamic: field.GetIsDynamic(),
}, nil
}
@ -146,7 +155,8 @@ func genEmptyFloatVectorFieldData(field *schemapb.FieldSchema) (*schemapb.FieldD
},
},
},
FieldId: field.GetFieldID(),
FieldId: field.GetFieldID(),
IsDynamic: field.GetIsDynamic(),
}, nil
}
@ -166,7 +176,8 @@ func genEmptyFloat16VectorFieldData(field *schemapb.FieldSchema) (*schemapb.Fiel
},
},
},
FieldId: field.GetFieldID(),
FieldId: field.GetFieldID(),
IsDynamic: field.GetIsDynamic(),
}, nil
}

7
pkg/util/typeutil/schema.go
View File

@ -23,6 +23,7 @@ import (
"unsafe"
"github.com/cockroachdb/errors"
"github.com/samber/lo"
"go.uber.org/zap"
"github.com/milvus-io/milvus-proto/go-api/v2/schemapb"
@ -833,6 +834,12 @@ func GetPrimaryFieldData(datas []*schemapb.FieldData, primaryFieldSchema *schema
return primaryFieldData, nil
}
func GetField(schema *schemapb.CollectionSchema, fieldID int64) *schemapb.FieldSchema {
return lo.FindOrElse(schema.GetFields(), nil, func(field *schemapb.FieldSchema) bool {
return field.GetFieldID() == fieldID
})
}
func IsPrimaryFieldDataExist(datas []*schemapb.FieldData, primaryFieldSchema *schemapb.FieldSchema) bool {
primaryFieldID := primaryFieldSchema.FieldID
primaryFieldName := primaryFieldSchema.Name