mirror of
https://github.com/milvus-io/milvus.git
synced 2026-07-21 10:15:43 +00:00
congqixiaandGitHub
89659fd182
Related to #49706 Move L0 V3 manifest delta commits into the segment meta update path so manifest pointer updates and deltalog metadata appends are protected by the same DataCoord lock under concurrent compaction tasks. --------- Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
804 lines
30 KiB
Go
804 lines
30 KiB
Go
// Licensed to the LF AI & Data foundation under one
|
|
// or more contributor license agreements. See the NOTICE file
|
|
// distributed with this work for additional information
|
|
// regarding copyright ownership. The ASF licenses this file
|
|
// to you under the Apache License, Version 2.0 (the
|
|
// "License"); you may not use this file except in compliance
|
|
// with the License. You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package datacoord
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
|
|
"github.com/cockroachdb/errors"
|
|
"github.com/samber/lo"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/mock"
|
|
"github.com/stretchr/testify/suite"
|
|
|
|
"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
|
|
"github.com/milvus-io/milvus-proto/go-api/v3/msgpb"
|
|
"github.com/milvus-io/milvus/internal/datacoord/allocator"
|
|
"github.com/milvus-io/milvus/internal/datacoord/session"
|
|
"github.com/milvus-io/milvus/pkg/v3/proto/datapb"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/merr"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
|
|
)
|
|
|
|
func TestL0CompactionTaskSuite(t *testing.T) {
|
|
suite.Run(t, new(L0CompactionTaskSuite))
|
|
}
|
|
|
|
type L0CompactionTaskSuite struct {
|
|
suite.Suite
|
|
|
|
mockAlloc *allocator.MockAllocator
|
|
mockMeta *MockCompactionMeta
|
|
}
|
|
|
|
func (s *L0CompactionTaskSuite) SetupTest() {
|
|
s.mockMeta = NewMockCompactionMeta(s.T())
|
|
s.mockAlloc = allocator.NewMockAllocator(s.T())
|
|
// s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything).Return(nil)
|
|
}
|
|
|
|
func (s *L0CompactionTaskSuite) SetupSubTest() {
|
|
s.SetupTest()
|
|
}
|
|
|
|
func (s *L0CompactionTaskSuite) TestSaveSegmentMetaUsesAtomicDeltalogOperator() {
|
|
actualDeltaPath := "/tmp/milvus/insert_log/1/10/200/_delta/not-log-id-suffix"
|
|
|
|
task := s.generateTestL0Task(datapb.CompactionTaskState_executing)
|
|
output := []*datapb.CompactionSegment{{
|
|
SegmentID: 200,
|
|
Deltalogs: []*datapb.FieldBinlog{{
|
|
Binlogs: []*datapb.Binlog{{LogID: 9001, LogPath: actualDeltaPath, EntriesNum: 3}},
|
|
}},
|
|
}}
|
|
|
|
s.mockMeta.EXPECT().UpdateSegmentsInfo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
|
|
func(ctx context.Context, operators ...UpdateOperator) error {
|
|
s.Len(operators, 5)
|
|
s.Equal(actualDeltaPath, output[0].GetDeltalogs()[0].GetBinlogs()[0].GetLogPath())
|
|
s.EqualValues(9001, output[0].GetDeltalogs()[0].GetBinlogs()[0].GetLogID())
|
|
return nil
|
|
},
|
|
).Once()
|
|
|
|
s.NoError(task.saveSegmentMeta(output))
|
|
}
|
|
|
|
func (s *L0CompactionTaskSuite) TestProcessRefreshPlan_NormalL0() {
|
|
channel := "Ch-1"
|
|
deltaLogs := []*datapb.FieldBinlog{getFieldBinlogIDs(101, 3)}
|
|
|
|
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,
|
|
}},
|
|
},
|
|
)
|
|
|
|
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},
|
|
}, nil, s.mockMeta)
|
|
alloc := allocator.NewMockAllocator(s.T())
|
|
alloc.EXPECT().AllocN(mock.Anything).Return(100, 200, nil)
|
|
task.allocator = alloc
|
|
plan, err := task.BuildCompactionRequest()
|
|
s.Require().NoError(err)
|
|
|
|
s.Equal(5, len(plan.GetSegmentBinlogs()))
|
|
segIDs := lo.Map(plan.GetSegmentBinlogs(), func(b *datapb.CompactionSegmentBinlogs, _ int) int64 {
|
|
return b.GetSegmentID()
|
|
})
|
|
|
|
s.ElementsMatch([]int64{200, 201, 202, 100, 101}, segIDs)
|
|
}
|
|
|
|
func (s *L0CompactionTaskSuite) TestProcessRefreshPlan_SegmentNotFoundL0() {
|
|
channel := "Ch-1"
|
|
s.mockMeta.EXPECT().GetHealthySegment(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, segID int64) *SegmentInfo {
|
|
return nil
|
|
}).Once()
|
|
task := newL0CompactionTask(&datapb.CompactionTask{
|
|
InputSegments: []int64{102},
|
|
PlanID: 1,
|
|
TriggerID: 19530,
|
|
CollectionID: 1,
|
|
PartitionID: 10,
|
|
Channel: channel,
|
|
Type: datapb.CompactionType_Level0DeleteCompaction,
|
|
NodeID: 1,
|
|
State: datapb.CompactionTaskState_executing,
|
|
}, nil, s.mockMeta)
|
|
|
|
_, err := task.BuildCompactionRequest()
|
|
s.Error(err)
|
|
s.ErrorIs(err, merr.ErrSegmentNotFound)
|
|
}
|
|
|
|
func (s *L0CompactionTaskSuite) TestProcessRefreshPlan_SelectZeroSegmentsL0() {
|
|
channel := "Ch-1"
|
|
deltaLogs := []*datapb.FieldBinlog{getFieldBinlogIDs(101, 3)}
|
|
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)
|
|
s.mockMeta.EXPECT().SelectSegments(mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()
|
|
|
|
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},
|
|
}, nil, s.mockMeta)
|
|
plan, err := task.BuildCompactionRequest()
|
|
// Fast finish: should return a plan with only L0 segments (no error)
|
|
s.NoError(err)
|
|
s.Require().NotNil(plan)
|
|
// Verify plan only contains L0 input segments (2 segments)
|
|
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)
|
|
// Verify no binlog IDs were allocated for fast finish
|
|
s.Nil(plan.GetPreAllocatedLogIDs())
|
|
}
|
|
|
|
func (s *L0CompactionTaskSuite) TestBuildCompactionRequestFailed_AllocFailed() {
|
|
channel := "Ch-1"
|
|
deltaLogs := []*datapb.FieldBinlog{getFieldBinlogIDs(101, 3)}
|
|
|
|
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,
|
|
}},
|
|
},
|
|
)
|
|
|
|
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)
|
|
|
|
s.mockAlloc.EXPECT().AllocN(mock.Anything).Return(0, 0, errors.New("mock alloc err"))
|
|
|
|
_, err := task.BuildCompactionRequest()
|
|
s.T().Logf("err=%v", err)
|
|
s.Error(err)
|
|
}
|
|
|
|
func (s *L0CompactionTaskSuite) generateTestL0Task(state datapb.CompactionTaskState) *l0CompactionTask {
|
|
return newL0CompactionTask(&datapb.CompactionTask{
|
|
PlanID: 1,
|
|
TriggerID: 19530,
|
|
CollectionID: 1,
|
|
PartitionID: 10,
|
|
Type: datapb.CompactionType_Level0DeleteCompaction,
|
|
NodeID: NullNodeID,
|
|
State: state,
|
|
Channel: "ch-1",
|
|
InputSegments: []int64{100, 101},
|
|
}, s.mockAlloc, s.mockMeta)
|
|
}
|
|
|
|
func (s *L0CompactionTaskSuite) TestPorcessStateTrans() {
|
|
s.Run("test pipelining Compaction failed", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
|
|
s.mockAlloc.EXPECT().AllocN(mock.Anything).Return(100, 200, nil)
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_pipelining)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
channel := "ch-1"
|
|
deltaLogs := []*datapb.FieldBinlog{getFieldBinlogIDs(101, 3)}
|
|
|
|
s.mockMeta.EXPECT().SelectSegments(mock.Anything, mock.Anything, mock.Anything).Return(
|
|
[]*SegmentInfo{
|
|
{SegmentInfo: &datapb.SegmentInfo{
|
|
ID: 200,
|
|
Level: datapb.SegmentLevel_L1,
|
|
InsertChannel: channel,
|
|
}},
|
|
},
|
|
)
|
|
|
|
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,
|
|
}}
|
|
}).Twice()
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
|
|
|
|
cluster := session.NewMockCluster(s.T())
|
|
cluster.EXPECT().CreateCompaction(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(nodeID int64, plan *datapb.CompactionPlan, collectionID int64) error {
|
|
s.Require().EqualValues(t.GetTaskProto().NodeID, nodeID)
|
|
s.Require().EqualValues(t.GetTaskProto().GetCollectionID(), collectionID)
|
|
return errors.New("mock error")
|
|
})
|
|
|
|
t.CreateTaskOnWorker(100, cluster)
|
|
s.Equal(datapb.CompactionTaskState_pipelining, t.GetTaskProto().State)
|
|
s.EqualValues(NullNodeID, t.GetTaskProto().NodeID)
|
|
})
|
|
|
|
s.Run("test pipelining success", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
|
|
s.mockAlloc.EXPECT().AllocN(mock.Anything).Return(100, 200, nil)
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_pipelining)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
channel := "ch-1"
|
|
deltaLogs := []*datapb.FieldBinlog{getFieldBinlogIDs(101, 3)}
|
|
|
|
s.mockMeta.EXPECT().SelectSegments(mock.Anything, mock.Anything, mock.Anything).Return(
|
|
[]*SegmentInfo{
|
|
{SegmentInfo: &datapb.SegmentInfo{
|
|
ID: 200,
|
|
Level: datapb.SegmentLevel_L1,
|
|
InsertChannel: channel,
|
|
}},
|
|
},
|
|
)
|
|
|
|
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,
|
|
}}
|
|
}).Twice()
|
|
|
|
cluster := session.NewMockCluster(s.T())
|
|
cluster.EXPECT().CreateCompaction(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(nodeID int64, plan *datapb.CompactionPlan, collectionID int64) error {
|
|
s.Require().EqualValues(t.GetTaskProto().NodeID, nodeID)
|
|
s.Require().EqualValues(t.GetTaskProto().GetCollectionID(), collectionID)
|
|
return nil
|
|
})
|
|
|
|
t.CreateTaskOnWorker(100, cluster)
|
|
s.Equal(datapb.CompactionTaskState_executing, t.GetTaskProto().GetState())
|
|
})
|
|
|
|
// stay in executing state when GetCompactionPlanResults error except ErrNodeNotFound
|
|
s.Run("test executing GetCompactionPlanResult fail NodeNotFound", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
|
|
cluster := session.NewMockCluster(s.T())
|
|
cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).Return(nil, merr.WrapErrNodeNotFound(t.GetTaskProto().NodeID)).Once()
|
|
|
|
t.QueryTaskOnWorker(cluster)
|
|
s.Equal(datapb.CompactionTaskState_pipelining, t.GetTaskProto().GetState())
|
|
s.EqualValues(NullNodeID, t.GetTaskProto().GetNodeID())
|
|
})
|
|
|
|
// stay in executing state when GetCompactionPlanResults error except ErrNodeNotFound
|
|
s.Run("test executing GetCompactionPlanResult fail mock error", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
|
|
cluster := session.NewMockCluster(s.T())
|
|
cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).Return(nil, errors.New("mock error"))
|
|
t.QueryTaskOnWorker(cluster)
|
|
s.Equal(datapb.CompactionTaskState_pipelining, t.GetTaskProto().GetState())
|
|
s.EqualValues(-1, t.GetTaskProto().GetNodeID())
|
|
})
|
|
|
|
s.Run("test executing with result executing", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
cluster := session.NewMockCluster(s.T())
|
|
cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).
|
|
Return(&datapb.CompactionPlanResult{
|
|
PlanID: t.GetTaskProto().GetPlanID(),
|
|
State: datapb.CompactionTaskState_executing,
|
|
}, nil).Once()
|
|
|
|
t.QueryTaskOnWorker(cluster)
|
|
})
|
|
|
|
s.Run("test executing with result completed", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil).Once()
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
|
|
cluster := session.NewMockCluster(s.T())
|
|
cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).
|
|
Return(&datapb.CompactionPlanResult{
|
|
PlanID: t.GetTaskProto().GetPlanID(),
|
|
State: datapb.CompactionTaskState_completed,
|
|
}, nil).Once()
|
|
|
|
s.mockMeta.EXPECT().ValidateSegmentStateBeforeCompleteCompactionMutation(mock.Anything).Return(nil).Once()
|
|
s.mockMeta.EXPECT().UpdateSegmentsInfo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil).Times(2)
|
|
s.mockMeta.EXPECT().SetSegmentsCompacting(mock.Anything, mock.Anything, false).Return().Once()
|
|
|
|
t.QueryTaskOnWorker(cluster)
|
|
s.Equal(datapb.CompactionTaskState_completed, t.GetTaskProto().GetState())
|
|
})
|
|
s.Run("test executing with result completed save segment meta failed", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
|
|
cluster := session.NewMockCluster(s.T())
|
|
cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).
|
|
Return(&datapb.CompactionPlanResult{
|
|
PlanID: t.GetTaskProto().GetPlanID(),
|
|
State: datapb.CompactionTaskState_completed,
|
|
}, nil).Once()
|
|
|
|
s.mockMeta.EXPECT().ValidateSegmentStateBeforeCompleteCompactionMutation(mock.Anything).Return(nil).Once()
|
|
s.mockMeta.EXPECT().UpdateSegmentsInfo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
|
|
Return(errors.New("mock error")).Once()
|
|
|
|
t.QueryTaskOnWorker(cluster)
|
|
s.Equal(datapb.CompactionTaskState_executing, t.GetTaskProto().GetState())
|
|
})
|
|
s.Run("test executing with result completed save compaction meta failed", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil).Once()
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
|
|
cluster := session.NewMockCluster(s.T())
|
|
cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).
|
|
Return(&datapb.CompactionPlanResult{
|
|
PlanID: t.GetTaskProto().GetPlanID(),
|
|
State: datapb.CompactionTaskState_completed,
|
|
}, nil).Once()
|
|
|
|
s.mockMeta.EXPECT().ValidateSegmentStateBeforeCompleteCompactionMutation(mock.Anything).Return(nil).Once()
|
|
s.mockMeta.EXPECT().UpdateSegmentsInfo(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once()
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(errors.New("mock error")).Once()
|
|
|
|
t.QueryTaskOnWorker(cluster)
|
|
s.Equal(datapb.CompactionTaskState_executing, t.GetTaskProto().GetState())
|
|
})
|
|
|
|
s.Run("test executing with result failed", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
|
|
cluster := session.NewMockCluster(s.T())
|
|
cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).
|
|
Return(&datapb.CompactionPlanResult{
|
|
PlanID: t.GetTaskProto().GetPlanID(),
|
|
State: datapb.CompactionTaskState_failed,
|
|
}, nil).Once()
|
|
|
|
t.QueryTaskOnWorker(cluster)
|
|
s.Equal(datapb.CompactionTaskState_failed, t.GetTaskProto().GetState())
|
|
})
|
|
s.Run("test executing with result failed save compaction meta failed", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil).Once()
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_executing)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
|
|
cluster := session.NewMockCluster(s.T())
|
|
cluster.EXPECT().QueryCompaction(t.GetTaskProto().NodeID, mock.Anything).
|
|
Return(&datapb.CompactionPlanResult{
|
|
PlanID: t.GetTaskProto().GetPlanID(),
|
|
State: datapb.CompactionTaskState_failed,
|
|
}, nil).Once()
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(errors.New("mock error")).Once()
|
|
|
|
t.QueryTaskOnWorker(cluster)
|
|
s.Equal(datapb.CompactionTaskState_executing, t.GetTaskProto().GetState())
|
|
})
|
|
|
|
s.Run("test metaSaved success", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_meta_saved)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
|
|
s.mockMeta.EXPECT().SetSegmentsCompacting(mock.Anything, mock.Anything, false).RunAndReturn(func(ctx context.Context, segIDs []int64, isCompacting bool) {
|
|
s.ElementsMatch(segIDs, t.GetTaskProto().GetInputSegments())
|
|
}).Once()
|
|
|
|
got := t.Process()
|
|
s.True(got)
|
|
s.Equal(datapb.CompactionTaskState_completed, t.GetTaskProto().GetState())
|
|
})
|
|
|
|
s.Run("test metaSaved failed", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil).Once()
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_meta_saved)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(errors.New("mock error")).Once()
|
|
|
|
got := t.Process()
|
|
s.False(got)
|
|
s.Equal(datapb.CompactionTaskState_meta_saved, t.GetTaskProto().GetState())
|
|
})
|
|
|
|
s.Run("test complete drop failed", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_completed)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
s.mockMeta.EXPECT().SetSegmentsCompacting(mock.Anything, mock.Anything, false).RunAndReturn(func(ctx context.Context, segIDs []int64, isCompacting bool) {
|
|
s.ElementsMatch(segIDs, t.GetTaskProto().GetInputSegments())
|
|
}).Once()
|
|
|
|
got := t.Process()
|
|
s.True(got)
|
|
s.Equal(datapb.CompactionTaskState_completed, t.GetTaskProto().GetState())
|
|
})
|
|
|
|
s.Run("test complete success", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_completed)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
s.mockMeta.EXPECT().SetSegmentsCompacting(mock.Anything, mock.Anything, false).RunAndReturn(func(ctx context.Context, segIDs []int64, isCompacting bool) {
|
|
s.ElementsMatch(segIDs, t.GetTaskProto().GetInputSegments())
|
|
}).Once()
|
|
|
|
got := t.Process()
|
|
s.True(got)
|
|
s.Equal(datapb.CompactionTaskState_completed, t.GetTaskProto().GetState())
|
|
})
|
|
|
|
s.Run("test process failed success", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_failed)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
|
|
got := t.Process()
|
|
s.True(got)
|
|
s.Equal(datapb.CompactionTaskState_failed, t.GetTaskProto().GetState())
|
|
})
|
|
|
|
s.Run("test process failed failed", func() {
|
|
s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil)
|
|
t := s.generateTestL0Task(datapb.CompactionTaskState_failed)
|
|
t.updateAndSaveTaskMeta(setNodeID(100))
|
|
s.Require().True(t.GetTaskProto().GetNodeID() > 0)
|
|
|
|
got := t.Process()
|
|
s.True(got)
|
|
s.Equal(datapb.CompactionTaskState_failed, t.GetTaskProto().GetState())
|
|
})
|
|
|
|
s.Run("test process states", func() {
|
|
testCases := []struct {
|
|
state datapb.CompactionTaskState
|
|
processResult bool
|
|
}{
|
|
{state: datapb.CompactionTaskState_unknown, processResult: false},
|
|
{state: datapb.CompactionTaskState_pipelining, processResult: false},
|
|
{state: datapb.CompactionTaskState_executing, processResult: false},
|
|
{state: datapb.CompactionTaskState_failed, processResult: true},
|
|
{state: datapb.CompactionTaskState_timeout, processResult: true},
|
|
}
|
|
|
|
for _, tc := range testCases {
|
|
t := s.generateTestL0Task(tc.state)
|
|
res := t.Process()
|
|
s.Equal(tc.processResult, res)
|
|
}
|
|
})
|
|
}
|
|
|
|
// TestSelectFlushedSegment_ForceSelectAllFlag exercises the flag end-to-end:
|
|
// build an L0 view, call Trigger() with the flag off / on, feed the resulting
|
|
// latestDeletePos into an l0CompactionTask (the same wiring
|
|
// compaction_trigger_v2.go uses), and verify selectFlushedSegment's output
|
|
// actually changes based on the flag. A high-StartPosition segment (the
|
|
// shape silently dropped by the import-position bug) is excluded with the
|
|
// flag off and included with the flag on.
|
|
func (s *L0CompactionTaskSuite) TestSelectFlushedSegment_ForceSelectAllFlag() {
|
|
paramtable.Init()
|
|
const flagKey = "dataCoord.compaction.levelzero.forceSelectAllSegments"
|
|
|
|
channel := "ch-1"
|
|
collectionID := int64(1)
|
|
partitionID := int64(10)
|
|
label := &CompactionGroupLabel{
|
|
CollectionID: collectionID,
|
|
PartitionID: partitionID,
|
|
Channel: channel,
|
|
}
|
|
|
|
// Flushed L1 segments visible to selectFlushedSegment. The one with
|
|
// StartPosition > realL0DmlTs is the case we care about — it would
|
|
// normally be filtered out by `startPos < taskPos`.
|
|
const realL0DmlTs = uint64(5000)
|
|
lowPosSeg := &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
|
|
ID: 200,
|
|
CollectionID: collectionID,
|
|
PartitionID: partitionID,
|
|
InsertChannel: channel,
|
|
Level: datapb.SegmentLevel_L1,
|
|
State: commonpb.SegmentState_Flushed,
|
|
StartPosition: &msgpb.MsgPosition{ChannelName: channel, Timestamp: 3000},
|
|
}}
|
|
highPosSeg := &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
|
|
ID: 201,
|
|
CollectionID: collectionID,
|
|
PartitionID: partitionID,
|
|
InsertChannel: channel,
|
|
Level: datapb.SegmentLevel_L1,
|
|
State: commonpb.SegmentState_Flushed,
|
|
StartPosition: &msgpb.MsgPosition{ChannelName: channel, Timestamp: 20000},
|
|
}}
|
|
|
|
// Mockery's SelectSegments returns whatever we tell it without applying
|
|
// filters. We need the real filter logic to run so the `startPos <
|
|
// taskPos` predicate is actually exercised — install RunAndReturn that
|
|
// evaluates each SegmentFilter.Match against our fixed segment set.
|
|
installFilteringMock := func() {
|
|
s.mockMeta.EXPECT().SelectSegments(mock.Anything, mock.Anything, mock.Anything).
|
|
RunAndReturn(func(ctx context.Context, filters ...SegmentFilter) []*SegmentInfo {
|
|
all := []*SegmentInfo{lowPosSeg, highPosSeg}
|
|
result := make([]*SegmentInfo, 0, len(all))
|
|
for _, seg := range all {
|
|
matched := true
|
|
for _, f := range filters {
|
|
if !f.Match(seg) {
|
|
matched = false
|
|
break
|
|
}
|
|
}
|
|
if matched {
|
|
result = append(result, seg)
|
|
}
|
|
}
|
|
return result
|
|
})
|
|
}
|
|
|
|
// Build an L0 view with a couple of L0 segments whose dmlPos is
|
|
// `realL0DmlTs`. Trigger() runs resolveLatestDeletePos under the current
|
|
// flag value, so the returned view's latestDeletePos reflects the flag.
|
|
buildView := func() *LevelZeroCompactionView {
|
|
l0Segs := []*SegmentView{
|
|
{
|
|
ID: 100,
|
|
label: label,
|
|
dmlPos: &msgpb.MsgPosition{ChannelName: channel, Timestamp: realL0DmlTs},
|
|
Level: datapb.SegmentLevel_L0,
|
|
State: commonpb.SegmentState_Flushed,
|
|
DeltalogCount: 100,
|
|
DeltaSize: 1,
|
|
DeltaRowCount: 1,
|
|
},
|
|
{
|
|
ID: 101,
|
|
label: label,
|
|
dmlPos: &msgpb.MsgPosition{ChannelName: channel, Timestamp: realL0DmlTs},
|
|
Level: datapb.SegmentLevel_L0,
|
|
State: commonpb.SegmentState_Flushed,
|
|
DeltalogCount: 100,
|
|
DeltaSize: 1,
|
|
DeltaRowCount: 1,
|
|
},
|
|
}
|
|
return &LevelZeroCompactionView{
|
|
label: label,
|
|
l0Segments: l0Segs,
|
|
latestDeletePos: &msgpb.MsgPosition{ChannelName: channel, Timestamp: realL0DmlTs},
|
|
triggerID: 19530,
|
|
}
|
|
}
|
|
|
|
// Mirrors compaction_trigger_v2.go:406 — feed the triggered view's
|
|
// latestDeletePos into task.Pos and run selectFlushedSegment.
|
|
runSelectWithTriggeredPos := func() []int64 {
|
|
installFilteringMock()
|
|
srcView := buildView()
|
|
triggered, reason := srcView.Trigger()
|
|
s.Require().NotNil(triggered, "Trigger returned nil: %s", reason)
|
|
triggeredView := triggered.(*LevelZeroCompactionView)
|
|
|
|
task := newL0CompactionTask(&datapb.CompactionTask{
|
|
PlanID: 1,
|
|
TriggerID: 19530,
|
|
CollectionID: collectionID,
|
|
PartitionID: partitionID,
|
|
Channel: channel,
|
|
Type: datapb.CompactionType_Level0DeleteCompaction,
|
|
NodeID: 1,
|
|
State: datapb.CompactionTaskState_executing,
|
|
InputSegments: []int64{100, 101},
|
|
Pos: triggeredView.latestDeletePos,
|
|
}, nil, s.mockMeta)
|
|
|
|
flushed, _, err := task.selectFlushedSegment()
|
|
s.Require().NoError(err)
|
|
return lo.Map(flushed, func(seg *SegmentInfo, _ int) int64 { return seg.GetID() })
|
|
}
|
|
|
|
s.Run("flag_off_excludes_high_start_position_segment", func() {
|
|
paramtable.Get().Save(flagKey, "false")
|
|
defer paramtable.Get().Reset(flagKey)
|
|
|
|
gotIDs := runSelectWithTriggeredPos()
|
|
s.ElementsMatch([]int64{200}, gotIDs,
|
|
"with flag off, segment 201 (StartPosition=20000 > taskPos=%d) must be excluded", realL0DmlTs)
|
|
})
|
|
|
|
s.Run("flag_on_includes_high_start_position_segment", func() {
|
|
paramtable.Get().Save(flagKey, "true")
|
|
defer paramtable.Get().Reset(flagKey)
|
|
|
|
gotIDs := runSelectWithTriggeredPos()
|
|
s.ElementsMatch([]int64{200, 201}, gotIDs,
|
|
"with flag on, resolveLatestDeletePos must lift taskPos so segment 201 is included")
|
|
})
|
|
}
|
|
|
|
// TestSelectFlushedSegment_RespectsCommitTimestamp verifies that import segments
|
|
// with a commit_timestamp are excluded from L0 compaction when the trigger
|
|
// position is before the commit_timestamp.
|
|
func TestSelectFlushedSegment_RespectsCommitTimestamp(t *testing.T) {
|
|
channel := "ch-1"
|
|
|
|
// Import segment: start_position.ts=1000, commit_ts=5000.
|
|
// Its effective timestamp is 5000 (controlled by segmentEffectiveTs).
|
|
importSeg := &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
|
|
ID: 777,
|
|
CollectionID: 1,
|
|
PartitionID: 10,
|
|
InsertChannel: channel,
|
|
State: commonpb.SegmentState_Flushed,
|
|
Level: datapb.SegmentLevel_L1,
|
|
CommitTimestamp: 5000,
|
|
StartPosition: &msgpb.MsgPosition{Timestamp: 1000},
|
|
}}
|
|
|
|
// applyFilters applies a SegmentFilter slice to a candidate list,
|
|
// mirroring what meta.SelectSegments does internally.
|
|
applyFilters := func(candidates []*SegmentInfo, filters ...SegmentFilter) []*SegmentInfo {
|
|
var result []*SegmentInfo
|
|
for _, seg := range candidates {
|
|
pass := true
|
|
for _, f := range filters {
|
|
if !f.Match(seg) {
|
|
pass = false
|
|
break
|
|
}
|
|
}
|
|
if pass {
|
|
result = append(result, seg)
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
makeTask := func(triggerTs uint64) *l0CompactionTask {
|
|
mockAlloc := allocator.NewMockAllocator(t)
|
|
mockMeta := NewMockCompactionMeta(t)
|
|
mockMeta.EXPECT().SelectSegments(mock.Anything, mock.Anything, mock.Anything).
|
|
RunAndReturn(func(ctx context.Context, filters ...SegmentFilter) []*SegmentInfo {
|
|
return applyFilters([]*SegmentInfo{importSeg}, filters...)
|
|
})
|
|
return newL0CompactionTask(&datapb.CompactionTask{
|
|
PlanID: 1,
|
|
TriggerID: 19530,
|
|
CollectionID: 1,
|
|
PartitionID: 10,
|
|
Type: datapb.CompactionType_Level0DeleteCompaction,
|
|
Channel: channel,
|
|
Pos: &msgpb.MsgPosition{Timestamp: triggerTs},
|
|
}, mockAlloc, mockMeta)
|
|
}
|
|
|
|
t.Run("import segment not selected when trigger pos < commit_timestamp", func(t *testing.T) {
|
|
// triggerTs=3000 < commit_ts=5000 → segment must NOT be selected
|
|
task := makeTask(3000)
|
|
selected, _, err := task.selectFlushedSegment()
|
|
assert.NoError(t, err)
|
|
assert.Empty(t, selected, "import segment with commit_ts=5000 must not be selected at triggerTs=3000")
|
|
})
|
|
|
|
t.Run("import segment selected when trigger pos > commit_timestamp", func(t *testing.T) {
|
|
// triggerTs=6000 > commit_ts=5000 → segment must be selected
|
|
task := makeTask(6000)
|
|
selected, _, err := task.selectFlushedSegment()
|
|
assert.NoError(t, err)
|
|
assert.Len(t, selected, 1, "import segment with commit_ts=5000 must be selected at triggerTs=6000")
|
|
assert.Equal(t, int64(777), selected[0].GetID())
|
|
})
|
|
}
|