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

fix: [2.4] Get all children deltalogs for segment to load (#40964) 

issue: https://github.com/milvus-io/milvus/issues/40207
master pr: https://github.com/milvus-io/milvus/pull/40956
2.5 pr: #40957

Signed-off-by: Cai Zhang <cai.zhang@zilliz.com>
pull/41047/head
cai.zhang 2025-03-31 14:36:22 +08:00 committed by GitHub
parent 4898b9d3c8
commit a84ba1967d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 34 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
internal/datacoord/services.go
View File

@ -395,16 +395,38 @@ func (s *Server) GetSegmentInfo(ctx context.Context, req *datapb.GetSegmentInfoR
}
infos := make([]*datapb.SegmentInfo, 0, len(req.GetSegmentIDs()))
channelCPs := make(map[string]*msgpb.MsgPosition)
var getChildrenDelta func(id UniqueID) ([]*datapb.FieldBinlog, error)
getChildrenDelta = func(id UniqueID) ([]*datapb.FieldBinlog, error) {
children, ok := s.meta.GetCompactionTo(id)
// double-check the segment, maybe the segment is being dropped concurrently.
if !ok {
log.Warn("failed to get segment, this may have been cleaned", zap.Int64("segmentID", id))
err := merr.WrapErrSegmentNotFound(id)
return nil, err
}
allDeltaLogs := make([]*datapb.FieldBinlog, 0)
for _, child := range children {
clonedChild := child.Clone()
// child segment should decompress binlog path
binlog.DecompressBinLog(storage.DeleteBinlog, clonedChild.GetCollectionID(), clonedChild.GetPartitionID(), clonedChild.GetID(), clonedChild.GetDeltalogs())
allDeltaLogs = append(allDeltaLogs, clonedChild.GetDeltalogs()...)
allChildrenDeltas, err := getChildrenDelta(child.GetID())
if err != nil {
return nil, err
}
allDeltaLogs = append(allDeltaLogs, allChildrenDeltas...)
}
return allDeltaLogs, nil
}
for _, id := range req.SegmentIDs {
var info *SegmentInfo
if req.IncludeUnHealthy {
info = s.meta.GetSegment(id)
// TODO: GetCompactionTo should be removed and add into GetSegment method and protected by lock.
// Too much modification need to be applied to SegmentInfo, a refactor is needed.
children, ok := s.meta.GetCompactionTo(id)
// info may be not-nil, but ok is false when the segment is being dropped concurrently.
if info == nil || !ok {
if info == nil {
log.Warn("failed to get segment, this may have been cleaned", zap.Int64("segmentID", id))
err := merr.WrapErrSegmentNotFound(id)
resp.Status = merr.Status(err)
@ -412,13 +434,14 @@ func (s *Server) GetSegmentInfo(ctx context.Context, req *datapb.GetSegmentInfoR
}
clonedInfo := info.Clone()
for _, child := range children {
clonedChild := child.Clone()
// child segment should decompress binlog path
binlog.DecompressBinLog(storage.DeleteBinlog, clonedChild.GetCollectionID(), clonedChild.GetPartitionID(), clonedChild.GetID(), clonedChild.GetDeltalogs())
clonedInfo.Deltalogs = append(clonedInfo.Deltalogs, clonedChild.GetDeltalogs()...)
clonedInfo.DmlPosition = clonedChild.GetDmlPosition()
// We should retrieve the deltalog of all child segments,
// but due to the compaction constraint based on indexed segment, there will be at most two generations.
allChildrenDeltalogs, err := getChildrenDelta(id)
if err != nil {
resp.Status = merr.Status(err)
return resp, nil
}
clonedInfo.Deltalogs = append(clonedInfo.Deltalogs, allChildrenDeltalogs...)
segmentutil.ReCalcRowCount(info.SegmentInfo, clonedInfo.SegmentInfo)
infos = append(infos, clonedInfo.SegmentInfo)
} else {