test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-11-30 10:59:32 +08:00
Code Issues Actions 11 Packages Projects Releases Wiki Activity

Add check for stats log and bin log number (#21264)

Browse Source 
Signed-off-by: xiaofan-luan <xiaofan.luan@zilliz.com>
This commit is contained in:
Xiaofan 2022-12-26 16:35:30 +08:00 committed by GitHub
parent 1e5cd9dc6e
commit 5d044c4f29
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 19 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
internal/datacoord/meta.go
View File

@ -1017,6 +1017,10 @@ func (m *meta) alterMetaStoreAfterCompaction(modSegments []*SegmentInfo, newSegm
}
log.Info("meta update: alter meta store for compaction updates",
zap.Int64s("compact from segments (segments to be updated as dropped)", modSegIDs),
zap.Int64("new segmentId", newSegment.GetID()),
zap.Int("binlog", len(newSegment.GetBinlogs())),
zap.Int("stats log", len(newSegment.GetStatslogs())),
zap.Int("delta logs", len(newSegment.GetDeltalogs())),
zap.Int64("compact to segment", newSegment.GetID()))
m.Lock()

15
internal/metastore/kv/datacoord/kv_catalog.go
View File

@ -519,6 +519,21 @@ func buildBinlogKvsWithLogID(collectionID, partitionID, segmentID typeutil.Uniqu
checkBinlogs(storage.InsertBinlog, segmentID, binlogs)
checkBinlogs(storage.DeleteBinlog, segmentID, deltalogs)
checkBinlogs(storage.StatsBinlog, segmentID, statslogs)
// check stats log and bin log size match
if len(binlogs) != 0 && len(statslogs) != 0 {
if len(statslogs[0].GetBinlogs()) != len(binlogs[0].GetBinlogs()) {
log.Warn("find invalid segment while bin log size didn't match stat log size",
zap.Int64("collection", collectionID),
zap.Int64("partition", partitionID),
zap.Int64("segment", segmentID),
zap.Any("binlogs", binlogs),
zap.Any("stats", statslogs),
zap.Any("delta", deltalogs),
)
return nil, fmt.Errorf("Segment can not be saved because of binlog "+
"file not match stat log number: collection %v, segment %v", collectionID, segmentID)
}
}
fillLogIDByLogPath(binlogs, deltalogs, statslogs)
kvs, err := buildBinlogKvs(collectionID, partitionID, segmentID, binlogs, deltalogs, statslogs)