milvus-io--milvus
wehub-resource-sync
498b235461
Build and test / Build and test AMD64 Ubuntu 22.04 (push) Failing after 0s
Publish Builder / amazonlinux2023 (push) Failing after 1s
Build and test / UT for Go (push) Has been skipped
Publish KRTE Images / KRTE (push) Failing after 1s
Build and test / Integration Test (push) Has been skipped
Build and test / Upload Code Coverage (push) Has been skipped
Publish Builder / rockylinux9 (push) Failing after 1s
Publish Builder / ubuntu22.04 (push) Failing after 0s
Publish Builder / ubuntu24.04 (push) Failing after 0s
Publish Gpu Builder / publish-gpu-builder (push) Failing after 1s
Publish Test Images / PyTest (push) Failing after 0s
Build and test / UT for Cpp (push) Has been cancelled
804 行
30 KiB
Go
804 行
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())
|
|
})
|
|
}
|