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

Fix shard leader misses to fill target ID while delegates query/search (#20973) 

Signed-off-by: yah01 <yang.cen@zilliz.com>

Signed-off-by: yah01 <yang.cen@zilliz.com>
pull/20981/head
yah01 2022-12-05 14:09:17 +08:00 committed by GitHub
parent 2791f5526c
commit 47b76e13be
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 36 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
internal/querynode/shard_cluster.go
View File

@ -954,8 +954,10 @@ func (sc *ShardCluster) Search(ctx context.Context, req *querypb.SearchRequest,
// dispatch request to followers
for nodeID, segments := range segAllocs {
internalReq := typeutil.Clone(req.GetReq())
internalReq.GetBase().TargetID = nodeID
nodeReq := &querypb.SearchRequest{
Req: req.Req,
Req: internalReq,
DmlChannels: req.DmlChannels,
FromShardLeader: true,
Scope: querypb.DataScope_Historical,
@ -1038,8 +1040,10 @@ func (sc *ShardCluster) Query(ctx context.Context, req *querypb.QueryRequest, wi
// dispatch request to followers
for nodeID, segments := range segAllocs {
internalReq := typeutil.Clone(req.GetReq())
internalReq.GetBase().TargetID = nodeID
nodeReq := &querypb.QueryRequest{
Req: req.Req,
Req: internalReq,
FromShardLeader: true,
SegmentIDs: segments,
Scope: querypb.DataScope_Historical,

30
internal/querynode/shard_cluster_test.go
View File

@ -1164,6 +1164,9 @@ func TestShardCluster_Search(t *testing.T) {
require.EqualValues(t, available, sc.state.Load())
result, err := sc.Search(ctx, &querypb.SearchRequest{
Req: &internalpb.SearchRequest{
Base: &commonpb.MsgBase{},
},
DmlChannels: []string{vchannelName},
}, streamingDoNothing)
assert.NoError(t, err)
@ -1215,6 +1218,9 @@ func TestShardCluster_Search(t *testing.T) {
require.EqualValues(t, available, sc.state.Load())
_, err := sc.Search(ctx, &querypb.SearchRequest{
Req: &internalpb.SearchRequest{
Base: &commonpb.MsgBase{},
},
DmlChannels: []string{vchannelName},
}, func(ctx context.Context) error { return errors.New("mocked") })
assert.Error(t, err)
@ -1273,6 +1279,9 @@ func TestShardCluster_Search(t *testing.T) {
require.EqualValues(t, available, sc.state.Load())
_, err := sc.Search(ctx, &querypb.SearchRequest{
Req: &internalpb.SearchRequest{
Base: &commonpb.MsgBase{},
},
DmlChannels: []string{vchannelName},
}, streamingDoNothing)
assert.Error(t, err)
@ -1325,6 +1334,9 @@ func TestShardCluster_Search(t *testing.T) {
require.EqualValues(t, available, sc.state.Load())
_, err := sc.Search(ctx, &querypb.SearchRequest{
Req: &internalpb.SearchRequest{
Base: &commonpb.MsgBase{},
},
DmlChannels: []string{vchannelName},
}, streamingDoNothing)
assert.Error(t, err)
@ -1385,6 +1397,9 @@ func TestShardCluster_Query(t *testing.T) {
require.EqualValues(t, unavailable, sc.state.Load())
_, err := sc.Query(ctx, &querypb.QueryRequest{
Req: &internalpb.RetrieveRequest{
Base: &commonpb.MsgBase{},
},
DmlChannels: []string{vchannelName},
}, streamingDoNothing)
assert.Error(t, err)
@ -1398,6 +1413,9 @@ func TestShardCluster_Query(t *testing.T) {
sc.SetupFirstVersion()
_, err := sc.Query(ctx, &querypb.QueryRequest{
Req: &internalpb.RetrieveRequest{
Base: &commonpb.MsgBase{},
},
DmlChannels: []string{vchannelName + "_suffix"},
}, streamingDoNothing)
assert.Error(t, err)
@ -1447,6 +1465,9 @@ func TestShardCluster_Query(t *testing.T) {
require.EqualValues(t, available, sc.state.Load())
result, err := sc.Query(ctx, &querypb.QueryRequest{
Req: &internalpb.RetrieveRequest{
Base: &commonpb.MsgBase{},
},
DmlChannels: []string{vchannelName},
}, streamingDoNothing)
assert.NoError(t, err)
@ -1497,6 +1518,9 @@ func TestShardCluster_Query(t *testing.T) {
require.EqualValues(t, available, sc.state.Load())
_, err := sc.Query(ctx, &querypb.QueryRequest{
Req: &internalpb.RetrieveRequest{
Base: &commonpb.MsgBase{},
},
DmlChannels: []string{vchannelName},
}, func(ctx context.Context) error { return errors.New("mocked") })
assert.Error(t, err)
@ -1555,6 +1579,9 @@ func TestShardCluster_Query(t *testing.T) {
require.EqualValues(t, available, sc.state.Load())
_, err := sc.Query(ctx, &querypb.QueryRequest{
Req: &internalpb.RetrieveRequest{
Base: &commonpb.MsgBase{},
},
DmlChannels: []string{vchannelName},
}, streamingDoNothing)
assert.Error(t, err)
@ -1608,6 +1635,9 @@ func TestShardCluster_Query(t *testing.T) {
require.EqualValues(t, available, sc.state.Load())
_, err := sc.Query(ctx, &querypb.QueryRequest{
Req: &internalpb.RetrieveRequest{
Base: &commonpb.MsgBase{},
},
DmlChannels: []string{vchannelName},
}, streamingDoNothing)
assert.Error(t, err)