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

enhance: Enable to upload by batch (#32788)

Browse Source 
Signed-off-by: yangxuan <xuan.yang@zilliz.com>
This commit is contained in:
XuanYang-cn 2024-05-13 19:07:32 +08:00 committed by GitHub
parent 29b621f759
commit efdbd8e7c1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 47 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
internal/datanode/io/binlog_io.go
View File

@ -85,17 +85,28 @@ func (b *BinlogIoImpl) Download(ctx context.Context, paths []string) ([][]byte,
func (b *BinlogIoImpl) Upload(ctx context.Context, kvs map[string][]byte) error {
ctx, span := otel.Tracer(typeutil.DataNodeRole).Start(ctx, "Upload")
defer span.End()
future := b.pool.Submit(func() (any, error) {
log.Debug("BinlogIO uplaod", zap.Strings("paths", lo.Keys(kvs)))
err := retry.Do(ctx, func() error {
return b.MultiWrite(ctx, kvs)
futures := make([]*conc.Future[any], 0, len(kvs))
for k, v := range kvs {
innerK, innerV := k, v
future := b.pool.Submit(func() (any, error) {
var err error
log.Debug("BinlogIO upload", zap.String("paths", innerK))
err = retry.Do(ctx, func() error {
err = b.Write(ctx, innerK, innerV)
if err != nil {
log.Warn("BinlogIO fail to upload", zap.String("paths", innerK), zap.Error(err))
}
return err
})
return struct{}{}, err
})
return nil, err
})
futures = append(futures, future)
}
_, err := future.Await()
return err
return conc.AwaitAll(futures...)
}
func (b *BinlogIoImpl) JoinFullPath(paths ...string) string {

41
internal/datanode/l0_compactor.go
View File

@ -355,6 +355,8 @@ func (t *levelZeroCompactionTask) composeDeltalog(segmentID int64, dData *storag
}
func (t *levelZeroCompactionTask) uploadByCheck(ctx context.Context, requireCheck bool, alteredSegments map[int64]*storage.DeleteData, resultSegments map[int64]*datapb.CompactionSegment) error {
allBlobs := make(map[string][]byte)
tmpResults := make(map[int64]*datapb.CompactionSegment)
for segID, dData := range alteredSegments {
if !requireCheck || (dData.Size() >= paramtable.Get().DataNodeCfg.FlushDeleteBufferBytes.GetAsInt64()) {
blobs, binlog, err := t.composeDeltalog(segID, dData)
@ -362,24 +364,33 @@ func (t *levelZeroCompactionTask) uploadByCheck(ctx context.Context, requireChec
log.Warn("L0 compaction composeDelta fail", zap.Int64("segmentID", segID), zap.Error(err))
return err
}
err = t.Upload(ctx, blobs)
if err != nil {
log.Warn("L0 compaction upload blobs fail", zap.Int64("segmentID", segID), zap.Any("binlog", binlog), zap.Error(err))
return err
allBlobs = lo.Assign(blobs, allBlobs)
tmpResults[segID] = &datapb.CompactionSegment{
SegmentID: segID,
Deltalogs: []*datapb.FieldBinlog{{Binlogs: []*datapb.Binlog{binlog}}},
Channel: t.plan.GetChannel(),
}
if _, ok := resultSegments[segID]; !ok {
resultSegments[segID] = &datapb.CompactionSegment{
SegmentID: segID,
Deltalogs: []*datapb.FieldBinlog{{Binlogs: []*datapb.Binlog{binlog}}},
Channel: t.plan.GetChannel(),
}
} else {
resultSegments[segID].Deltalogs[0].Binlogs = append(resultSegments[segID].Deltalogs[0].Binlogs, binlog)
}
delete(alteredSegments, segID)
}
}
if len(allBlobs) == 0 {
return nil
}
if err := t.Upload(ctx, allBlobs); err != nil {
log.Warn("L0 compaction upload blobs fail", zap.Error(err))
return err
}
for segID, compSeg := range tmpResults {
if _, ok := resultSegments[segID]; !ok {
resultSegments[segID] = compSeg
} else {
binlog := compSeg.Deltalogs[0].Binlogs[0]
resultSegments[segID].Deltalogs[0].Binlogs = append(resultSegments[segID].Deltalogs[0].Binlogs, binlog)
}
}
return nil
}

4
internal/datanode/l0_compactor_test.go
View File

@ -220,7 +220,7 @@ func (s *LevelZeroCompactionTaskSuite) TestCompactLinear() {
RunAndReturn(func(paths ...string) string {
return path.Join(paths...)
}).Times(2)
s.mockBinlogIO.EXPECT().Upload(mock.Anything, mock.Anything).Return(nil).Times(2)
s.mockBinlogIO.EXPECT().Upload(mock.Anything, mock.Anything).Return(nil).Once()
s.Require().Equal(plan.GetPlanID(), s.task.getPlanID())
s.Require().Equal(plan.GetChannel(), s.task.getChannelName())
@ -322,7 +322,7 @@ func (s *LevelZeroCompactionTaskSuite) TestCompactBatch() {
RunAndReturn(func(paths ...string) string {
return path.Join(paths...)
}).Times(2)
s.mockBinlogIO.EXPECT().Upload(mock.Anything, mock.Anything).Return(nil).Times(2)
s.mockBinlogIO.EXPECT().Upload(mock.Anything, mock.Anything).Return(nil).Once()
l0Segments := lo.Filter(s.task.plan.GetSegmentBinlogs(), func(s *datapb.CompactionSegmentBinlogs, _ int) bool {
return s.Level == datapb.SegmentLevel_L0