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

fix: Add IndexList check for load segment request (#28601) 

See also #28022 #28034
The load segment may reaches before watch dml channel, so the index meta
may be empty as well

---------

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
pull/28702/head
congqixia 2023-11-23 19:42:30 +08:00 committed by GitHub
parent a2fe9dad49
commit 098c1c926d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 81 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
internal/querynodev2/services.go
View File

@ -446,6 +446,12 @@ func (node *QueryNode) LoadSegments(ctx context.Context, req *querypb.LoadSegmen
return merr.Status(err), nil
}
// check index
if len(req.GetIndexInfoList()) == 0 {
err := merr.WrapErrIndexNotFoundForCollection(req.GetSchema().GetName())
return merr.Status(err), nil
}
// Delegates request to workers
if req.GetNeedTransfer() {
delegator, ok := node.delegators.Get(segment.GetInsertChannel())

127
internal/querynodev2/services_test.go
View File

@ -567,6 +567,9 @@ func (suite *ServiceSuite) TestLoadSegments_Int64() {
Schema: schema,
DeltaPositions: []*msgpb.MsgPosition{{Timestamp: 20000}},
NeedTransfer: true,
IndexInfoList: []*indexpb.IndexInfo{
{},
},
}
// LoadSegment
@ -603,6 +606,7 @@ func (suite *ServiceSuite) TestLoadSegments_VarChar() {
DeltaPositions: []*msgpb.MsgPosition{{Timestamp: 20000}},
NeedTransfer: true,
LoadMeta: loadMeta,
IndexInfoList: []*indexpb.IndexInfo{{}},
}
// LoadSegment
@ -622,12 +626,13 @@ func (suite *ServiceSuite) TestLoadDeltaInt64() {
MsgID: rand.Int63(),
TargetID: suite.node.session.ServerID,
},
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: suite.genSegmentLoadInfos(schema),
Schema: schema,
NeedTransfer: true,
LoadScope: querypb.LoadScope_Delta,
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: suite.genSegmentLoadInfos(schema),
Schema: schema,
NeedTransfer: true,
LoadScope: querypb.LoadScope_Delta,
IndexInfoList: []*indexpb.IndexInfo{{}},
}
// LoadSegment
@ -646,12 +651,13 @@ func (suite *ServiceSuite) TestLoadDeltaVarchar() {
MsgID: rand.Int63(),
TargetID: suite.node.session.ServerID,
},
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: suite.genSegmentLoadInfos(schema),
Schema: schema,
NeedTransfer: true,
LoadScope: querypb.LoadScope_Delta,
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: suite.genSegmentLoadInfos(schema),
Schema: schema,
NeedTransfer: true,
LoadScope: querypb.LoadScope_Delta,
IndexInfoList: []*indexpb.IndexInfo{{}},
}
// LoadSegment
@ -681,12 +687,13 @@ func (suite *ServiceSuite) TestLoadIndex_Success() {
MsgID: rand.Int63(),
TargetID: suite.node.session.ServerID,
},
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: rawInfo,
Schema: schema,
NeedTransfer: false,
LoadScope: querypb.LoadScope_Full,
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: rawInfo,
Schema: schema,
NeedTransfer: false,
LoadScope: querypb.LoadScope_Full,
IndexInfoList: []*indexpb.IndexInfo{{}},
}
// Load segment
@ -705,12 +712,13 @@ func (suite *ServiceSuite) TestLoadIndex_Success() {
MsgID: rand.Int63(),
TargetID: suite.node.session.ServerID,
},
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: infos,
Schema: schema,
NeedTransfer: false,
LoadScope: querypb.LoadScope_Index,
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: infos,
Schema: schema,
NeedTransfer: false,
LoadScope: querypb.LoadScope_Index,
IndexInfoList: []*indexpb.IndexInfo{{}},
}
// Load segment
@ -749,12 +757,13 @@ func (suite *ServiceSuite) TestLoadIndex_Failed() {
MsgID: rand.Int63(),
TargetID: suite.node.session.ServerID,
},
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: rawInfo,
Schema: schema,
NeedTransfer: false,
LoadScope: querypb.LoadScope_Index,
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: rawInfo,
Schema: schema,
NeedTransfer: false,
LoadScope: querypb.LoadScope_Index,
IndexInfoList: []*indexpb.IndexInfo{{}},
}
// Load segment
@ -781,12 +790,13 @@ func (suite *ServiceSuite) TestLoadIndex_Failed() {
MsgID: rand.Int63(),
TargetID: suite.node.session.ServerID,
},
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: infos,
Schema: schema,
NeedTransfer: false,
LoadScope: querypb.LoadScope_Index,
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: infos,
Schema: schema,
NeedTransfer: false,
LoadScope: querypb.LoadScope_Index,
IndexInfoList: []*indexpb.IndexInfo{{}},
}
// Load segment
@ -810,6 +820,9 @@ func (suite *ServiceSuite) TestLoadSegments_Failed() {
Infos: suite.genSegmentLoadInfos(schema),
Schema: schema,
NeedTransfer: true,
IndexInfoList: []*indexpb.IndexInfo{
{},
},
}
// Delegator not found
@ -817,10 +830,18 @@ func (suite *ServiceSuite) TestLoadSegments_Failed() {
suite.NoError(err)
suite.ErrorIs(merr.Error(status), merr.ErrChannelNotFound)
// IndexIndex not found
nonIndexReq := typeutil.Clone(req)
nonIndexReq.IndexInfoList = nil
status, err = suite.node.LoadSegments(ctx, nonIndexReq)
suite.NoError(err)
suite.ErrorIs(merr.Error(status), merr.ErrIndexNotFound)
// target not match
req.Base.TargetID = -1
status, err = suite.node.LoadSegments(ctx, req)
suite.NoError(err)
suite.T().Log(merr.Error(status))
suite.ErrorIs(merr.Error(status), merr.ErrNodeNotMatch)
// node not healthy
@ -846,11 +867,12 @@ func (suite *ServiceSuite) TestLoadSegments_Transfer() {
MsgID: rand.Int63(),
TargetID: suite.node.session.ServerID,
},
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: suite.genSegmentLoadInfos(schema),
Schema: schema,
NeedTransfer: true,
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: suite.genSegmentLoadInfos(schema),
Schema: schema,
NeedTransfer: true,
IndexInfoList: []*indexpb.IndexInfo{{}},
}
// LoadSegment
@ -867,11 +889,12 @@ func (suite *ServiceSuite) TestLoadSegments_Transfer() {
MsgID: rand.Int63(),
TargetID: suite.node.session.ServerID,
},
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: suite.genSegmentLoadInfos(schema),
Schema: schema,
NeedTransfer: true,
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: suite.genSegmentLoadInfos(schema),
Schema: schema,
NeedTransfer: true,
IndexInfoList: []*indexpb.IndexInfo{{}},
}
// LoadSegment
@ -893,14 +916,14 @@ func (suite *ServiceSuite) TestLoadSegments_Transfer() {
MsgID: rand.Int63(),
TargetID: suite.node.session.ServerID,
},
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: suite.genSegmentLoadInfos(schema),
Schema: schema,
NeedTransfer: true,
CollectionID: suite.collectionID,
DstNodeID: suite.node.session.ServerID,
Infos: suite.genSegmentLoadInfos(schema),
Schema: schema,
NeedTransfer: true,
IndexInfoList: []*indexpb.IndexInfo{{}},
}
// LoadSegment
// LoadSegment
status, err := suite.node.LoadSegments(ctx, req)
suite.NoError(err)