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

Fix duplicate delete bugs (#10715) 

Signed-off-by: fishpenguin <kun.yu@zilliz.com>
pull/10738/head
yukun 2021-10-27 14:36:23 +08:00 committed by GitHub
parent 5cd935729c
commit 902e7172cb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 9 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
internal/querynode/flow_graph_insert_node.go
View File

@ -147,14 +147,21 @@ func (iNode *insertNode) Operate(in []flowgraph.Msg) []flowgraph.Msg {
}
// 1. filter segment by bloom filter
for _, delMsg := range iMsg.deleteMessages {
if iNode.streamingReplica != nil {
if iNode.streamingReplica.getSegmentNum() != 0 {
processDeleteMessages(iNode.streamingReplica, delMsg, delData)
}
if iNode.historicalReplica != nil {
if iNode.historicalReplica.getSegmentNum() != 0 {
processDeleteMessages(iNode.historicalReplica, delMsg, delData)
}
}
// 2. do predelete
for segmentID, pks := range delData.deleteIDs {
segment := iNode.getSegmentInReplica(segmentID)
offset := segment.segmentPreDelete(len(pks))
delData.deleteOffset[segmentID] = offset
}
// 2. do delete
for segmentID := range delData.deleteIDs {
wg.Add(1)
@ -205,15 +212,9 @@ func processDeleteMessages(replica ReplicaInterface, msg *msgstream.DeleteMsg, d
continue
}
if len(pks) > 0 {
offset := segment.segmentPreDelete(len(pks))
if err != nil {
log.Warn(err.Error())
continue
}
delData.deleteIDs[segmentID] = append(delData.deleteIDs[segmentID], pks...)
// TODO(yukun) get offset of pks
delData.deleteTimestamps[segmentID] = append(delData.deleteTimestamps[segmentID], msg.Timestamps[:len(pks)]...)
delData.deleteOffset[segmentID] = offset
}
}
}