furyhawk/milvus
1
0
Fork 0
mirror of https://github.com/milvus-io/milvus.git synced 2026-07-21 10:15:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: [StorageV2] sort field binlogs paths for packed reader and writer (#43585)

Browse Source 
key changes:
* fix unstable storage v2 compaction unit test by guaranteeing the order
of paths during sync.
* bump milvus-storage version, include
https://github.com/milvus-io/milvus-storage/pull/222
https://github.com/milvus-io/milvus-storage/pull/223
https://github.com/milvus-io/milvus-storage/pull/224
https://github.com/milvus-io/milvus-storage/pull/225
https://github.com/milvus-io/milvus-storage/pull/226
* Also fix the below related oom issue.
related: https://github.com/milvus-io/milvus/issues/43310

Signed-off-by: shaoting-huang <shaoting.huang@zilliz.com>
This commit is contained in:
sthuang
2025-07-30 08:09:36 +08:00
committed by GitHub
parent 4fe55e3008
commit a2c7ed2780
15 changed files with 37 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/core/thirdparty/milvus-storage/CMakeLists.txt
Vendored
+1 -1
View File
@@ -14,7 +14,7 @@
# Update milvus-storage_VERSION for the first occurrence
milvus_add_pkg_config("milvus-storage")
set_property(DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR} PROPERTY INCLUDE_DIRECTORIES "")
set( milvus-storage_VERSION d757696 )
set( milvus-storage_VERSION aca750a )
set( GIT_REPOSITORY "https://github.com/milvus-io/milvus-storage.git")
message(STATUS "milvus-storage repo: ${GIT_REPOSITORY}")
message(STATUS "milvus-storage version: ${milvus-storage_VERSION}")
internal/datanode/compactor/clustering_compactor_storage_v2_test.go
+2 -2
View File
@@ -99,7 +99,7 @@ func (s *ClusteringCompactionTaskStorageV2Suite) TestScalarCompactionNormal_V2To
s.task.plan.SegmentBinlogs = []*datapb.CompactionSegmentBinlogs{
{
SegmentID: segmentID,
FieldBinlogs: lo.Values(fBinlogs),
FieldBinlogs: storage.SortFieldBinlogs(fBinlogs),
Deltalogs: []*datapb.FieldBinlog{deltalogs},
StorageVersion: storage.StorageV2,
},
@@ -161,7 +161,7 @@ func (s *ClusteringCompactionTaskStorageV2Suite) TestScalarCompactionNormal_V2To
s.task.plan.SegmentBinlogs = []*datapb.CompactionSegmentBinlogs{
{
SegmentID: segmentID,
FieldBinlogs: lo.Values(fBinlogs),
FieldBinlogs: storage.SortFieldBinlogs(fBinlogs),
Deltalogs: []*datapb.FieldBinlog{deltalogs},
StorageVersion: storage.StorageV2,
},
internal/datanode/compactor/mix_compactor_storage_v2_test.go
+3 -3
View File
@@ -123,7 +123,7 @@ func (s *MixCompactionTaskStorageV2Suite) TestCompactDupPK_MixToV2Format() {
s.NoError(err)
s.task.plan.SegmentBinlogs = append(s.task.plan.SegmentBinlogs, &datapb.CompactionSegmentBinlogs{
SegmentID: segID,
FieldBinlogs: lo.Values(binlogs),
FieldBinlogs: storage.SortFieldBinlogs(binlogs),
Deltalogs: []*datapb.FieldBinlog{},
StorageVersion: storage.StorageV2,
})
@@ -157,7 +157,7 @@ func (s *MixCompactionTaskStorageV2Suite) TestCompactDupPK_V2ToV2Format() {
s.NoError(err)
s.task.plan.SegmentBinlogs = append(s.task.plan.SegmentBinlogs, &datapb.CompactionSegmentBinlogs{
SegmentID: segID,
FieldBinlogs: lo.Values(binlogs),
FieldBinlogs: storage.SortFieldBinlogs(binlogs),
Deltalogs: []*datapb.FieldBinlog{},
StorageVersion: storage.StorageV2,
})
@@ -192,7 +192,7 @@ func (s *MixCompactionTaskStorageV2Suite) TestCompactDupPK_V2ToV1Format() {
s.NoError(err)
s.task.plan.SegmentBinlogs = append(s.task.plan.SegmentBinlogs, &datapb.CompactionSegmentBinlogs{
SegmentID: segID,
FieldBinlogs: lo.Values(binlogs),
FieldBinlogs: storage.SortFieldBinlogs(binlogs),
Deltalogs: []*datapb.FieldBinlog{},
StorageVersion: storage.StorageV2,
})
internal/datanode/compactor/segment_writer.go
+1 -1
View File
@@ -124,7 +124,7 @@ func (w *MultiSegmentWriter) closeWriter() error {
result := &datapb.CompactionSegment{
SegmentID: w.currentSegmentID,
InsertLogs: lo.Values(fieldBinlogs),
InsertLogs: storage.SortFieldBinlogs(fieldBinlogs),
Field2StatslogPaths: []*datapb.FieldBinlog{statsLog},
NumOfRows: w.writer.GetRowNum(),
Channel: w.channel,
internal/datanode/compactor/sort_compaction.go
+1 -1
View File
@@ -225,7 +225,7 @@ func (t *sortCompactionTask) sortSegment(ctx context.Context) (*datapb.Compactio
}
binlogs, stats, bm25stats := srw.GetLogs()
insertLogs := lo.Values(binlogs)
insertLogs := storage.SortFieldBinlogs(binlogs)
if err := binlog.CompressFieldBinlogs(insertLogs); err != nil {
return nil, err
}
internal/datanode/index/task_stats.go
+1 -1
View File
@@ -254,7 +254,7 @@ func (st *statsTask) sort(ctx context.Context) ([]*datapb.FieldBinlog, error) {
}
binlogs, stats, bm25stats := srw.GetLogs()
insertLogs := lo.Values(binlogs)
insertLogs := storage.SortFieldBinlogs(binlogs)
if err := binlog.CompressFieldBinlogs(insertLogs); err != nil {
return nil, err
}
internal/flushcommon/syncmgr/meta_writer.go
+2 -1
View File
@@ -10,6 +10,7 @@ import (
"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
"github.com/milvus-io/milvus/internal/flushcommon/broker"
"github.com/milvus-io/milvus/internal/flushcommon/metacache"
storage "github.com/milvus-io/milvus/internal/storage"
"github.com/milvus-io/milvus/pkg/v2/log"
"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
"github.com/milvus-io/milvus/pkg/v2/util/commonpbutil"
@@ -45,7 +46,7 @@ func (b *brokerMetaWriter) UpdateSync(ctx context.Context, pack *SyncTask) error
return merr.WrapErrSegmentNotFound(pack.segmentID)
}
insertFieldBinlogs := append(segment.Binlogs(), lo.MapToSlice(pack.insertBinlogs, func(_ int64, fieldBinlog *datapb.FieldBinlog) *datapb.FieldBinlog { return fieldBinlog })...)
insertFieldBinlogs := append(segment.Binlogs(), storage.SortFieldBinlogs(pack.insertBinlogs)...)
statsFieldBinlogs := append(segment.Statslogs(), lo.MapToSlice(pack.statsBinlogs, func(_ int64, fieldBinlog *datapb.FieldBinlog) *datapb.FieldBinlog { return fieldBinlog })...)
deltaFieldBinlogs := segment.Deltalogs()
internal/flushcommon/syncmgr/options.go
-10
View File
@@ -78,13 +78,3 @@ func (t *SyncTask) WithFailureCallback(callback func(error)) *SyncTask {
t.failureCallback = callback
return t
}
func (t *SyncTask) WithSyncBufferSize(size int64) *SyncTask {
t.syncBufferSize = size
return t
}
func (t *SyncTask) WithMultiPartUploadSize(size int64) *SyncTask {
t.multiPartUploadSize = size
return t
}
internal/flushcommon/syncmgr/task.go
+6 -6
View File
@@ -30,6 +30,7 @@ import (
"github.com/milvus-io/milvus/internal/flushcommon/metacache"
"github.com/milvus-io/milvus/internal/json"
"github.com/milvus-io/milvus/internal/storage"
"github.com/milvus-io/milvus/internal/storagev2/packed"
"github.com/milvus-io/milvus/pkg/v2/log"
"github.com/milvus-io/milvus/pkg/v2/metrics"
"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
@@ -78,10 +79,8 @@ type SyncTask struct {
tr *timerecord.TimeRecorder
flushedSize int64
execTime time.Duration
multiPartUploadSize int64
syncBufferSize int64
flushedSize int64
execTime time.Duration
// storage config used in pooled tasks, optional
// use singleton config for non-pooled tasks
@@ -131,8 +130,9 @@ func (t *SyncTask) Run(ctx context.Context) (err error) {
switch segmentInfo.GetStorageVersion() {
case storage.StorageV2:
writer := NewBulkPackWriterV2(t.metacache, t.schema, t.chunkManager, t.allocator, t.syncBufferSize,
t.multiPartUploadSize, t.storageConfig, t.writeRetryOpts...)
// New sync task means needs to flush data immediately, so do not need to buffer data in writer again.
writer := NewBulkPackWriterV2(t.metacache, t.schema, t.chunkManager, t.allocator, 0,
packed.DefaultMultiPartUploadSize, t.storageConfig, t.writeRetryOpts...)
t.insertBinlogs, t.deltaBinlog, t.statsBinlogs, t.bm25Binlogs, t.flushedSize, err = writer.Write(ctx, t.pack)
if err != nil {
log.Warn("failed to write sync data with storage v2 format", zap.Error(err))
internal/flushcommon/syncmgr/task_test.go
-18
View File
@@ -38,7 +38,6 @@ import (
"github.com/milvus-io/milvus/internal/json"
"github.com/milvus-io/milvus/internal/mocks"
"github.com/milvus-io/milvus/internal/storage"
"github.com/milvus-io/milvus/internal/storagev2/packed"
"github.com/milvus-io/milvus/internal/util/initcore"
"github.com/milvus-io/milvus/pkg/v2/common"
"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
@@ -224,11 +223,6 @@ func (s *SyncTaskSuite) runTestRunNormal(storageVersion int64) {
Timestamp: 100,
}))
task.WithMetaWriter(BrokerMetaWriter(s.broker, 1))
if storageVersion == storage.StorageV2 {
task.WithMultiPartUploadSize(0)
task.WithSyncBufferSize(packed.DefaultWriteBufferSize)
}
err := task.Run(ctx)
s.NoError(err)
s.True(isDataReleased(task)) // data should be released after task finished
@@ -244,10 +238,6 @@ func (s *SyncTaskSuite) runTestRunNormal(storageVersion int64) {
Timestamp: 100,
}))
task.WithMetaWriter(BrokerMetaWriter(s.broker, 1)).WithSchema(s.schema)
if storageVersion == storage.StorageV2 {
task.WithMultiPartUploadSize(0)
task.WithSyncBufferSize(packed.DefaultWriteBufferSize)
}
err := task.Run(ctx)
s.NoError(err)
@@ -265,10 +255,6 @@ func (s *SyncTaskSuite) runTestRunNormal(storageVersion int64) {
Timestamp: 100,
}))
task.WithMetaWriter(BrokerMetaWriter(s.broker, 1)).WithSchema(s.schema)
if storageVersion == storage.StorageV2 {
task.WithMultiPartUploadSize(0)
task.WithSyncBufferSize(packed.DefaultWriteBufferSize)
}
err := task.Run(ctx)
s.NoError(err)
s.True(isDataReleased(task)) // data should be released after task finished
@@ -285,10 +271,6 @@ func (s *SyncTaskSuite) runTestRunNormal(storageVersion int64) {
Timestamp: 100,
}))
task.WithMetaWriter(BrokerMetaWriter(s.broker, 1)).WithSchema(s.schema)
if storageVersion == storage.StorageV2 {
task.WithMultiPartUploadSize(0)
task.WithSyncBufferSize(packed.DefaultWriteBufferSize)
}
err := task.Run(ctx)
s.NoError(err)
s.True(isDataReleased(task)) // data should be released after task finished
internal/flushcommon/writebuffer/write_buffer.go
+1 -3
View File
@@ -18,7 +18,6 @@ import (
"github.com/milvus-io/milvus/internal/flushcommon/metacache/pkoracle"
"github.com/milvus-io/milvus/internal/flushcommon/syncmgr"
"github.com/milvus-io/milvus/internal/storage"
"github.com/milvus-io/milvus/internal/storagev2/packed"
"github.com/milvus-io/milvus/internal/util/streamingutil"
"github.com/milvus-io/milvus/pkg/v2/log"
"github.com/milvus-io/milvus/pkg/v2/metrics"
@@ -601,8 +600,7 @@ func (wb *writeBufferBase) getSyncTask(ctx context.Context, segmentID int64) (sy
WithMetaWriter(wb.metaWriter).
WithMetaCache(wb.metaCache).
WithSchema(schema).
WithSyncPack(pack).
WithMultiPartUploadSize(packed.DefaultMultiPartUploadSize)
WithSyncPack(pack)
return task, nil
}
internal/storage/rw.go
+3
View File
@@ -241,6 +241,9 @@ func NewBinlogRecordReader(ctx context.Context, binlogs []*datapb.FieldBinlog, s
if len(binlogs) <= 0 {
return nil, sio.EOF
}
sort.Slice(binlogs, func(i, j int) bool {
return binlogs[i].GetFieldID() < binlogs[j].GetFieldID()
})
binlogLists := lo.Map(binlogs, func(fieldBinlog *datapb.FieldBinlog, _ int) []*datapb.Binlog {
return fieldBinlog.GetBinlogs()
})
internal/storage/rw_test.go
+1 -1
View File
@@ -168,7 +168,7 @@ func (s *PackedBinlogRecordSuite) TestPackedBinlogRecordIntegration() {
s.Equal(len(bm25StatsLog), 0)
binlogs := lo.Values(fieldBinlogs)
binlogs := SortFieldBinlogs(fieldBinlogs)
rOption := []RwOption{
WithVersion(StorageV2),
WithStorageConfig(s.storageConfig),
internal/storage/utils.go
+14
View File
@@ -39,6 +39,7 @@ import (
"github.com/milvus-io/milvus/pkg/v2/log"
"github.com/milvus-io/milvus/pkg/v2/mq/msgstream"
"github.com/milvus-io/milvus/pkg/v2/objectstorage"
"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
"github.com/milvus-io/milvus/pkg/v2/proto/segcorepb"
"github.com/milvus-io/milvus/pkg/v2/util/merr"
"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
@@ -1568,3 +1569,16 @@ func fillMissingFields(schema *schemapb.CollectionSchema, insertData *InsertData
}
return nil
}
// sort by field binlogs key
func SortFieldBinlogs(fieldBinlogs map[int64]*datapb.FieldBinlog) []*datapb.FieldBinlog {
fieldIDs := lo.Keys(fieldBinlogs)
sort.Slice(fieldIDs, func(i, j int) bool {
return fieldIDs[i] < fieldIDs[j]
})
binlogs := make([]*datapb.FieldBinlog, 0, len(fieldIDs))
for _, fieldID := range fieldIDs {
binlogs = append(binlogs, fieldBinlogs[fieldID])
}
return binlogs
}
internal/storagev2/packed/packed_test.go
+1 -1
View File
@@ -132,7 +132,7 @@ func (suite *PackedTestSuite) TestPackedMultiFiles() {
defer rec.Release()
paths := []string{"/tmp/100", "/tmp/101"}
columnGroups := []storagecommon.ColumnGroup{{Columns: []int{2}, GroupID: 2}, {Columns: []int{0, 1}, GroupID: storagecommon.DefaultShortColumnGroupID}}
bufferSize := int64(-1) // unlimited
bufferSize := int64(10 * 1024 * 1024) // 10MB
multiPartUploadSize := int64(0)
pw, err := NewPackedWriter(paths, suite.schema, bufferSize, multiPartUploadSize, columnGroups, nil)
suite.NoError(err)