mirror of
https://gitee.com/milvus-io/milvus.git
synced 2024-12-02 20:09:57 +08:00
fix: compact operation on datacoord meta should preform as a transcation (#29775)
issue: #29691 Signed-off-by: chyezh <chyezh@outlook.com>
This commit is contained in:
chyezh
GitHub
parent
d87726e4c7
commit
f2985d8454
@ -79,8 +79,7 @@ type CompactionMeta interface {
|
||||
UpdateSegmentsInfo(operators ...UpdateOperator) error
|
||||
SetSegmentCompacting(segmentID int64, compacting bool)
|
||||
|
||||
PrepareCompleteCompactionMutation(plan *datapb.CompactionPlan, result *datapb.CompactionPlanResult) ([]*SegmentInfo, *SegmentInfo, *segMetricMutation, error)
|
||||
alterMetaStoreAfterCompaction(segmentCompactTo *SegmentInfo, segmentsCompactFrom []*SegmentInfo) error
|
||||
CompleteCompactionMutation(plan *datapb.CompactionPlan, result *datapb.CompactionPlanResult) (*SegmentInfo, *segMetricMutation, error)
|
||||
}
|
||||
|
||||
var _ CompactionMeta = (*meta)(nil)
|
||||
@ -449,19 +448,12 @@ func (c *compactionPlanHandler) handleMergeCompactionResult(plan *datapb.Compact
|
||||
log.Info("meta has already been changed, skip meta change and retry sync segments")
|
||||
} else {
|
||||
// Also prepare metric updates.
|
||||
modSegments, newSegment, metricMutation, err := c.meta.PrepareCompleteCompactionMutation(plan, result)
|
||||
newSegment, metricMutation, err := c.meta.CompleteCompactionMutation(plan, result)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := c.meta.alterMetaStoreAfterCompaction(newSegment, modSegments); err != nil {
|
||||
log.Warn("fail to alter meta store", zap.Error(err))
|
||||
return err
|
||||
}
|
||||
|
||||
// Apply metrics after successful meta update.
|
||||
metricMutation.commit()
|
||||
|
||||
newSegmentInfo = newSegment
|
||||
}
|
||||
|
||||
|
||||
@ -339,34 +339,12 @@ func (s *CompactionPlanHandlerSuite) TestHandleMergeCompactionResult() {
|
||||
err := handler.handleMergeCompactionResult(plan, compactionResult)
|
||||
s.NoError(err)
|
||||
})
|
||||
s.Run("prepare error", func() {
|
||||
|
||||
s.Run("complete compaction mutation error", func() {
|
||||
s.SetupTest()
|
||||
s.mockMeta.EXPECT().GetHealthySegment(mock.Anything).Return(nil).Once()
|
||||
s.mockMeta.EXPECT().PrepareCompleteCompactionMutation(mock.Anything, mock.Anything).Return(
|
||||
nil, nil, nil, errors.New("mock error")).Once()
|
||||
|
||||
handler := newCompactionPlanHandler(s.mockSessMgr, s.mockCm, s.mockMeta, s.mockAlloc)
|
||||
handler.plans[plan.PlanID] = &compactionTask{dataNodeID: 111, plan: plan}
|
||||
compactionResult := &datapb.CompactionPlanResult{
|
||||
PlanID: plan.PlanID,
|
||||
Segments: []*datapb.CompactionSegment{
|
||||
{SegmentID: 4, NumOfRows: 15},
|
||||
},
|
||||
}
|
||||
|
||||
err := handler.handleMergeCompactionResult(plan, compactionResult)
|
||||
s.Error(err)
|
||||
})
|
||||
|
||||
s.Run("alter error", func() {
|
||||
s.SetupTest()
|
||||
s.mockMeta.EXPECT().GetHealthySegment(mock.Anything).Return(nil).Once()
|
||||
s.mockMeta.EXPECT().PrepareCompleteCompactionMutation(mock.Anything, mock.Anything).Return(
|
||||
[]*SegmentInfo{},
|
||||
NewSegmentInfo(&datapb.SegmentInfo{ID: 100}),
|
||||
&segMetricMutation{}, nil).Once()
|
||||
s.mockMeta.EXPECT().alterMetaStoreAfterCompaction(mock.Anything, mock.Anything).
|
||||
Return(errors.New("mock error")).Once()
|
||||
s.mockMeta.EXPECT().CompleteCompactionMutation(mock.Anything, mock.Anything).Return(
|
||||
nil, nil, errors.New("mock error")).Once()
|
||||
|
||||
handler := newCompactionPlanHandler(s.mockSessMgr, s.mockCm, s.mockMeta, s.mockAlloc)
|
||||
handler.plans[plan.PlanID] = &compactionTask{dataNodeID: 111, plan: plan}
|
||||
@ -384,12 +362,9 @@ func (s *CompactionPlanHandlerSuite) TestHandleMergeCompactionResult() {
|
||||
s.Run("sync segment error", func() {
|
||||
s.SetupTest()
|
||||
s.mockMeta.EXPECT().GetHealthySegment(mock.Anything).Return(nil).Once()
|
||||
s.mockMeta.EXPECT().PrepareCompleteCompactionMutation(mock.Anything, mock.Anything).Return(
|
||||
[]*SegmentInfo{},
|
||||
s.mockMeta.EXPECT().CompleteCompactionMutation(mock.Anything, mock.Anything).Return(
|
||||
NewSegmentInfo(&datapb.SegmentInfo{ID: 100}),
|
||||
&segMetricMutation{}, nil).Once()
|
||||
s.mockMeta.EXPECT().alterMetaStoreAfterCompaction(mock.Anything, mock.Anything).
|
||||
Return(nil).Once()
|
||||
s.mockSessMgr.EXPECT().SyncSegments(mock.Anything, mock.Anything).Return(errors.New("mock error")).Once()
|
||||
|
||||
handler := newCompactionPlanHandler(s.mockSessMgr, s.mockCm, s.mockMeta, s.mockAlloc)
|
||||
@ -425,12 +400,9 @@ func (s *CompactionPlanHandlerSuite) TestCompleteCompaction() {
|
||||
s.mockSessMgr.EXPECT().SyncSegments(mock.Anything, mock.Anything).Return(nil).Once()
|
||||
// mock for handleMergeCompactionResult
|
||||
s.mockMeta.EXPECT().GetHealthySegment(mock.Anything).Return(nil).Once()
|
||||
s.mockMeta.EXPECT().PrepareCompleteCompactionMutation(mock.Anything, mock.Anything).Return(
|
||||
[]*SegmentInfo{},
|
||||
s.mockMeta.EXPECT().CompleteCompactionMutation(mock.Anything, mock.Anything).Return(
|
||||
NewSegmentInfo(&datapb.SegmentInfo{ID: 100}),
|
||||
&segMetricMutation{}, nil).Once()
|
||||
s.mockMeta.EXPECT().alterMetaStoreAfterCompaction(mock.Anything, mock.Anything).
|
||||
Return(nil).Once()
|
||||
s.mockSch.EXPECT().Finish(mock.Anything, mock.Anything).Return()
|
||||
|
||||
dataNodeID := UniqueID(111)
|
||||
|
||||
@ -974,18 +974,35 @@ func (m *meta) SetSegmentCompacting(segmentID UniqueID, compacting bool) {
|
||||
m.segments.SetIsCompacting(segmentID, compacting)
|
||||
}
|
||||
|
||||
// PrepareCompleteCompactionMutation returns
|
||||
// CompleteCompactionMutation completes compaction mutation.
|
||||
func (m *meta) CompleteCompactionMutation(plan *datapb.CompactionPlan,
|
||||
result *datapb.CompactionPlanResult,
|
||||
) (*SegmentInfo, *segMetricMutation, error) {
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
modSegments, segment, metricMutation, err := m.prepareCompactionMutation(plan, result)
|
||||
if err != nil {
|
||||
log.Warn("fail to prepare for complete compaction mutation", zap.Error(err), zap.Int64("planID", plan.GetPlanID()))
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
if err := m.alterMetaStoreAfterCompaction(segment, modSegments); err != nil {
|
||||
log.Warn("fail to alert meta store", zap.Error(err), zap.Int64("segmentID", segment.GetID()), zap.Int64("planID", plan.GetPlanID()))
|
||||
return nil, nil, err
|
||||
}
|
||||
return segment, metricMutation, err
|
||||
}
|
||||
|
||||
// prepareCompactionMutation returns
|
||||
// - the segment info of compactedFrom segments after compaction to alter
|
||||
// - the segment info of compactedTo segment after compaction to add
|
||||
// The compactedTo segment could contain 0 numRows
|
||||
// TODO: too complicated
|
||||
func (m *meta) PrepareCompleteCompactionMutation(plan *datapb.CompactionPlan,
|
||||
func (m *meta) prepareCompactionMutation(plan *datapb.CompactionPlan,
|
||||
result *datapb.CompactionPlanResult,
|
||||
) ([]*SegmentInfo, *SegmentInfo, *segMetricMutation, error) {
|
||||
log.Info("meta update: prepare for complete compaction mutation")
|
||||
compactionLogs := plan.GetSegmentBinlogs()
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
|
||||
modSegments := make([]*SegmentInfo, 0, len(compactionLogs))
|
||||
|
||||
@ -1136,8 +1153,6 @@ func (m *meta) alterMetaStoreAfterCompaction(segmentCompactTo *SegmentInfo, segm
|
||||
for _, v := range segmentsCompactFrom {
|
||||
compactFromIDs = append(compactFromIDs, v.GetID())
|
||||
}
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
for _, s := range segmentsCompactFrom {
|
||||
m.segments.SetSegment(s.GetID(), s)
|
||||
}
|
||||
|
||||
@ -257,7 +257,7 @@ func (suite *MetaBasicSuite) TestPrepareCompleteCompactionMutation() {
|
||||
inCompactionResult := &datapb.CompactionPlanResult{
|
||||
Segments: []*datapb.CompactionSegment{inSegment},
|
||||
}
|
||||
afterCompact, newSegment, metricMutation, err := m.PrepareCompleteCompactionMutation(plan, inCompactionResult)
|
||||
afterCompact, newSegment, metricMutation, err := m.prepareCompactionMutation(plan, inCompactionResult)
|
||||
suite.NoError(err)
|
||||
suite.NotNil(afterCompact)
|
||||
suite.NotNil(newSegment)
|
||||
|
||||
@ -20,6 +20,70 @@ func (_m *MockCompactionMeta) EXPECT() *MockCompactionMeta_Expecter {
|
||||
return &MockCompactionMeta_Expecter{mock: &_m.Mock}
|
||||
}
|
||||
|
||||
// CompleteCompactionMutation provides a mock function with given fields: plan, result
|
||||
func (_m *MockCompactionMeta) CompleteCompactionMutation(plan *datapb.CompactionPlan, result *datapb.CompactionPlanResult) (*SegmentInfo, *segMetricMutation, error) {
|
||||
ret := _m.Called(plan, result)
|
||||
|
||||
var r0 *SegmentInfo
|
||||
var r1 *segMetricMutation
|
||||
var r2 error
|
||||
if rf, ok := ret.Get(0).(func(*datapb.CompactionPlan, *datapb.CompactionPlanResult) (*SegmentInfo, *segMetricMutation, error)); ok {
|
||||
return rf(plan, result)
|
||||
}
|
||||
if rf, ok := ret.Get(0).(func(*datapb.CompactionPlan, *datapb.CompactionPlanResult) *SegmentInfo); ok {
|
||||
r0 = rf(plan, result)
|
||||
} else {
|
||||
if ret.Get(0) != nil {
|
||||
r0 = ret.Get(0).(*SegmentInfo)
|
||||
}
|
||||
}
|
||||
|
||||
if rf, ok := ret.Get(1).(func(*datapb.CompactionPlan, *datapb.CompactionPlanResult) *segMetricMutation); ok {
|
||||
r1 = rf(plan, result)
|
||||
} else {
|
||||
if ret.Get(1) != nil {
|
||||
r1 = ret.Get(1).(*segMetricMutation)
|
||||
}
|
||||
}
|
||||
|
||||
if rf, ok := ret.Get(2).(func(*datapb.CompactionPlan, *datapb.CompactionPlanResult) error); ok {
|
||||
r2 = rf(plan, result)
|
||||
} else {
|
||||
r2 = ret.Error(2)
|
||||
}
|
||||
|
||||
return r0, r1, r2
|
||||
}
|
||||
|
||||
// MockCompactionMeta_CompleteCompactionMutation_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'CompleteCompactionMutation'
|
||||
type MockCompactionMeta_CompleteCompactionMutation_Call struct {
|
||||
*mock.Call
|
||||
}
|
||||
|
||||
// CompleteCompactionMutation is a helper method to define mock.On call
|
||||
// - plan *datapb.CompactionPlan
|
||||
// - result *datapb.CompactionPlanResult
|
||||
func (_e *MockCompactionMeta_Expecter) CompleteCompactionMutation(plan interface{}, result interface{}) *MockCompactionMeta_CompleteCompactionMutation_Call {
|
||||
return &MockCompactionMeta_CompleteCompactionMutation_Call{Call: _e.mock.On("CompleteCompactionMutation", plan, result)}
|
||||
}
|
||||
|
||||
func (_c *MockCompactionMeta_CompleteCompactionMutation_Call) Run(run func(plan *datapb.CompactionPlan, result *datapb.CompactionPlanResult)) *MockCompactionMeta_CompleteCompactionMutation_Call {
|
||||
_c.Call.Run(func(args mock.Arguments) {
|
||||
run(args[0].(*datapb.CompactionPlan), args[1].(*datapb.CompactionPlanResult))
|
||||
})
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockCompactionMeta_CompleteCompactionMutation_Call) Return(_a0 *SegmentInfo, _a1 *segMetricMutation, _a2 error) *MockCompactionMeta_CompleteCompactionMutation_Call {
|
||||
_c.Call.Return(_a0, _a1, _a2)
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockCompactionMeta_CompleteCompactionMutation_Call) RunAndReturn(run func(*datapb.CompactionPlan, *datapb.CompactionPlanResult) (*SegmentInfo, *segMetricMutation, error)) *MockCompactionMeta_CompleteCompactionMutation_Call {
|
||||
_c.Call.Return(run)
|
||||
return _c
|
||||
}
|
||||
|
||||
// GetHealthySegment provides a mock function with given fields: segID
|
||||
func (_m *MockCompactionMeta) GetHealthySegment(segID int64) *SegmentInfo {
|
||||
ret := _m.Called(segID)
|
||||
@ -64,79 +128,6 @@ func (_c *MockCompactionMeta_GetHealthySegment_Call) RunAndReturn(run func(int64
|
||||
return _c
|
||||
}
|
||||
|
||||
// PrepareCompleteCompactionMutation provides a mock function with given fields: plan, result
|
||||
func (_m *MockCompactionMeta) PrepareCompleteCompactionMutation(plan *datapb.CompactionPlan, result *datapb.CompactionPlanResult) ([]*SegmentInfo, *SegmentInfo, *segMetricMutation, error) {
|
||||
ret := _m.Called(plan, result)
|
||||
|
||||
var r0 []*SegmentInfo
|
||||
var r1 *SegmentInfo
|
||||
var r2 *segMetricMutation
|
||||
var r3 error
|
||||
if rf, ok := ret.Get(0).(func(*datapb.CompactionPlan, *datapb.CompactionPlanResult) ([]*SegmentInfo, *SegmentInfo, *segMetricMutation, error)); ok {
|
||||
return rf(plan, result)
|
||||
}
|
||||
if rf, ok := ret.Get(0).(func(*datapb.CompactionPlan, *datapb.CompactionPlanResult) []*SegmentInfo); ok {
|
||||
r0 = rf(plan, result)
|
||||
} else {
|
||||
if ret.Get(0) != nil {
|
||||
r0 = ret.Get(0).([]*SegmentInfo)
|
||||
}
|
||||
}
|
||||
|
||||
if rf, ok := ret.Get(1).(func(*datapb.CompactionPlan, *datapb.CompactionPlanResult) *SegmentInfo); ok {
|
||||
r1 = rf(plan, result)
|
||||
} else {
|
||||
if ret.Get(1) != nil {
|
||||
r1 = ret.Get(1).(*SegmentInfo)
|
||||
}
|
||||
}
|
||||
|
||||
if rf, ok := ret.Get(2).(func(*datapb.CompactionPlan, *datapb.CompactionPlanResult) *segMetricMutation); ok {
|
||||
r2 = rf(plan, result)
|
||||
} else {
|
||||
if ret.Get(2) != nil {
|
||||
r2 = ret.Get(2).(*segMetricMutation)
|
||||
}
|
||||
}
|
||||
|
||||
if rf, ok := ret.Get(3).(func(*datapb.CompactionPlan, *datapb.CompactionPlanResult) error); ok {
|
||||
r3 = rf(plan, result)
|
||||
} else {
|
||||
r3 = ret.Error(3)
|
||||
}
|
||||
|
||||
return r0, r1, r2, r3
|
||||
}
|
||||
|
||||
// MockCompactionMeta_PrepareCompleteCompactionMutation_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'PrepareCompleteCompactionMutation'
|
||||
type MockCompactionMeta_PrepareCompleteCompactionMutation_Call struct {
|
||||
*mock.Call
|
||||
}
|
||||
|
||||
// PrepareCompleteCompactionMutation is a helper method to define mock.On call
|
||||
// - plan *datapb.CompactionPlan
|
||||
// - result *datapb.CompactionPlanResult
|
||||
func (_e *MockCompactionMeta_Expecter) PrepareCompleteCompactionMutation(plan interface{}, result interface{}) *MockCompactionMeta_PrepareCompleteCompactionMutation_Call {
|
||||
return &MockCompactionMeta_PrepareCompleteCompactionMutation_Call{Call: _e.mock.On("PrepareCompleteCompactionMutation", plan, result)}
|
||||
}
|
||||
|
||||
func (_c *MockCompactionMeta_PrepareCompleteCompactionMutation_Call) Run(run func(plan *datapb.CompactionPlan, result *datapb.CompactionPlanResult)) *MockCompactionMeta_PrepareCompleteCompactionMutation_Call {
|
||||
_c.Call.Run(func(args mock.Arguments) {
|
||||
run(args[0].(*datapb.CompactionPlan), args[1].(*datapb.CompactionPlanResult))
|
||||
})
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockCompactionMeta_PrepareCompleteCompactionMutation_Call) Return(_a0 []*SegmentInfo, _a1 *SegmentInfo, _a2 *segMetricMutation, _a3 error) *MockCompactionMeta_PrepareCompleteCompactionMutation_Call {
|
||||
_c.Call.Return(_a0, _a1, _a2, _a3)
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockCompactionMeta_PrepareCompleteCompactionMutation_Call) RunAndReturn(run func(*datapb.CompactionPlan, *datapb.CompactionPlanResult) ([]*SegmentInfo, *SegmentInfo, *segMetricMutation, error)) *MockCompactionMeta_PrepareCompleteCompactionMutation_Call {
|
||||
_c.Call.Return(run)
|
||||
return _c
|
||||
}
|
||||
|
||||
// SelectSegments provides a mock function with given fields: selector
|
||||
func (_m *MockCompactionMeta) SelectSegments(selector SegmentInfoSelector) []*SegmentInfo {
|
||||
ret := _m.Called(selector)
|
||||
@ -270,49 +261,6 @@ func (_c *MockCompactionMeta_UpdateSegmentsInfo_Call) RunAndReturn(run func(...U
|
||||
return _c
|
||||
}
|
||||
|
||||
// alterMetaStoreAfterCompaction provides a mock function with given fields: segmentCompactTo, segmentsCompactFrom
|
||||
func (_m *MockCompactionMeta) alterMetaStoreAfterCompaction(segmentCompactTo *SegmentInfo, segmentsCompactFrom []*SegmentInfo) error {
|
||||
ret := _m.Called(segmentCompactTo, segmentsCompactFrom)
|
||||
|
||||
var r0 error
|
||||
if rf, ok := ret.Get(0).(func(*SegmentInfo, []*SegmentInfo) error); ok {
|
||||
r0 = rf(segmentCompactTo, segmentsCompactFrom)
|
||||
} else {
|
||||
r0 = ret.Error(0)
|
||||
}
|
||||
|
||||
return r0
|
||||
}
|
||||
|
||||
// MockCompactionMeta_alterMetaStoreAfterCompaction_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'alterMetaStoreAfterCompaction'
|
||||
type MockCompactionMeta_alterMetaStoreAfterCompaction_Call struct {
|
||||
*mock.Call
|
||||
}
|
||||
|
||||
// alterMetaStoreAfterCompaction is a helper method to define mock.On call
|
||||
// - segmentCompactTo *SegmentInfo
|
||||
// - segmentsCompactFrom []*SegmentInfo
|
||||
func (_e *MockCompactionMeta_Expecter) alterMetaStoreAfterCompaction(segmentCompactTo interface{}, segmentsCompactFrom interface{}) *MockCompactionMeta_alterMetaStoreAfterCompaction_Call {
|
||||
return &MockCompactionMeta_alterMetaStoreAfterCompaction_Call{Call: _e.mock.On("alterMetaStoreAfterCompaction", segmentCompactTo, segmentsCompactFrom)}
|
||||
}
|
||||
|
||||
func (_c *MockCompactionMeta_alterMetaStoreAfterCompaction_Call) Run(run func(segmentCompactTo *SegmentInfo, segmentsCompactFrom []*SegmentInfo)) *MockCompactionMeta_alterMetaStoreAfterCompaction_Call {
|
||||
_c.Call.Run(func(args mock.Arguments) {
|
||||
run(args[0].(*SegmentInfo), args[1].([]*SegmentInfo))
|
||||
})
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockCompactionMeta_alterMetaStoreAfterCompaction_Call) Return(_a0 error) *MockCompactionMeta_alterMetaStoreAfterCompaction_Call {
|
||||
_c.Call.Return(_a0)
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockCompactionMeta_alterMetaStoreAfterCompaction_Call) RunAndReturn(run func(*SegmentInfo, []*SegmentInfo) error) *MockCompactionMeta_alterMetaStoreAfterCompaction_Call {
|
||||
_c.Call.Return(run)
|
||||
return _c
|
||||
}
|
||||
|
||||
// NewMockCompactionMeta creates a new instance of MockCompactionMeta. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
|
||||
// The first argument is typically a *testing.T value.
|
||||
func NewMockCompactionMeta(t interface {
|
||||
|
||||
Loading…
Reference in New Issue
Block a user