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

Fix panic while applying deletions (#19956) 

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

Signed-off-by: yah01 <yang.cen@zilliz.com>
pull/19964/head
yah01 2022-10-21 14:25:29 +08:00 committed by GitHub
parent 3d3289ce3e
commit a4fb3cc3e2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 28 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
internal/querynode/flow_graph_delete_node.go
View File

@ -17,6 +17,7 @@
package querynode
import (
"errors"
"fmt"
"reflect"
"sync"
@ -143,9 +144,17 @@ func (dNode *deleteNode) Operate(in []flowgraph.Msg) []flowgraph.Msg {
err := dNode.delete(delData, segmentID, &wg)
if err != nil {
// error occurs when segment cannot be found, calling cgo function delete failed and etc...
err = fmt.Errorf("segment delete failed, segmentID = %d, err = %s", segmentID, err)
log.Error(err.Error())
panic(err)
log.Error("failed to apply deletions to segment",
zap.Int64("segmentID", segmentID),
zap.Error(err),
)
// For cases: segment compacted, not loaded yet, or just released,
// to ignore the error,
// panic otherwise.
if !errors.Is(err, ErrSegmentNotFound) && !errors.Is(err, ErrSegmentUnhealthy) {
panic(err)
}
}
}()
}
@ -166,7 +175,7 @@ func (dNode *deleteNode) delete(deleteData *deleteData, segmentID UniqueID, wg *
defer wg.Done()
targetSegment, err := dNode.metaReplica.getSegmentByID(segmentID, segmentTypeSealed)
if err != nil {
return fmt.Errorf("getSegmentByID failed, err = %s", err)
return WrapSegmentNotFound(segmentID)
}
if targetSegment.getType() != segmentTypeSealed {
@ -179,7 +188,7 @@ func (dNode *deleteNode) delete(deleteData *deleteData, segmentID UniqueID, wg *
err = targetSegment.segmentDelete(offset, ids, timestamps)
if err != nil {
return fmt.Errorf("segmentDelete failed, segmentID = %d", segmentID)
return fmt.Errorf("segmentDelete failed, segmentID = %d, err=%w", segmentID, err)
}
log.Debug("Do delete done", zap.Int("len", len(deleteData.deleteIDs[segmentID])), zap.Int64("segmentID", segmentID), zap.Any("SegmentType", targetSegment.segmentType), zap.String("channel", dNode.channel))

6
internal/querynode/flow_graph_insert_node.go
View File

@ -186,7 +186,7 @@ func (iNode *insertNode) Operate(in []flowgraph.Msg) []flowgraph.Msg {
if targetSegment != nil {
offset, err := targetSegment.segmentPreInsert(numOfRecords)
if err != nil {
if errors.Is(err, errSegmentUnhealthy) {
if errors.Is(err, ErrSegmentUnhealthy) {
log.Debug("segment removed before preInsert")
continue
}
@ -387,7 +387,7 @@ func (iNode *insertNode) insert(iData *insertData, segmentID UniqueID) error {
err = targetSegment.segmentInsert(offsets, ids, timestamps, insertRecord)
if err != nil {
if errors.Is(err, errSegmentUnhealthy) {
if errors.Is(err, ErrSegmentUnhealthy) {
log.Debug("segment removed before insert")
return nil
}
@ -425,7 +425,7 @@ func (iNode *insertNode) delete(deleteData *deleteData, segmentID UniqueID) erro
err = targetSegment.segmentDelete(offset, ids, timestamps)
if err != nil {
if errors.Is(err, errSegmentUnhealthy) {
if errors.Is(err, ErrSegmentUnhealthy) {
log.Debug("segment removed before delete")
return nil
}

20
internal/querynode/segment.go
View File

@ -68,7 +68,9 @@ const (
maxBloomFalsePositive float64 = 0.005
)
var errSegmentUnhealthy = errors.New("segment unhealthy")
var (
ErrSegmentUnhealthy = errors.New("segment unhealthy")
)
// IndexedFieldInfo contains binlog info of vector field
type IndexedFieldInfo struct {
@ -344,7 +346,7 @@ func (s *Segment) search(searchReq *searchRequest) (*SearchResult, error) {
s.mut.RLock()
defer s.mut.RUnlock()
if !s.healthy() {
return nil, fmt.Errorf("%w(segmentID=%d)", errSegmentUnhealthy, s.segmentID)
return nil, fmt.Errorf("%w(segmentID=%d)", ErrSegmentUnhealthy, s.segmentID)
}
if searchReq.plan == nil {
@ -382,7 +384,7 @@ func (s *Segment) retrieve(plan *RetrievePlan) (*segcorepb.RetrieveResults, erro
s.mut.RLock()
defer s.mut.RUnlock()
if !s.healthy() {
return nil, fmt.Errorf("%w(segmentID=%d)", errSegmentUnhealthy, s.segmentID)
return nil, fmt.Errorf("%w(segmentID=%d)", ErrSegmentUnhealthy, s.segmentID)
}
var retrieveResult RetrieveResult
@ -646,7 +648,7 @@ func (s *Segment) segmentPreInsert(numOfRecords int) (int64, error) {
s.mut.RLock()
defer s.mut.RUnlock()
if !s.healthy() {
return -1, fmt.Errorf("%w(segmentID=%d)", errSegmentUnhealthy, s.segmentID)
return -1, fmt.Errorf("%w(segmentID=%d)", ErrSegmentUnhealthy, s.segmentID)
}
var offset int64
@ -691,7 +693,7 @@ func (s *Segment) segmentInsert(offset int64, entityIDs []UniqueID, timestamps [
s.mut.RLock()
defer s.mut.RUnlock()
if !s.healthy() {
return fmt.Errorf("%w(segmentID=%d)", errSegmentUnhealthy, s.segmentID)
return fmt.Errorf("%w(segmentID=%d)", ErrSegmentUnhealthy, s.segmentID)
}
insertRecordBlob, err := proto.Marshal(record)
@ -743,7 +745,7 @@ func (s *Segment) segmentDelete(offset int64, entityIDs []primaryKey, timestamps
s.mut.RLock()
defer s.mut.RUnlock()
if !s.healthy() {
return fmt.Errorf("%w(segmentID=%d)", errSegmentUnhealthy, s.segmentID)
return fmt.Errorf("%w(segmentID=%d)", ErrSegmentUnhealthy, s.segmentID)
}
if len(entityIDs) != len(timestamps) {
@ -813,7 +815,7 @@ func (s *Segment) segmentLoadFieldData(fieldID int64, rowCount int64, data *sche
s.mut.RLock()
defer s.mut.RUnlock()
if !s.healthy() {
return fmt.Errorf("%w(segmentID=%d)", errSegmentUnhealthy, s.segmentID)
return fmt.Errorf("%w(segmentID=%d)", ErrSegmentUnhealthy, s.segmentID)
}
dataBlob, err := proto.Marshal(data)
@ -850,7 +852,7 @@ func (s *Segment) segmentLoadDeletedRecord(primaryKeys []primaryKey, timestamps
s.mut.RLock()
defer s.mut.RUnlock()
if !s.healthy() {
return fmt.Errorf("%w(segmentID=%d)", errSegmentUnhealthy, s.segmentID)
return fmt.Errorf("%w(segmentID=%d)", ErrSegmentUnhealthy, s.segmentID)
}
if len(primaryKeys) <= 0 {
@ -938,7 +940,7 @@ func (s *Segment) segmentLoadIndexData(bytesIndex [][]byte, indexInfo *querypb.F
s.mut.RLock()
defer s.mut.RUnlock()
if !s.healthy() {
return fmt.Errorf("%w(segmentID=%d)", errSegmentUnhealthy, s.segmentID)
return fmt.Errorf("%w(segmentID=%d)", ErrSegmentUnhealthy, s.segmentID)
}
var status C.CStatus