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

fix: use LoadDeltaData instead of Delete for L0 growing forward (#46657)

Browse Source 
Related to #46660
Replace segment.Delete() with segment.LoadDeltaData() when forwarding L0
deletions to growing segments. LoadDeltaData is the more appropriate API
for bulk loading delta data compared to individual Delete calls.

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
• Core invariant: forwarding L0 deletions to growing segments must use
the bulk-delta API (storage.DeltaData + segment.LoadDeltaData) because
LoadDeltaData preserves paired primary keys and timestamps as a single
atomic delta payload; segment.Delete was intended for per-delete RPCs
and not for loading L0 delta payloads.

• Logic removed/simplified: addL0GrowingBF() no longer calls
segment.Delete for buffered L0 keys. Instead the buffered callback
builds a storage.DeltaData via storage.NewDeltaDataWithData(pks, tss)
and calls segment.LoadDeltaData(ctx, dd). This eliminates the previous
per-batch Delete call path and centralizes forwarding as a single
delta-load operation.

• Why this does not cause data loss or regression: the new path supplies
identical PK+timestamp pairs to the segment via DeltaData; LoadDeltaData
applies the same delete semantics but accepts batched delta payloads.
The change is limited to the L0→growing Bloom-Filter forward path
(addL0GrowingBF/addL0ForGrowingLoad), leaving sealed-segment deletes,
streaming direct forwarding, and remote-load policies unchanged. Also,
the prior code would fail on L0Segment.Delete (L0 segments prohibit
Delete), so switching to LoadDeltaData prevents lost-forwarding caused
by unsupported Delete calls.

• Category: Enhancement / Refactor — replaces inappropriate per-delete
calls with the correct bulk delta-load API, simplifying error handling
around NewDeltaDataWithData and ensuring API contract correctness for
L0→growing forwarding.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
This commit is contained in:
congqixia 2025-12-30 14:05:21 +08:00 committed by GitHub
parent 898e6d6e94
commit b4682b7352
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 9 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
internal/querynodev2/delegator/delta_forward.go
View File

@ -101,7 +101,11 @@ func (sd *shardDelegator) addL0ForGrowing(ctx context.Context, segment segments.
func (sd *shardDelegator) addL0GrowingBF(ctx context.Context, segment segments.Segment) error {
bufferedForwarder := NewBufferedForwarder(paramtable.Get().QueryNodeCfg.ForwardBatchSize.GetAsInt64(),
func(pks storage.PrimaryKeys, tss []uint64) error {
return segment.Delete(ctx, pks, tss)
dd, err := storage.NewDeltaDataWithData(pks, tss)
if err != nil {
return err
}
return segment.LoadDeltaData(ctx, dd)
})
if err := sd.rangeHitL0Deletions(segment.Partition(), segment, func(pk storage.PrimaryKey, ts uint64) error {

8
internal/querynodev2/delegator/delta_forward_test.go
View File

@ -425,16 +425,16 @@ func (s *GrowingMergeL0Suite) TestAddL0ForGrowingBF() {
seg.EXPECT().ID().Return(1)
seg.EXPECT().Partition().Return(100)
seg.EXPECT().BatchPkExist(mock.Anything).Return(lo.RepeatBy(n, func(i int) bool { return true }))
seg.EXPECT().Delete(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, pk storage.PrimaryKeys, u []uint64) error {
s.Equal(deltaData.DeletePks(), pk)
s.Equal(deltaData.DeleteTimestamps(), u)
seg.EXPECT().LoadDeltaData(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, dd *storage.DeltaData) error {
s.Equal(deltaData.DeletePks(), dd.DeletePks())
s.Equal(deltaData.DeleteTimestamps(), dd.DeleteTimestamps())
return nil
}).Once()
err = sd.addL0ForGrowing(context.Background(), seg)
s.NoError(err)
seg.EXPECT().Delete(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, pk storage.PrimaryKeys, u []uint64) error {
seg.EXPECT().LoadDeltaData(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, dd *storage.DeltaData) error {
return errors.New("mocked")
}).Once()
err = sd.addL0ForGrowing(context.Background(), seg)