admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2026-02-02 01:06:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

Use serialized size before compression in log meta (#18337)

Browse Source 
Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
This commit is contained in:
congqixia 2022-07-19 21:38:30 +08:00 committed by GitHub
parent 8fcd6aa56b
commit 69aea00c09
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 12 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
internal/datanode/flush_manager.go
View File

@ -349,6 +349,11 @@ func (m *rendezvousFlushManager) flushBufferData(data *BufferData, segmentID Uni
if err != nil {
return err
}
// get memory size of buffer data
fieldMemorySize := make(map[int64]int)
for fieldID, fieldData := range data.buffer.Data {
fieldMemorySize[fieldID] = fieldData.GetMemorySize()
}
// encode data and convert output data
inCodec := storage.NewInsertCodec(meta)
@ -385,7 +390,7 @@ func (m *rendezvousFlushManager) flushBufferData(data *BufferData, segmentID Uni
TimestampFrom: 0, //TODO
TimestampTo: 0, //TODO,
LogPath: key,
LogSize: int64(len(blob.Value)),
LogSize: int64(fieldMemorySize[fieldID]),
}
field2Logidx[fieldID] = logidx
}

10
internal/indexnode/task.go
View File

@ -468,6 +468,12 @@ func (it *IndexBuildTask) buildIndex(ctx context.Context) ([]*storage.Blob, erro
}
it.tr.Record("index serialize done")
// use serialized size before encoding
it.serializedSize = 0
for _, blob := range indexBlobs {
it.serializedSize += uint64(len(blob.Value))
}
// early release index for gc, and we can ensure that Delete is idempotent.
if err := it.index.Delete(); err != nil {
log.Error("IndexNode IndexBuildTask Execute CIndexDelete failed",
@ -499,10 +505,6 @@ func (it *IndexBuildTask) buildIndex(ctx context.Context) ([]*storage.Blob, erro
func (it *IndexBuildTask) saveIndex(ctx context.Context, blobs []*storage.Blob) error {
blobCnt := len(blobs)
it.serializedSize = 0
for i := range blobs {
it.serializedSize += uint64(len(blobs[i].Value))
}
getSavePathByKey := func(key string) string {
return path.Join(Params.IndexNodeCfg.IndexStorageRootPath, strconv.Itoa(int(it.req.IndexBuildID)), strconv.Itoa(int(it.req.Version)),