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

Remove partition target check in querynodev2 (#23188) 

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
pull/23188/merge
congqixia 2023-04-04 13:36:32 +08:00 committed by GitHub
parent 9ad6d9f1a0
commit 3af3fe52d8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 5 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
internal/querynodev2/pipeline/filter_node_test.go
View File

@ -111,7 +111,7 @@ func (suite *FilterNodeSuite) TestWithLoadCollection() {
//test filter node with collection load partition
func (suite *FilterNodeSuite) TestWithLoadPartation() {
//data
suite.validSegmentIDs = []int64{2, 4, 6}
suite.validSegmentIDs = []int64{2, 3, 4, 5, 6}
//mock
collection := segments.NewCollectionWithoutSchema(suite.collectionID, querypb.LoadType_LoadPartition)
@ -138,7 +138,7 @@ func (suite *FilterNodeSuite) TestWithLoadPartation() {
for _, msg := range nodeMsg.insertMsgs {
suite.True(lo.Contains(suite.validSegmentIDs, msg.SegmentID))
}
suite.Equal(suite.deleteSegmentSum/2, len(nodeMsg.deleteMsgs))
suite.Equal(suite.deleteSegmentSum, len(nodeMsg.deleteMsgs))
}
func (suite *FilterNodeSuite) buildMsgPack() *msgstream.MsgPack {

15
internal/querynodev2/pipeline/filter_policy.go
View File

@ -16,8 +16,6 @@
package pipeline
import "github.com/milvus-io/milvus/internal/common"
//MsgFilter will return error if Msg was invalid
type InsertMsgFilter = func(n *filterNode, c *Collection, msg *InsertMsg) error
type DeleteMsgFilter = func(n *filterNode, c *Collection, msg *DeleteMsg) error
@ -44,11 +42,7 @@ func InsertOutOfTarget(n *filterNode, c *Collection, msg *InsertMsg) error {
return WrapErrMsgNotTarget("Collection")
}
if c.GetLoadType() == loadTypePartition {
if msg.PartitionID != common.InvalidPartitionID && !c.ExistPartition(msg.PartitionID) {
return WrapErrMsgNotTarget("Partition")
}
}
// all growing will be be in-memory to support dynamic partition load/release
return nil
}
@ -82,10 +76,7 @@ func DeleteOutOfTarget(n *filterNode, c *Collection, msg *DeleteMsg) error {
if msg.GetCollectionID() != c.ID() {
return WrapErrMsgNotTarget("Collection")
}
if c.GetLoadType() == loadTypePartition {
if msg.PartitionID != common.InvalidPartitionID && !c.ExistPartition(msg.PartitionID) {
return WrapErrMsgNotTarget("Partition")
}
}
// all growing will be be in-memory to support dynamic partition load/release
return nil
}