diff --git a/.golangci.yml b/.golangci.yml index 91895ce0cc11..0d343d6e67e8 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -8,6 +8,9 @@ run: - scripts - internal/core - cmake_build + - mmap + - data + - ci skip-files: - partial_search_test.go diff --git a/go.mod b/go.mod index 611d60687d9b..281be7e1e975 100644 --- a/go.mod +++ b/go.mod @@ -69,6 +69,7 @@ require ( github.com/milvus-io/milvus/pkg v0.0.0-00010101000000-000000000000 github.com/pkg/errors v0.9.1 github.com/valyala/fastjson v1.6.4 + google.golang.org/protobuf v1.33.0 gopkg.in/yaml.v3 v3.0.1 ) @@ -234,7 +235,6 @@ require ( google.golang.org/genproto v0.0.0-20230706204954-ccb25ca9f130 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20230629202037-9506855d4529 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20240401170217-c3f982113cda // indirect - google.golang.org/protobuf v1.33.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/ini.v1 v1.67.0 // indirect gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect @@ -248,6 +248,7 @@ replace ( github.com/bketelsen/crypt => github.com/bketelsen/crypt v0.0.4 // Fix security alert for core-os/etcd github.com/expr-lang/expr => github.com/SimFG/expr v0.0.0-20231218130003-94d085776dc5 github.com/go-kit/kit => github.com/go-kit/kit v0.1.0 + github.com/greatroar/blobloom => github.com/milvus-io/blobloom v0.0.0-20240603110411-471ae49f3b93 // github.com/milvus-io/milvus-storage/go => ../milvus-storage/go github.com/milvus-io/milvus/pkg => ./pkg github.com/streamnative/pulsarctl => github.com/xiaofan-luan/pulsarctl v0.5.1 diff --git a/internal/datacoord/compaction.go b/internal/datacoord/compaction.go index 2fa65ddbcb45..4044e90bd260 100644 --- a/internal/datacoord/compaction.go +++ b/internal/datacoord/compaction.go @@ -275,8 +275,37 @@ func (c *compactionPlanHandler) loadMeta() { triggers := c.meta.(*meta).compactionTaskMeta.GetCompactionTasks() for _, tasks := range triggers { for _, task := range tasks { - if task.State != datapb.CompactionTaskState_completed && task.State != datapb.CompactionTaskState_cleaned { - c.enqueueCompaction(task) + state := task.GetState() + if state == datapb.CompactionTaskState_completed || + state == datapb.CompactionTaskState_cleaned || + state == datapb.CompactionTaskState_unknown { + log.Info("compactionPlanHandler loadMeta abandon compactionTask", + zap.Int64("planID", task.GetPlanID()), + zap.String("State", task.GetState().String())) + continue + } else { + t, err := c.createCompactTask(task) + if err != nil { + log.Warn("compactionPlanHandler loadMeta create compactionTask failed", + zap.Int64("planID", task.GetPlanID()), + zap.String("State", task.GetState().String())) + continue + } + if t.NeedReAssignNodeID() { + c.submitTask(t) + log.Info("compactionPlanHandler loadMeta submitTask", + zap.Int64("planID", t.GetPlanID()), + zap.Int64("triggerID", t.GetTriggerID()), + zap.Int64("collectionID", t.GetCollectionID()), + zap.String("state", t.GetState().String())) + } else { + c.restoreTask(t) + log.Info("compactionPlanHandler loadMeta restoreTask", + zap.Int64("planID", t.GetPlanID()), + zap.Int64("triggerID", t.GetTriggerID()), + zap.Int64("collectionID", t.GetCollectionID()), + zap.String("state", t.GetState().String())) + } } } } @@ -467,6 +496,8 @@ func (c *compactionPlanHandler) removeTasksByChannel(channel string) { } func (c *compactionPlanHandler) submitTask(t CompactionTask) { + _, span := otel.Tracer(typeutil.DataCoordRole).Start(context.Background(), fmt.Sprintf("Compaction-%s", t.GetType())) + t.SetSpan(span) c.mu.Lock() c.queueTasks[t.GetPlanID()] = t c.mu.Unlock() @@ -475,6 +506,8 @@ func (c *compactionPlanHandler) submitTask(t CompactionTask) { // restoreTask used to restore Task from etcd func (c *compactionPlanHandler) restoreTask(t CompactionTask) { + _, span := otel.Tracer(typeutil.DataCoordRole).Start(context.Background(), fmt.Sprintf("Compaction-%s", t.GetType())) + t.SetSpan(span) c.executingMu.Lock() c.executingTasks[t.GetPlanID()] = t c.executingMu.Unlock() @@ -505,38 +538,23 @@ func (c *compactionPlanHandler) enqueueCompaction(task *datapb.CompactionTask) e if c.isFull() { return errCompactionBusy } - // TODO change to set this on scheduling task - exist, succeed := c.checkAndSetSegmentsCompacting(task) - if !exist { - return merr.WrapErrIllegalCompactionPlan("segment not exist") - } - if !succeed { - return merr.WrapErrCompactionPlanConflict("segment is compacting") - } - - // TODO change to set this on scheduling task - t := c.createCompactTask(task) - if t == nil { - return merr.WrapErrIllegalCompactionPlan("illegal compaction type") + t, err := c.createCompactTask(task) + if err != nil { + return err } - if task.StartTime != 0 { - t.SetTask(t.ShadowClone(setStartTime(time.Now().Unix()))) - err := t.SaveTaskMeta() - if err != nil { - return err - } + t.SetTask(t.ShadowClone(setStartTime(time.Now().Unix()))) + err = t.SaveTaskMeta() + if err != nil { + c.meta.SetSegmentsCompacting(t.GetInputSegments(), false) + return err } - - _, span := otel.Tracer(typeutil.DataCoordRole).Start(context.Background(), fmt.Sprintf("Compaction-%s", task.GetType())) - t.SetSpan(span) - c.submitTask(t) log.Info("Compaction plan submitted") return nil } // set segments compacting, one segment can only participate one compactionTask -func (c *compactionPlanHandler) createCompactTask(t *datapb.CompactionTask) CompactionTask { +func (c *compactionPlanHandler) createCompactTask(t *datapb.CompactionTask) (CompactionTask, error) { var task CompactionTask switch t.GetType() { case datapb.CompactionType_MixCompaction: @@ -559,19 +577,17 @@ func (c *compactionPlanHandler) createCompactTask(t *datapb.CompactionTask) Comp handler: c.handler, analyzeScheduler: c.analyzeScheduler, } + default: + return nil, merr.WrapErrIllegalCompactionPlan("illegal compaction type") } - return task -} - -// set segments compacting, one segment can only participate one compactionTask -func (c *compactionPlanHandler) setSegmentsCompacting(task CompactionTask, compacting bool) { - for _, segmentID := range task.GetInputSegments() { - c.meta.SetSegmentCompacting(segmentID, compacting) + exist, succeed := c.meta.CheckAndSetSegmentsCompacting(t.GetInputSegments()) + if !exist { + return nil, merr.WrapErrIllegalCompactionPlan("segment not exist") } -} - -func (c *compactionPlanHandler) checkAndSetSegmentsCompacting(task *datapb.CompactionTask) (bool, bool) { - return c.meta.CheckAndSetSegmentsCompacting(task.GetInputSegments()) + if !succeed { + return nil, merr.WrapErrCompactionPlanConflict("segment is compacting") + } + return task, nil } func (c *compactionPlanHandler) assignNodeIDs(tasks []CompactionTask) { diff --git a/internal/datacoord/compaction_task_clustering.go b/internal/datacoord/compaction_task_clustering.go index eaf7117fd03a..e5de48cc3ea5 100644 --- a/internal/datacoord/compaction_task_clustering.go +++ b/internal/datacoord/compaction_task_clustering.go @@ -210,7 +210,7 @@ func (t *clusteringCompactionTask) processExecuting() error { if errors.Is(err, merr.ErrNodeNotFound) { log.Warn("GetCompactionPlanResult fail", zap.Error(err)) // todo reassign node ID - t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_pipelining), setNodeID(0)) + t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_pipelining), setNodeID(NullNodeID)) return nil } return err @@ -230,7 +230,7 @@ func (t *clusteringCompactionTask) processExecuting() error { return segment.GetSegmentID() }) - _, metricMutation, err := t.meta.CompleteCompactionMutation(t.GetPlan(), t.result) + _, metricMutation, err := t.meta.CompleteCompactionMutation(t.CompactionTask, t.result) if err != nil { return err } @@ -334,9 +334,11 @@ func (t *clusteringCompactionTask) processAnalyzing() error { } func (t *clusteringCompactionTask) resetSegmentCompacting() { - for _, segmentBinlogs := range t.GetPlan().GetSegmentBinlogs() { - t.meta.SetSegmentCompacting(segmentBinlogs.GetSegmentID(), false) + var segmentIDs []UniqueID + for _, binLogs := range t.GetPlan().GetSegmentBinlogs() { + segmentIDs = append(segmentIDs, binLogs.GetSegmentID()) } + t.meta.SetSegmentsCompacting(segmentIDs, false) } func (t *clusteringCompactionTask) processFailedOrTimeout() error { @@ -412,7 +414,7 @@ func (t *clusteringCompactionTask) doCompact() error { err = t.sessions.Compaction(context.Background(), t.GetNodeID(), t.GetPlan()) if err != nil { log.Warn("Failed to notify compaction tasks to DataNode", zap.Error(err)) - t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_pipelining), setNodeID(0)) + t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_pipelining), setNodeID(NullNodeID)) return err } t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_executing)) diff --git a/internal/datacoord/compaction_task_l0.go b/internal/datacoord/compaction_task_l0.go index a18a14cac469..87b185db0ec5 100644 --- a/internal/datacoord/compaction_task_l0.go +++ b/internal/datacoord/compaction_task_l0.go @@ -74,7 +74,7 @@ func (t *l0CompactionTask) processPipelining() bool { err = t.sessions.Compaction(context.Background(), t.GetNodeID(), t.GetPlan()) if err != nil { log.Warn("Failed to notify compaction tasks to DataNode", zap.Error(err)) - t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_pipelining), setNodeID(0)) + t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_pipelining), setNodeID(NullNodeID)) return false } t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_executing)) @@ -85,7 +85,7 @@ func (t *l0CompactionTask) processExecuting() bool { result, err := t.sessions.GetCompactionPlanResult(t.GetNodeID(), t.GetPlanID()) if err != nil || result == nil { if errors.Is(err, merr.ErrNodeNotFound) { - t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_pipelining), setNodeID(0)) + t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_pipelining), setNodeID(NullNodeID)) } return false } @@ -186,7 +186,7 @@ func (t *l0CompactionTask) SetStartTime(startTime int64) { } func (t *l0CompactionTask) NeedReAssignNodeID() bool { - return t.GetState() == datapb.CompactionTaskState_pipelining && t.GetNodeID() == 0 + return t.GetState() == datapb.CompactionTaskState_pipelining && t.GetNodeID() == NullNodeID } func (t *l0CompactionTask) SetResult(result *datapb.CompactionPlanResult) { @@ -250,7 +250,7 @@ func (t *l0CompactionTask) BuildCompactionRequest() (*datapb.CompactionPlan, err //!info.isCompacting && !info.GetIsImporting() && info.GetLevel() != datapb.SegmentLevel_L0 && - info.GetDmlPosition().GetTimestamp() < t.GetPos().GetTimestamp() + info.GetStartPosition().GetTimestamp() < t.GetPos().GetTimestamp() })) if len(sealedSegments) == 0 { @@ -307,9 +307,11 @@ func (t *l0CompactionTask) processCompleted() bool { } func (t *l0CompactionTask) resetSegmentCompacting() { - for _, segmentBinlogs := range t.GetPlan().GetSegmentBinlogs() { - t.meta.SetSegmentCompacting(segmentBinlogs.GetSegmentID(), false) + var segmentIDs []UniqueID + for _, binLogs := range t.GetPlan().GetSegmentBinlogs() { + segmentIDs = append(segmentIDs, binLogs.GetSegmentID()) } + t.meta.SetSegmentsCompacting(segmentIDs, false) } func (t *l0CompactionTask) processTimeout() bool { diff --git a/internal/datacoord/compaction_task_mix.go b/internal/datacoord/compaction_task_mix.go index ab291cb89e69..ef796cb1faf6 100644 --- a/internal/datacoord/compaction_task_mix.go +++ b/internal/datacoord/compaction_task_mix.go @@ -32,6 +32,7 @@ func (t *mixCompactionTask) processPipelining() bool { } var err error t.plan, err = t.BuildCompactionRequest() + // Segment not found if err != nil { err2 := t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_failed), setFailReason(err.Error())) return err2 == nil @@ -39,7 +40,7 @@ func (t *mixCompactionTask) processPipelining() bool { err = t.sessions.Compaction(context.Background(), t.GetNodeID(), t.GetPlan()) if err != nil { log.Warn("Failed to notify compaction tasks to DataNode", zap.Error(err)) - t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_pipelining), setNodeID(0)) + t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_pipelining), setNodeID(NullNodeID)) return false } t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_executing)) @@ -59,7 +60,7 @@ func (t *mixCompactionTask) processExecuting() bool { result, err := t.sessions.GetCompactionPlanResult(t.GetNodeID(), t.GetPlanID()) if err != nil || result == nil { if errors.Is(err, merr.ErrNodeNotFound) { - t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_pipelining), setNodeID(0)) + t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_pipelining), setNodeID(NullNodeID)) } return false } @@ -82,13 +83,20 @@ func (t *mixCompactionTask) processExecuting() bool { } return t.processFailed() } - saveSuccess := t.saveSegmentMeta() - if !saveSuccess { + err2 := t.saveSegmentMeta() + if err2 != nil { + if errors.Is(err2, merr.ErrIllegalCompactionPlan) { + err3 := t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_failed)) + if err3 != nil { + log.Warn("fail to updateAndSaveTaskMeta") + } + return true + } return false } segments := []UniqueID{t.newSegment.GetID()} - err := t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_meta_saved), setResultSegments(segments)) - if err == nil { + err3 := t.updateAndSaveTaskMeta(setState(datapb.CompactionTaskState_meta_saved), setResultSegments(segments)) + if err3 == nil { return t.processMetaSaved() } return false @@ -110,18 +118,18 @@ func (t *mixCompactionTask) SaveTaskMeta() error { return t.saveTaskMeta(t.CompactionTask) } -func (t *mixCompactionTask) saveSegmentMeta() bool { +func (t *mixCompactionTask) saveSegmentMeta() error { log := log.With(zap.Int64("planID", t.GetPlanID()), zap.String("type", t.GetType().String())) // Also prepare metric updates. - newSegments, metricMutation, err := t.meta.CompleteCompactionMutation(t.GetPlan(), t.result) + newSegments, metricMutation, err := t.meta.CompleteCompactionMutation(t.CompactionTask, t.result) if err != nil { - return false + return err } // Apply metrics after successful meta update. t.newSegment = newSegments[0] metricMutation.commit() log.Info("mixCompactionTask success to save segment meta") - return true + return nil } func (t *mixCompactionTask) Process() bool { @@ -161,7 +169,7 @@ func (t *mixCompactionTask) GetLabel() string { } func (t *mixCompactionTask) NeedReAssignNodeID() bool { - return t.GetState() == datapb.CompactionTaskState_pipelining && t.GetNodeID() == 0 + return t.GetState() == datapb.CompactionTaskState_pipelining && t.GetNodeID() == NullNodeID } func (t *mixCompactionTask) processCompleted() bool { @@ -178,9 +186,11 @@ func (t *mixCompactionTask) processCompleted() bool { } func (t *mixCompactionTask) resetSegmentCompacting() { - for _, segmentBinlogs := range t.GetPlan().GetSegmentBinlogs() { - t.meta.SetSegmentCompacting(segmentBinlogs.GetSegmentID(), false) + var segmentIDs []UniqueID + for _, binLogs := range t.GetPlan().GetSegmentBinlogs() { + segmentIDs = append(segmentIDs, binLogs.GetSegmentID()) } + t.meta.SetSegmentsCompacting(segmentIDs, false) } func (t *mixCompactionTask) processTimeout() bool { diff --git a/internal/datacoord/compaction_test.go b/internal/datacoord/compaction_test.go index 561f36f2b365..cb48e6949677 100644 --- a/internal/datacoord/compaction_test.go +++ b/internal/datacoord/compaction_test.go @@ -529,6 +529,7 @@ func (s *CompactionPlanHandlerSuite) TestGetCompactionTask() { func (s *CompactionPlanHandlerSuite) TestExecCompactionPlan() { s.SetupTest() s.mockMeta.EXPECT().CheckAndSetSegmentsCompacting(mock.Anything).Return(true, true).Once() + s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything).Return(nil) handler := newCompactionPlanHandler(nil, s.mockSessMgr, s.mockCm, s.mockMeta, s.mockAlloc, nil, nil) task := &datapb.CompactionTask{ @@ -569,6 +570,7 @@ func (s *CompactionPlanHandlerSuite) TestCheckCompaction() { }, nil).Once() s.mockSessMgr.EXPECT().DropCompactionPlan(mock.Anything, mock.Anything).Return(nil) + s.mockMeta.EXPECT().SetSegmentsCompacting(mock.Anything, mock.Anything).Return() inTasks := map[int64]CompactionTask{ 1: &mixCompactionTask{ @@ -658,11 +660,11 @@ func (s *CompactionPlanHandlerSuite) TestCheckCompaction() { // s.mockMeta.EXPECT().UpdateSegmentsInfo(mock.Anything).Return(nil) s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything).Return(nil) s.mockMeta.EXPECT().CompleteCompactionMutation(mock.Anything, mock.Anything).RunAndReturn( - func(plan *datapb.CompactionPlan, result *datapb.CompactionPlanResult) ([]*SegmentInfo, *segMetricMutation, error) { - if plan.GetPlanID() == 2 { + func(t *datapb.CompactionTask, result *datapb.CompactionPlanResult) ([]*SegmentInfo, *segMetricMutation, error) { + if t.GetPlanID() == 2 { segment := NewSegmentInfo(&datapb.SegmentInfo{ID: 100}) return []*SegmentInfo{segment}, &segMetricMutation{}, nil - } else if plan.GetPlanID() == 6 { + } else if t.GetPlanID() == 6 { return nil, nil, errors.Errorf("intended error") } return nil, nil, errors.Errorf("unexpected error") @@ -706,7 +708,7 @@ func (s *CompactionPlanHandlerSuite) TestProcessCompleteCompaction() { // s.mockSessMgr.EXPECT().SyncSegments(mock.Anything, mock.Anything).Return(nil).Once() s.mockMeta.EXPECT().SaveCompactionTask(mock.Anything).Return(nil) - s.mockMeta.EXPECT().SetSegmentCompacting(mock.Anything, mock.Anything).Return().Twice() + s.mockMeta.EXPECT().SetSegmentsCompacting(mock.Anything, mock.Anything).Return().Once() segment := NewSegmentInfo(&datapb.SegmentInfo{ID: 100}) s.mockMeta.EXPECT().CompleteCompactionMutation(mock.Anything, mock.Anything).Return( []*SegmentInfo{segment}, @@ -749,13 +751,14 @@ func (s *CompactionPlanHandlerSuite) TestProcessCompleteCompaction() { task := &mixCompactionTask{ CompactionTask: &datapb.CompactionTask{ - PlanID: plan.GetPlanID(), - TriggerID: 1, - Type: plan.GetType(), - State: datapb.CompactionTaskState_executing, - NodeID: dataNodeID, + PlanID: plan.GetPlanID(), + TriggerID: 1, + Type: plan.GetType(), + State: datapb.CompactionTaskState_executing, + NodeID: dataNodeID, + InputSegments: []UniqueID{1, 2}, }, - plan: plan, + // plan: plan, sessions: s.mockSessMgr, meta: s.mockMeta, } diff --git a/internal/datacoord/meta.go b/internal/datacoord/meta.go index d37eb4535aa1..e421c043a92d 100644 --- a/internal/datacoord/meta.go +++ b/internal/datacoord/meta.go @@ -56,9 +56,9 @@ type CompactionMeta interface { SelectSegments(filters ...SegmentFilter) []*SegmentInfo GetHealthySegment(segID UniqueID) *SegmentInfo UpdateSegmentsInfo(operators ...UpdateOperator) error - SetSegmentCompacting(segmentID int64, compacting bool) + SetSegmentsCompacting(segmentID []int64, compacting bool) CheckAndSetSegmentsCompacting(segmentIDs []int64) (bool, bool) - CompleteCompactionMutation(plan *datapb.CompactionPlan, result *datapb.CompactionPlanResult) ([]*SegmentInfo, *segMetricMutation, error) + CompleteCompactionMutation(t *datapb.CompactionTask, result *datapb.CompactionPlanResult) ([]*SegmentInfo, *segMetricMutation, error) CleanPartitionStatsInfo(info *datapb.PartitionStatsInfo) error SaveCompactionTask(task *datapb.CompactionTask) error @@ -1168,7 +1168,7 @@ func (m *meta) GetSegmentsIDOfCollection(collectionID UniqueID) []UniqueID { }) } -// GetSegmentsIDOfCollection returns all segment ids which collection equals to provided `collectionID` +// GetSegmentsIDOfCollectionWithDropped returns all dropped segment ids which collection equals to provided `collectionID` func (m *meta) GetSegmentsIDOfCollectionWithDropped(collectionID UniqueID) []UniqueID { segments := m.SelectSegments(WithCollection(collectionID), SegmentFilterFunc(func(segment *SegmentInfo) bool { return segment != nil && @@ -1193,7 +1193,7 @@ func (m *meta) GetSegmentsIDOfPartition(collectionID, partitionID UniqueID) []Un }) } -// GetSegmentsIDOfPartition returns all segments ids which collection & partition equals to provided `collectionID`, `partitionID` +// GetSegmentsIDOfPartitionWithDropped returns all dropped segments ids which collection & partition equals to provided `collectionID`, `partitionID` func (m *meta) GetSegmentsIDOfPartitionWithDropped(collectionID, partitionID UniqueID) []UniqueID { segments := m.SelectSegments(WithCollection(collectionID), SegmentFilterFunc(func(segment *SegmentInfo) bool { return segment.GetState() != commonpb.SegmentState_SegmentStateNone && @@ -1310,24 +1310,29 @@ func (m *meta) SetSegmentCompacting(segmentID UniqueID, compacting bool) { // CheckAndSetSegmentsCompacting check all segments are not compacting // if true, set them compacting and return true // if false, skip setting and -func (m *meta) CheckAndSetSegmentsCompacting(segmentIDs []UniqueID) (exist, hasCompactingSegment bool) { +func (m *meta) CheckAndSetSegmentsCompacting(segmentIDs []UniqueID) (exist, canDo bool) { m.Lock() defer m.Unlock() + var hasCompacting bool + exist = true for _, segmentID := range segmentIDs { seg := m.segments.GetSegment(segmentID) if seg != nil { - hasCompactingSegment = seg.isCompacting + if seg.isCompacting { + hasCompacting = true + } } else { - return false, false + exist = false + break } } - if hasCompactingSegment { - return true, false - } - for _, segmentID := range segmentIDs { - m.segments.SetIsCompacting(segmentID, true) + canDo = exist && !hasCompacting + if canDo { + for _, segmentID := range segmentIDs { + m.segments.SetIsCompacting(segmentID, true) + } } - return true, true + return exist, canDo } func (m *meta) SetSegmentsCompacting(segmentIDs []UniqueID, compacting bool) { @@ -1346,19 +1351,16 @@ func (m *meta) SetSegmentLevel(segmentID UniqueID, level datapb.SegmentLevel) { m.segments.SetLevel(segmentID, level) } -func (m *meta) CompleteCompactionMutation(plan *datapb.CompactionPlan, result *datapb.CompactionPlanResult) ([]*SegmentInfo, *segMetricMutation, error) { - m.Lock() - defer m.Unlock() - - log := log.With(zap.Int64("planID", plan.GetPlanID()), zap.String("type", plan.GetType().String())) +func (m *meta) completeClusterCompactionMutation(t *datapb.CompactionTask, result *datapb.CompactionPlanResult) ([]*SegmentInfo, *segMetricMutation, error) { + log := log.With(zap.Int64("planID", t.GetPlanID()), zap.String("type", t.GetType().String())) metricMutation := &segMetricMutation{stateChange: make(map[string]map[string]int)} var compactFromSegIDs []int64 var latestCompactFromSegments []*SegmentInfo - for _, segmentBinlogs := range plan.GetSegmentBinlogs() { - segment := m.segments.GetSegment(segmentBinlogs.GetSegmentID()) + for _, segmentID := range t.GetInputSegments() { + segment := m.segments.GetSegment(segmentID) if segment == nil { - return nil, nil, merr.WrapErrSegmentNotFound(segmentBinlogs.GetSegmentID()) + return nil, nil, merr.WrapErrSegmentNotFound(segmentID) } cloned := segment.Clone() @@ -1372,15 +1374,6 @@ func (m *meta) CompleteCompactionMutation(plan *datapb.CompactionPlan, result *d updateSegStateAndPrepareMetrics(cloned, commonpb.SegmentState_Dropped, metricMutation) } - logIDsFromPlan := make(map[int64]struct{}) - for _, segBinlogs := range plan.GetSegmentBinlogs() { - for _, fieldBinlog := range segBinlogs.GetDeltalogs() { - for _, binlog := range fieldBinlog.GetBinlogs() { - logIDsFromPlan[binlog.GetLogID()] = struct{}{} - } - } - } - getMinPosition := func(positions []*msgpb.MsgPosition) *msgpb.MsgPosition { var minPos *msgpb.MsgPosition for _, pos := range positions { @@ -1391,90 +1384,112 @@ func (m *meta) CompleteCompactionMutation(plan *datapb.CompactionPlan, result *d } return minPos } - - if plan.GetType() == datapb.CompactionType_ClusteringCompaction { - newSegments := make([]*SegmentInfo, 0) - for _, seg := range result.GetSegments() { - segmentInfo := &datapb.SegmentInfo{ - ID: seg.GetSegmentID(), - CollectionID: latestCompactFromSegments[0].CollectionID, - PartitionID: latestCompactFromSegments[0].PartitionID, - InsertChannel: plan.GetChannel(), - NumOfRows: seg.NumOfRows, - State: commonpb.SegmentState_Flushed, - MaxRowNum: latestCompactFromSegments[0].MaxRowNum, - Binlogs: seg.GetInsertLogs(), - Statslogs: seg.GetField2StatslogPaths(), - CreatedByCompaction: true, - CompactionFrom: compactFromSegIDs, - LastExpireTime: tsoutil.ComposeTSByTime(time.Unix(plan.GetStartTime(), 0), 0), - Level: datapb.SegmentLevel_L2, - StartPosition: getMinPosition(lo.Map(latestCompactFromSegments, func(info *SegmentInfo, _ int) *msgpb.MsgPosition { - return info.GetStartPosition() - })), - DmlPosition: getMinPosition(lo.Map(latestCompactFromSegments, func(info *SegmentInfo, _ int) *msgpb.MsgPosition { - return info.GetDmlPosition() - })), - } - segment := NewSegmentInfo(segmentInfo) - newSegments = append(newSegments, segment) - metricMutation.addNewSeg(segment.GetState(), segment.GetLevel(), segment.GetNumOfRows()) - } - compactionTo := make([]UniqueID, 0, len(newSegments)) - for _, s := range newSegments { - compactionTo = append(compactionTo, s.GetID()) + newSegments := make([]*SegmentInfo, 0) + for _, seg := range result.GetSegments() { + segmentInfo := &datapb.SegmentInfo{ + ID: seg.GetSegmentID(), + CollectionID: latestCompactFromSegments[0].CollectionID, + PartitionID: latestCompactFromSegments[0].PartitionID, + InsertChannel: t.GetChannel(), + NumOfRows: seg.NumOfRows, + State: commonpb.SegmentState_Flushed, + MaxRowNum: latestCompactFromSegments[0].MaxRowNum, + Binlogs: seg.GetInsertLogs(), + Statslogs: seg.GetField2StatslogPaths(), + CreatedByCompaction: true, + CompactionFrom: compactFromSegIDs, + LastExpireTime: tsoutil.ComposeTSByTime(time.Unix(t.GetStartTime(), 0), 0), + Level: datapb.SegmentLevel_L2, + StartPosition: getMinPosition(lo.Map(latestCompactFromSegments, func(info *SegmentInfo, _ int) *msgpb.MsgPosition { + return info.GetStartPosition() + })), + DmlPosition: getMinPosition(lo.Map(latestCompactFromSegments, func(info *SegmentInfo, _ int) *msgpb.MsgPosition { + return info.GetDmlPosition() + })), } + segment := NewSegmentInfo(segmentInfo) + newSegments = append(newSegments, segment) + metricMutation.addNewSeg(segment.GetState(), segment.GetLevel(), segment.GetNumOfRows()) + } + compactionTo := make([]UniqueID, 0, len(newSegments)) + for _, s := range newSegments { + compactionTo = append(compactionTo, s.GetID()) + } - log.Info("meta update: prepare for complete compaction mutation - complete", - zap.Int64("collectionID", latestCompactFromSegments[0].CollectionID), - zap.Int64("partitionID", latestCompactFromSegments[0].PartitionID), - zap.Any("compacted from", compactFromSegIDs), - zap.Any("compacted to", compactionTo)) + log.Info("meta update: prepare for complete compaction mutation - complete", + zap.Int64("collectionID", latestCompactFromSegments[0].CollectionID), + zap.Int64("partitionID", latestCompactFromSegments[0].PartitionID), + zap.Any("compacted from", compactFromSegIDs), + zap.Any("compacted to", compactionTo)) - compactFromInfos := lo.Map(latestCompactFromSegments, func(info *SegmentInfo, _ int) *datapb.SegmentInfo { - return info.SegmentInfo - }) + compactFromInfos := lo.Map(latestCompactFromSegments, func(info *SegmentInfo, _ int) *datapb.SegmentInfo { + return info.SegmentInfo + }) - newSegmentInfos := lo.Map(newSegments, func(info *SegmentInfo, _ int) *datapb.SegmentInfo { - return info.SegmentInfo - }) + newSegmentInfos := lo.Map(newSegments, func(info *SegmentInfo, _ int) *datapb.SegmentInfo { + return info.SegmentInfo + }) + + binlogs := make([]metastore.BinlogsIncrement, 0) + for _, seg := range newSegmentInfos { + binlogs = append(binlogs, metastore.BinlogsIncrement{Segment: seg}) + } + if err := m.catalog.AlterSegments(m.ctx, append(compactFromInfos, newSegmentInfos...), binlogs...); err != nil { + log.Warn("fail to alter segments and new segment", zap.Error(err)) + return nil, nil, err + } + lo.ForEach(latestCompactFromSegments, func(info *SegmentInfo, _ int) { + m.segments.SetSegment(info.GetID(), info) + }) + lo.ForEach(newSegments, func(info *SegmentInfo, _ int) { + m.segments.SetSegment(info.GetID(), info) + }) + return newSegments, metricMutation, nil +} + +func (m *meta) completeMixCompactionMutation(t *datapb.CompactionTask, result *datapb.CompactionPlanResult) ([]*SegmentInfo, *segMetricMutation, error) { + log := log.With(zap.Int64("planID", t.GetPlanID()), zap.String("type", t.GetType().String())) - binlogs := make([]metastore.BinlogsIncrement, 0) - for _, seg := range newSegmentInfos { - binlogs = append(binlogs, metastore.BinlogsIncrement{Segment: seg}) + metricMutation := &segMetricMutation{stateChange: make(map[string]map[string]int)} + var compactFromSegIDs []int64 + var latestCompactFromSegments []*SegmentInfo + for _, segmentID := range t.GetInputSegments() { + segment := m.segments.GetSegment(segmentID) + if segment == nil { + return nil, nil, merr.WrapErrSegmentNotFound(segmentID) } - if err := m.catalog.AlterSegments(m.ctx, append(compactFromInfos, newSegmentInfos...), binlogs...); err != nil { - log.Warn("fail to alter segments and new segment", zap.Error(err)) - return nil, nil, err + + cloned := segment.Clone() + cloned.DroppedAt = uint64(time.Now().UnixNano()) + cloned.Compacted = true + + latestCompactFromSegments = append(latestCompactFromSegments, cloned) + compactFromSegIDs = append(compactFromSegIDs, cloned.GetID()) + + // metrics mutation for compaction from segments + updateSegStateAndPrepareMetrics(cloned, commonpb.SegmentState_Dropped, metricMutation) + } + + getMinPosition := func(positions []*msgpb.MsgPosition) *msgpb.MsgPosition { + var minPos *msgpb.MsgPosition + for _, pos := range positions { + if minPos == nil || + pos != nil && pos.GetTimestamp() < minPos.GetTimestamp() { + minPos = pos + } } - lo.ForEach(latestCompactFromSegments, func(info *SegmentInfo, _ int) { - m.segments.SetSegment(info.GetID(), info) - }) - lo.ForEach(newSegments, func(info *SegmentInfo, _ int) { - m.segments.SetSegment(info.GetID(), info) - }) - return newSegments, metricMutation, nil + return minPos } // MixCompaction / MergeCompaction will generates one and only one segment compactToSegment := result.GetSegments()[0] - // copy new deltalogs in compactFrom segments to compactTo segments. - // TODO: Not needed when enable L0 segments. - newDeltalogs, err := m.copyNewDeltalogs(latestCompactFromSegments, logIDsFromPlan, compactToSegment.GetSegmentID()) - if err != nil { - return nil, nil, err - } - if len(newDeltalogs) > 0 { - compactToSegment.Deltalogs = append(compactToSegment.GetDeltalogs(), &datapb.FieldBinlog{Binlogs: newDeltalogs}) - } - compactToSegmentInfo := NewSegmentInfo( &datapb.SegmentInfo{ ID: compactToSegment.GetSegmentID(), CollectionID: latestCompactFromSegments[0].CollectionID, PartitionID: latestCompactFromSegments[0].PartitionID, - InsertChannel: plan.GetChannel(), + InsertChannel: t.GetChannel(), NumOfRows: compactToSegment.NumOfRows, State: commonpb.SegmentState_Flushed, MaxRowNum: latestCompactFromSegments[0].MaxRowNum, @@ -1484,7 +1499,7 @@ func (m *meta) CompleteCompactionMutation(plan *datapb.CompactionPlan, result *d CreatedByCompaction: true, CompactionFrom: compactFromSegIDs, - LastExpireTime: tsoutil.ComposeTSByTime(time.Unix(plan.GetStartTime(), 0), 0), + LastExpireTime: tsoutil.ComposeTSByTime(time.Unix(t.GetStartTime(), 0), 0), Level: datapb.SegmentLevel_L1, StartPosition: getMinPosition(lo.Map(latestCompactFromSegments, func(info *SegmentInfo, _ int) *msgpb.MsgPosition { @@ -1504,7 +1519,7 @@ func (m *meta) CompleteCompactionMutation(plan *datapb.CompactionPlan, result *d } log = log.With( - zap.String("channel", plan.GetChannel()), + zap.String("channel", t.GetChannel()), zap.Int64("partitionID", compactToSegmentInfo.GetPartitionID()), zap.Int64("compactTo segmentID", compactToSegmentInfo.GetID()), zap.Int64("compactTo segment numRows", compactToSegmentInfo.GetNumOfRows()), @@ -1537,36 +1552,16 @@ func (m *meta) CompleteCompactionMutation(plan *datapb.CompactionPlan, result *d return []*SegmentInfo{compactToSegmentInfo}, metricMutation, nil } -func (m *meta) copyNewDeltalogs(latestCompactFromInfos []*SegmentInfo, logIDsInPlan map[int64]struct{}, toSegment int64) ([]*datapb.Binlog, error) { - newBinlogs := []*datapb.Binlog{} - for _, seg := range latestCompactFromInfos { - for _, fieldLog := range seg.GetDeltalogs() { - for _, l := range fieldLog.GetBinlogs() { - if _, ok := logIDsInPlan[l.GetLogID()]; !ok { - fromKey := metautil.BuildDeltaLogPath(m.chunkManager.RootPath(), seg.CollectionID, seg.PartitionID, seg.ID, l.GetLogID()) - toKey := metautil.BuildDeltaLogPath(m.chunkManager.RootPath(), seg.CollectionID, seg.PartitionID, toSegment, l.GetLogID()) - log.Warn("found new deltalog in compactFrom segment, copying it...", - zap.Any("deltalog", l), - zap.Int64("copyFrom segmentID", seg.GetID()), - zap.Int64("copyTo segmentID", toSegment), - zap.String("copyFrom key", fromKey), - zap.String("copyTo key", toKey), - ) - - blob, err := m.chunkManager.Read(m.ctx, fromKey) - if err != nil { - return nil, err - } - - if err := m.chunkManager.Write(m.ctx, toKey, blob); err != nil { - return nil, err - } - newBinlogs = append(newBinlogs, l) - } - } - } +func (m *meta) CompleteCompactionMutation(t *datapb.CompactionTask, result *datapb.CompactionPlanResult) ([]*SegmentInfo, *segMetricMutation, error) { + m.Lock() + defer m.Unlock() + switch t.GetType() { + case datapb.CompactionType_MixCompaction: + return m.completeMixCompactionMutation(t, result) + case datapb.CompactionType_ClusteringCompaction: + return m.completeClusterCompactionMutation(t, result) } - return newBinlogs, nil + return nil, nil, merr.WrapErrIllegalCompactionPlan("illegal compaction type") } // buildSegment utility function for compose datapb.SegmentInfo struct with provided info diff --git a/internal/datacoord/meta_test.go b/internal/datacoord/meta_test.go index 4f4ecbde7039..94593313d158 100644 --- a/internal/datacoord/meta_test.go +++ b/internal/datacoord/meta_test.go @@ -209,33 +209,12 @@ func (suite *MetaBasicSuite) TestCompleteCompactionMutation() { } mockChMgr := mocks.NewChunkManager(suite.T()) - mockChMgr.EXPECT().RootPath().Return("mockroot").Times(4) - mockChMgr.EXPECT().Read(mock.Anything, mock.Anything).Return(nil, nil).Twice() - mockChMgr.EXPECT().Write(mock.Anything, mock.Anything, mock.Anything).Return(nil).Twice() - m := &meta{ catalog: &datacoord.Catalog{MetaKv: NewMetaMemoryKV()}, segments: latestSegments, chunkManager: mockChMgr, } - plan := &datapb.CompactionPlan{ - SegmentBinlogs: []*datapb.CompactionSegmentBinlogs{ - { - SegmentID: 1, - FieldBinlogs: m.GetSegment(1).GetBinlogs(), - Field2StatslogPaths: m.GetSegment(1).GetStatslogs(), - Deltalogs: m.GetSegment(1).GetDeltalogs()[:1], // compaction plan use only 1 deltalog - }, - { - SegmentID: 2, - FieldBinlogs: m.GetSegment(2).GetBinlogs(), - Field2StatslogPaths: m.GetSegment(2).GetStatslogs(), - Deltalogs: m.GetSegment(2).GetDeltalogs()[:1], // compaction plan use only 1 deltalog - }, - }, - } - compactToSeg := &datapb.CompactionSegment{ SegmentID: 3, InsertLogs: []*datapb.FieldBinlog{getFieldBinlogIDs(0, 50000)}, @@ -246,8 +225,13 @@ func (suite *MetaBasicSuite) TestCompleteCompactionMutation() { result := &datapb.CompactionPlanResult{ Segments: []*datapb.CompactionSegment{compactToSeg}, } + task := &datapb.CompactionTask{ + InputSegments: []UniqueID{1, 2}, + Type: datapb.CompactionType_MixCompaction, + } - infos, mutation, err := m.CompleteCompactionMutation(plan, result) + infos, mutation, err := m.CompleteCompactionMutation(task, result) + assert.NoError(suite.T(), err) suite.Equal(1, len(infos)) info := infos[0] suite.NoError(err) @@ -275,16 +259,6 @@ func (suite *MetaBasicSuite) TestCompleteCompactionMutation() { } } - deltalogs := info.GetDeltalogs() - deltalogIDs := []int64{} - for _, fbinlog := range deltalogs { - for _, blog := range fbinlog.GetBinlogs() { - suite.Empty(blog.GetLogPath()) - deltalogIDs = append(deltalogIDs, blog.GetLogID()) - } - } - suite.ElementsMatch([]int64{30001, 31001}, deltalogIDs) - // check compactFrom segments for _, segID := range []int64{1, 2} { seg := m.GetSegment(segID) @@ -856,7 +830,7 @@ func TestUpdateSegmentsInfo(t *testing.T) { }) } -func Test_meta_SetSegmentCompacting(t *testing.T) { +func Test_meta_SetSegmentsCompacting(t *testing.T) { type fields struct { client kv.MetaKv segments *SegmentsInfo @@ -899,62 +873,13 @@ func Test_meta_SetSegmentCompacting(t *testing.T) { catalog: &datacoord.Catalog{MetaKv: tt.fields.client}, segments: tt.fields.segments, } - m.SetSegmentCompacting(tt.args.segmentID, tt.args.compacting) + m.SetSegmentsCompacting([]UniqueID{tt.args.segmentID}, tt.args.compacting) segment := m.GetHealthySegment(tt.args.segmentID) assert.Equal(t, tt.args.compacting, segment.isCompacting) }) } } -func Test_meta_SetSegmentImporting(t *testing.T) { - type fields struct { - client kv.MetaKv - segments *SegmentsInfo - } - type args struct { - segmentID UniqueID - importing bool - } - tests := []struct { - name string - fields fields - args args - }{ - { - "test set segment importing", - fields{ - NewMetaMemoryKV(), - &SegmentsInfo{ - segments: map[int64]*SegmentInfo{ - 1: { - SegmentInfo: &datapb.SegmentInfo{ - ID: 1, - State: commonpb.SegmentState_Flushed, - IsImporting: false, - }, - }, - }, - }, - }, - args{ - segmentID: 1, - importing: true, - }, - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - m := &meta{ - catalog: &datacoord.Catalog{MetaKv: tt.fields.client}, - segments: tt.fields.segments, - } - m.SetSegmentCompacting(tt.args.segmentID, tt.args.importing) - segment := m.GetHealthySegment(tt.args.segmentID) - assert.Equal(t, tt.args.importing, segment.isCompacting) - }) - } -} - func Test_meta_GetSegmentsOfCollection(t *testing.T) { storedSegments := NewSegmentsInfo() diff --git a/internal/datacoord/mock_compaction_meta.go b/internal/datacoord/mock_compaction_meta.go index e47624c3f6a2..ec90d4b21699 100644 --- a/internal/datacoord/mock_compaction_meta.go +++ b/internal/datacoord/mock_compaction_meta.go @@ -114,34 +114,34 @@ func (_c *MockCompactionMeta_CleanPartitionStatsInfo_Call) RunAndReturn(run func return _c } -// 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) +// CompleteCompactionMutation provides a mock function with given fields: t, result +func (_m *MockCompactionMeta) CompleteCompactionMutation(t *datapb.CompactionTask, result *datapb.CompactionPlanResult) ([]*SegmentInfo, *segMetricMutation, error) { + ret := _m.Called(t, 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.CompactionTask, *datapb.CompactionPlanResult) ([]*SegmentInfo, *segMetricMutation, error)); ok { + return rf(t, result) } - if rf, ok := ret.Get(0).(func(*datapb.CompactionPlan, *datapb.CompactionPlanResult) []*SegmentInfo); ok { - r0 = rf(plan, result) + if rf, ok := ret.Get(0).(func(*datapb.CompactionTask, *datapb.CompactionPlanResult) []*SegmentInfo); ok { + r0 = rf(t, 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) + if rf, ok := ret.Get(1).(func(*datapb.CompactionTask, *datapb.CompactionPlanResult) *segMetricMutation); ok { + r1 = rf(t, 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) + if rf, ok := ret.Get(2).(func(*datapb.CompactionTask, *datapb.CompactionPlanResult) error); ok { + r2 = rf(t, result) } else { r2 = ret.Error(2) } @@ -155,15 +155,15 @@ type MockCompactionMeta_CompleteCompactionMutation_Call struct { } // CompleteCompactionMutation is a helper method to define mock.On call -// - plan *datapb.CompactionPlan +// - t *datapb.CompactionTask // - 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 (_e *MockCompactionMeta_Expecter) CompleteCompactionMutation(t interface{}, result interface{}) *MockCompactionMeta_CompleteCompactionMutation_Call { + return &MockCompactionMeta_CompleteCompactionMutation_Call{Call: _e.mock.On("CompleteCompactionMutation", t, result)} } -func (_c *MockCompactionMeta_CompleteCompactionMutation_Call) Run(run func(plan *datapb.CompactionPlan, result *datapb.CompactionPlanResult)) *MockCompactionMeta_CompleteCompactionMutation_Call { +func (_c *MockCompactionMeta_CompleteCompactionMutation_Call) Run(run func(t *datapb.CompactionTask, result *datapb.CompactionPlanResult)) *MockCompactionMeta_CompleteCompactionMutation_Call { _c.Call.Run(func(args mock.Arguments) { - run(args[0].(*datapb.CompactionPlan), args[1].(*datapb.CompactionPlanResult)) + run(args[0].(*datapb.CompactionTask), args[1].(*datapb.CompactionPlanResult)) }) return _c } @@ -173,7 +173,7 @@ func (_c *MockCompactionMeta_CompleteCompactionMutation_Call) Return(_a0 []*Segm return _c } -func (_c *MockCompactionMeta_CompleteCompactionMutation_Call) RunAndReturn(run func(*datapb.CompactionPlan, *datapb.CompactionPlanResult) ([]*SegmentInfo, *segMetricMutation, error)) *MockCompactionMeta_CompleteCompactionMutation_Call { +func (_c *MockCompactionMeta_CompleteCompactionMutation_Call) RunAndReturn(run func(*datapb.CompactionTask, *datapb.CompactionPlanResult) ([]*SegmentInfo, *segMetricMutation, error)) *MockCompactionMeta_CompleteCompactionMutation_Call { _c.Call.Return(run) return _c } @@ -666,36 +666,36 @@ func (_c *MockCompactionMeta_SelectSegments_Call) RunAndReturn(run func(...Segme return _c } -// SetSegmentCompacting provides a mock function with given fields: segmentID, compacting -func (_m *MockCompactionMeta) SetSegmentCompacting(segmentID int64, compacting bool) { +// SetSegmentsCompacting provides a mock function with given fields: segmentID, compacting +func (_m *MockCompactionMeta) SetSegmentsCompacting(segmentID []int64, compacting bool) { _m.Called(segmentID, compacting) } -// MockCompactionMeta_SetSegmentCompacting_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SetSegmentCompacting' -type MockCompactionMeta_SetSegmentCompacting_Call struct { +// MockCompactionMeta_SetSegmentsCompacting_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SetSegmentsCompacting' +type MockCompactionMeta_SetSegmentsCompacting_Call struct { *mock.Call } -// SetSegmentCompacting is a helper method to define mock.On call -// - segmentID int64 +// SetSegmentsCompacting is a helper method to define mock.On call +// - segmentID []int64 // - compacting bool -func (_e *MockCompactionMeta_Expecter) SetSegmentCompacting(segmentID interface{}, compacting interface{}) *MockCompactionMeta_SetSegmentCompacting_Call { - return &MockCompactionMeta_SetSegmentCompacting_Call{Call: _e.mock.On("SetSegmentCompacting", segmentID, compacting)} +func (_e *MockCompactionMeta_Expecter) SetSegmentsCompacting(segmentID interface{}, compacting interface{}) *MockCompactionMeta_SetSegmentsCompacting_Call { + return &MockCompactionMeta_SetSegmentsCompacting_Call{Call: _e.mock.On("SetSegmentsCompacting", segmentID, compacting)} } -func (_c *MockCompactionMeta_SetSegmentCompacting_Call) Run(run func(segmentID int64, compacting bool)) *MockCompactionMeta_SetSegmentCompacting_Call { +func (_c *MockCompactionMeta_SetSegmentsCompacting_Call) Run(run func(segmentID []int64, compacting bool)) *MockCompactionMeta_SetSegmentsCompacting_Call { _c.Call.Run(func(args mock.Arguments) { - run(args[0].(int64), args[1].(bool)) + run(args[0].([]int64), args[1].(bool)) }) return _c } -func (_c *MockCompactionMeta_SetSegmentCompacting_Call) Return() *MockCompactionMeta_SetSegmentCompacting_Call { +func (_c *MockCompactionMeta_SetSegmentsCompacting_Call) Return() *MockCompactionMeta_SetSegmentsCompacting_Call { _c.Call.Return() return _c } -func (_c *MockCompactionMeta_SetSegmentCompacting_Call) RunAndReturn(run func(int64, bool)) *MockCompactionMeta_SetSegmentCompacting_Call { +func (_c *MockCompactionMeta_SetSegmentsCompacting_Call) RunAndReturn(run func([]int64, bool)) *MockCompactionMeta_SetSegmentsCompacting_Call { _c.Call.Return(run) return _c }