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: Fast finish compaction when L0Comp hit zero L1/L2 (#47154) (#49379)

Browse Source 
See also: #46435, #48907, #48909

pr: #47154

Signed-off-by: yangxuan <xuan.yang@zilliz.com>
This commit is contained in:
XuanYang-cn
2026-04-28 14:53:53 +08:00
committed by GitHub
parent 2b5074b98e
commit 9a5729097b
2 changed files with 78 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/datacoord/compaction_task_l0.go
+27 -3
View File
@@ -110,6 +110,30 @@ func (t *l0CompactionTask) processPipelining() bool {
return t.processFailed()
}
if len(t.GetPlan().GetSegmentBinlogs()) == len(t.GetTaskProto().GetInputSegments()) {
log.Info("l0CompactionTask fast finish: no target segments, directly marking L0 segments as dropped",
zap.Int64("planID", t.GetTaskProto().GetPlanID()))
t.result = &datapb.CompactionPlanResult{
PlanID: t.GetTaskProto().GetPlanID(),
State: datapb.CompactionTaskState_completed,
}
if err := t.saveSegmentMeta(); err != nil {
log.Warn("l0CompactionTask fast finish failed to save segment meta", zap.Error(err))
err = t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_failed), setFailReason(err.Error()))
if err != nil {
log.Warn("l0CompactionTask failed to updateAndSaveTaskMeta", zap.Error(err))
return false
}
return t.processFailed()
}
if err := t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_meta_saved)); err != nil {
log.Warn("l0CompactionTask fast finish failed to save task meta_saved state", zap.Error(err))
return false
}
log.Info("l0CompactionTask fast finish completed", zap.Int64("planID", t.GetTaskProto().GetPlanID()))
return t.processMetaSaved()
}
err = t.sessions.Compaction(context.TODO(), t.GetTaskProto().GetNodeID(), t.GetPlan())
if err != nil {
originNodeID := t.GetTaskProto().GetNodeID()
@@ -354,9 +378,9 @@ func (t *l0CompactionTask) BuildCompactionRequest() (*datapb.CompactionPlan, err
sealedSegments, sealedSegBinlogs := t.selectSealedSegment()
if len(sealedSegments) == 0 {
// TODO fast finish l0 segment, just drop l0 segment
log.Info("l0Compaction available non-L0 Segments is empty ")
return nil, errors.Errorf("Selected zero L1/L2 segments for the position=%v", taskProto.GetPos())
log.Info("l0Compaction available non-L0 Segments is empty, will fast finish",
zap.Any("target position", taskProto.GetPos()))
return plan, nil
}
for _, seg := range sealedSegments {
internal/datacoord/compaction_task_l0_test.go
+51 -28
View File
@@ -161,41 +161,64 @@ func (s *L0CompactionTaskSuite) TestProcessRefreshPlan_SelectZeroSegmentsL0() {
State: datapb.CompactionTaskState_executing,
InputSegments: []int64{100, 101},
}, nil, s.mockMeta, nil)
_, err := task.BuildCompactionRequest()
s.Error(err)
plan, err := task.BuildCompactionRequest()
s.NoError(err)
s.Require().NotNil(plan)
s.Equal(2, len(plan.GetSegmentBinlogs()))
segIDs := lo.Map(plan.GetSegmentBinlogs(), func(b *datapb.CompactionSegmentBinlogs, _ int) int64 {
return b.GetSegmentID()
})
s.ElementsMatch([]int64{100, 101}, segIDs)
s.Nil(plan.GetPreAllocatedLogIDs())
}
func (s *L0CompactionTaskSuite) TestBuildCompactionRequestFailed_AllocFailed() {
var task CompactionTask
channel := "Ch-1"
deltaLogs := []*datapb.FieldBinlog{getFieldBinlogIDs(101, 3)}
s.mockAlloc.EXPECT().AllocN(mock.Anything).Return(100, 200, errors.New("mock alloc err"))
s.mockMeta.EXPECT().SelectSegments(mock.Anything, mock.Anything, mock.Anything).Return(
[]*SegmentInfo{
{SegmentInfo: &datapb.SegmentInfo{
ID: 200,
Level: datapb.SegmentLevel_L1,
InsertChannel: channel,
}},
{SegmentInfo: &datapb.SegmentInfo{
ID: 201,
Level: datapb.SegmentLevel_L1,
InsertChannel: channel,
}},
{SegmentInfo: &datapb.SegmentInfo{
ID: 202,
Level: datapb.SegmentLevel_L1,
InsertChannel: channel,
}},
},
)
meta, err := newMemoryMeta(s.T())
s.NoError(err)
task = &l0CompactionTask{
allocator: s.mockAlloc,
meta: meta,
}
task.SetTask(&datapb.CompactionTask{})
_, err = task.BuildCompactionRequest()
s.T().Logf("err=%v", err)
s.Error(err)
s.mockMeta.EXPECT().GetHealthySegment(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, segID int64) *SegmentInfo {
return &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
ID: segID,
Level: datapb.SegmentLevel_L0,
InsertChannel: channel,
State: commonpb.SegmentState_Flushed,
Deltalogs: deltaLogs,
}}
}).Times(2)
task := newL0CompactionTask(&datapb.CompactionTask{
PlanID: 1,
TriggerID: 19530,
CollectionID: 1,
PartitionID: 10,
Type: datapb.CompactionType_Level0DeleteCompaction,
NodeID: 1,
State: datapb.CompactionTaskState_executing,
InputSegments: []int64{100, 101},
}, s.mockAlloc, s.mockMeta, nil)
task = &mixCompactionTask{
allocator: s.mockAlloc,
meta: meta,
}
task.SetTask(&datapb.CompactionTask{})
_, err = task.BuildCompactionRequest()
s.T().Logf("err=%v", err)
s.Error(err)
s.mockAlloc.EXPECT().AllocN(mock.Anything).Return(0, 0, errors.New("mock alloc err"))
task = &clusteringCompactionTask{
allocator: s.mockAlloc,
meta: meta,
}
task.SetTask(&datapb.CompactionTask{})
_, err = task.BuildCompactionRequest()
_, err := task.BuildCompactionRequest()
s.T().Logf("err=%v", err)
s.Error(err)
}