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

Fix bug of repack function of InsertTask in Proxy 

Signed-off-by: dragondriver <jiquan.long@zilliz.com>
pull/4973/head^2
dragondriver 2020-11-26 15:17:20 +08:00 committed by yefu.chen
parent d9e5a8774d
commit ab75438abf
1 changed files with 19 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
internal/proxy/repack_func.go
View File

@ -58,6 +58,9 @@ func insertRepackFunc(tsMsgs []msgstream.TsMsg,
reqSegCountMap := make(map[UniqueID]map[int32]map[UniqueID]uint32) reqSegCountMap := make(map[UniqueID]map[int32]map[UniqueID]uint32)
for reqID, countInfo := range channelCountMap { for reqID, countInfo := range channelCountMap {
if _, ok := reqSegCountMap[reqID]; !ok {
reqSegCountMap[reqID] = make(map[int32]map[UniqueID]uint32)
}
schema := reqSchemaMap[reqID] schema := reqSchemaMap[reqID]
collName, partitionTag := schema[0], schema[1] collName, partitionTag := schema[0], schema[1]
for channelID, count := range countInfo { for channelID, count := range countInfo {
@ -65,6 +68,7 @@ func insertRepackFunc(tsMsgs []msgstream.TsMsg,
if err != nil { if err != nil {
return nil, err return nil, err
} }
reqSegCountMap[reqID][channelID] = make(map[UniqueID]uint32)
reqSegCountMap[reqID][channelID] = mapInfo reqSegCountMap[reqID][channelID] = mapInfo
} }
} }
@ -74,6 +78,15 @@ func insertRepackFunc(tsMsgs []msgstream.TsMsg,
reqSegAllocateCounter := make(map[UniqueID]map[int32]uint32) reqSegAllocateCounter := make(map[UniqueID]map[int32]uint32)
for reqID, channelInfo := range reqSegCountMap { for reqID, channelInfo := range reqSegCountMap {
if _, ok := reqSegAccumulateCountMap[reqID]; !ok {
reqSegAccumulateCountMap[reqID] = make(map[int32][]uint32)
}
if _, ok := reqSegIDMap[reqID]; !ok {
reqSegIDMap[reqID] = make(map[int32][]UniqueID)
}
if _, ok := reqSegAllocateCounter[reqID]; !ok {
reqSegAllocateCounter[reqID] = make(map[int32]uint32)
}
for channelID, segInfo := range channelInfo { for channelID, segInfo := range channelInfo {
reqSegAllocateCounter[reqID][channelID] = 0 reqSegAllocateCounter[reqID][channelID] = 0
keys := make([]UniqueID, len(segInfo)) keys := make([]UniqueID, len(segInfo))
@ -86,10 +99,16 @@ func insertRepackFunc(tsMsgs []msgstream.TsMsg,
accumulate := uint32(0) accumulate := uint32(0)
for _, key := range keys { for _, key := range keys {
accumulate += segInfo[key] accumulate += segInfo[key]
if _, ok := reqSegAccumulateCountMap[reqID][channelID]; !ok {
reqSegAccumulateCountMap[reqID][channelID] = make([]uint32, 0)
}
reqSegAccumulateCountMap[reqID][channelID] = append( reqSegAccumulateCountMap[reqID][channelID] = append(
reqSegAccumulateCountMap[reqID][channelID], reqSegAccumulateCountMap[reqID][channelID],
accumulate, accumulate,
) )
if _, ok := reqSegIDMap[reqID][channelID]; !ok {
reqSegIDMap[reqID][channelID] = make([]UniqueID, 0)
}
reqSegIDMap[reqID][channelID] = append( reqSegIDMap[reqID][channelID] = append(
reqSegIDMap[reqID][channelID], reqSegIDMap[reqID][channelID],
key, key,