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

enhance: mmap load raw data if scalar index does not have raw data (#33175) 

Signed-off-by: sunby <sunbingyi1992@gmail.com>
pull/32810/head
Bingyi Sun 2024-05-21 11:53:39 +08:00 committed by GitHub
parent b3bcc107bb
commit 0f8c6f49ff
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 9 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/querynodev2/segments/reduce_test.go
View File

@ -101,7 +101,7 @@ func (suite *ReduceSuite) SetupTest() {
) )
suite.Require().NoError(err) suite.Require().NoError(err)
for _, binlog := range binlogs { for _, binlog := range binlogs {
err = suite.segment.(*LocalSegment).LoadFieldData(ctx, binlog.FieldID, int64(msgLength), binlog) err = suite.segment.(*LocalSegment).LoadFieldData(ctx, binlog.FieldID, int64(msgLength), binlog, false)
suite.Require().NoError(err) suite.Require().NoError(err)
} }
} }

2
internal/querynodev2/segments/retrieve_test.go
View File

@ -109,7 +109,7 @@ func (suite *RetrieveSuite) SetupTest() {
) )
suite.Require().NoError(err) suite.Require().NoError(err)
for _, binlog := range binlogs { for _, binlog := range binlogs {
err = suite.sealed.(*LocalSegment).LoadFieldData(ctx, binlog.FieldID, int64(msgLength), binlog) err = suite.sealed.(*LocalSegment).LoadFieldData(ctx, binlog.FieldID, int64(msgLength), binlog, false)
suite.Require().NoError(err) suite.Require().NoError(err)
} }

2
internal/querynodev2/segments/search_test.go
View File

@ -100,7 +100,7 @@ func (suite *SearchSuite) SetupTest() {
) )
suite.Require().NoError(err) suite.Require().NoError(err)
for _, binlog := range binlogs { for _, binlog := range binlogs {
err = suite.sealed.(*LocalSegment).LoadFieldData(ctx, binlog.FieldID, int64(msgLength), binlog) err = suite.sealed.(*LocalSegment).LoadFieldData(ctx, binlog.FieldID, int64(msgLength), binlog, false)
suite.Require().NoError(err) suite.Require().NoError(err)
} }

4
internal/querynodev2/segments/segment.go
View File

@ -967,7 +967,7 @@ func (s *LocalSegment) LoadMultiFieldData(ctx context.Context) error {
return nil return nil
} }
func (s *LocalSegment) LoadFieldData(ctx context.Context, fieldID int64, rowCount int64, field *datapb.FieldBinlog) error { func (s *LocalSegment) LoadFieldData(ctx context.Context, fieldID int64, rowCount int64, field *datapb.FieldBinlog, useMmap bool) error {
if !s.ptrLock.RLockIf(state.IsNotReleased) { if !s.ptrLock.RLockIf(state.IsNotReleased) {
return merr.WrapErrSegmentNotLoaded(s.ID(), "segment released") return merr.WrapErrSegmentNotLoaded(s.ID(), "segment released")
} }
@ -1006,7 +1006,7 @@ func (s *LocalSegment) LoadFieldData(ctx context.Context, fieldID int64, rowCoun
} }
collection := s.collection collection := s.collection
mmapEnabled := common.IsFieldMmapEnabled(collection.Schema(), fieldID) || mmapEnabled := useMmap || common.IsFieldMmapEnabled(collection.Schema(), fieldID) ||
(!common.FieldHasMmapKey(collection.Schema(), fieldID) && params.Params.QueryNodeCfg.MmapEnabled.GetAsBool()) (!common.FieldHasMmapKey(collection.Schema(), fieldID) && params.Params.QueryNodeCfg.MmapEnabled.GetAsBool())
loadFieldDataInfo.appendMMapDirPath(paramtable.Get().QueryNodeCfg.MmapDirPath.GetValue()) loadFieldDataInfo.appendMMapDirPath(paramtable.Get().QueryNodeCfg.MmapDirPath.GetValue())
loadFieldDataInfo.enableMmap(fieldID, mmapEnabled) loadFieldDataInfo.enableMmap(fieldID, mmapEnabled)

6
internal/querynodev2/segments/segment_loader.go
View File

@ -508,7 +508,7 @@ func (loader *segmentLoaderV2) loadSealedSegmentFields(ctx context.Context, segm
runningGroup, _ := errgroup.WithContext(ctx) runningGroup, _ := errgroup.WithContext(ctx)
fields.Range(func(fieldID int64, field *schemapb.FieldSchema) bool { fields.Range(func(fieldID int64, field *schemapb.FieldSchema) bool {
runningGroup.Go(func() error { runningGroup.Go(func() error {
return segment.LoadFieldData(ctx, fieldID, rowCount, nil) return segment.LoadFieldData(ctx, fieldID, rowCount, nil, false)
}) })
return true return true
}) })
@ -1058,7 +1058,7 @@ func (loader *segmentLoader) loadSealedSegment(ctx context.Context, loadInfo *qu
zap.String("index", info.IndexInfo.GetIndexName()), zap.String("index", info.IndexInfo.GetIndexName()),
) )
// for scalar index's raw data, only load to mmap not memory // for scalar index's raw data, only load to mmap not memory
if err = segment.LoadFieldData(ctx, fieldID, loadInfo.GetNumOfRows(), info.FieldBinlog); err != nil { if err = segment.LoadFieldData(ctx, fieldID, loadInfo.GetNumOfRows(), info.FieldBinlog, true); err != nil {
log.Warn("load raw data failed", zap.Int64("fieldID", fieldID), zap.Error(err)) log.Warn("load raw data failed", zap.Int64("fieldID", fieldID), zap.Error(err))
return err return err
} }
@ -1212,7 +1212,7 @@ func loadSealedSegmentFields(ctx context.Context, collection *Collection, segmen
fieldID, fieldID,
rowCount, rowCount,
fieldBinLog, fieldBinLog,
) false)
}) })
} }
err := runningGroup.Wait() err := runningGroup.Wait()

2
internal/querynodev2/segments/segment_test.go
View File

@ -100,7 +100,7 @@ func (suite *SegmentSuite) SetupTest() {
g, err := suite.sealed.(*LocalSegment).StartLoadData() g, err := suite.sealed.(*LocalSegment).StartLoadData()
suite.Require().NoError(err) suite.Require().NoError(err)
for _, binlog := range binlogs { for _, binlog := range binlogs {
err = suite.sealed.(*LocalSegment).LoadFieldData(ctx, binlog.FieldID, int64(msgLength), binlog) err = suite.sealed.(*LocalSegment).LoadFieldData(ctx, binlog.FieldID, int64(msgLength), binlog, false)
suite.Require().NoError(err) suite.Require().NoError(err)
} }
g.Done(nil) g.Done(nil)