fix: [2.6] remove interactive logic between import and L0 compaction (#48114)

The interactive coordination between import and L0 compaction
(pause/resume channels, skip collections, L0 import task waiting) causes
BulkInsert to fail with "too many input paths for binlog import" when L0
compaction triggers during import. Remove this coordination entirely so
import and L0 compaction operate independently.

issue: #47762

pr: #47768

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>

Author: bigsheeper

internal/datacoord/compaction_policy_l0.go

@@ -12,7 +12,6 @@ import (
 	"github.com/milvus-io/milvus/internal/datacoord/allocator"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
-	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
 )
@@ -23,53 +22,20 @@ type l0CompactionPolicy struct {
 
 	activeCollections *activeCollections
 	allocator         allocator.Allocator
-
-	// key: collectionID, value: reference count
-	skipCompactionCollections map[int64]int
-	skipLocker                sync.RWMutex
 }
 
 func newL0CompactionPolicy(meta *meta, allocator allocator.Allocator) *l0CompactionPolicy {
 	return &l0CompactionPolicy{
-		meta:                      meta,
-		activeCollections:         newActiveCollections(),
-		allocator:                 allocator,
-		skipCompactionCollections: make(map[int64]int),
+		meta:              meta,
+		activeCollections: newActiveCollections(),
+		allocator:         allocator,
 	}
 }
 
 func (policy *l0CompactionPolicy) Enable() bool {
 	return Params.DataCoordCfg.EnableAutoCompaction.GetAsBool()
 }
 
-func (policy *l0CompactionPolicy) AddSkipCollection(collectionID UniqueID) {
-	policy.skipLocker.Lock()
-	defer policy.skipLocker.Unlock()
-
-	if _, ok := policy.skipCompactionCollections[collectionID]; !ok {
-		policy.skipCompactionCollections[collectionID] = 1
-	} else {
-		policy.skipCompactionCollections[collectionID]++
-	}
-}
-
-func (policy *l0CompactionPolicy) RemoveSkipCollection(collectionID UniqueID) {
-	policy.skipLocker.Lock()
-	defer policy.skipLocker.Unlock()
-	refCount := policy.skipCompactionCollections[collectionID]
-	if refCount > 1 {
-		policy.skipCompactionCollections[collectionID]--
-	} else {
-		delete(policy.skipCompactionCollections, collectionID)
-	}
-}
-
-func (policy *l0CompactionPolicy) isSkipCollection(collectionID UniqueID) bool {
-	policy.skipLocker.RLock()
-	defer policy.skipLocker.RUnlock()
-	return policy.skipCompactionCollections[collectionID] > 0
-}
-
 // Notify policy to record the active updated(when adding a new L0 segment) collections.
 func (policy *l0CompactionPolicy) OnCollectionUpdate(collectionID int64) {
 	policy.activeCollections.Record(collectionID)
@@ -94,10 +60,6 @@ func (policy *l0CompactionPolicy) Trigger(ctx context.Context) (events map[Compa
 	}
 	events = make(map[CompactionTriggerType][]CompactionView)
 	for collID, segments := range latestCollSegs {
-		if policy.isSkipCollection(collID) {
-			continue
-		}
-
 		policy.activeCollections.Read(collID)
 		levelZeroSegments := lo.Filter(segments, func(info *SegmentInfo, _ int) bool {
 			return info.GetLevel() == datapb.SegmentLevel_L0
@@ -126,9 +88,6 @@ func (policy *l0CompactionPolicy) Trigger(ctx context.Context) (events map[Compa
 func (policy *l0CompactionPolicy) triggerOneCollection(ctx context.Context, collectionID int64) ([]CompactionView, int64, error) {
 	log := log.Ctx(ctx).With(zap.Int64("collectionID", collectionID))
 	log.Info("start trigger collection l0 compaction")
-	if policy.isSkipCollection(collectionID) {
-		return nil, 0, merr.WrapErrCollectionNotLoaded(collectionID, "the collection being paused by importing cannot do force l0 compaction")
-	}
 	allL0Segments := policy.meta.SelectSegments(ctx, WithCollection(collectionID), SegmentFilterFunc(func(segment *SegmentInfo) bool {
 		return isSegmentHealthy(segment) &&
 			isFlushed(segment) &&

internal/datacoord/compaction_policy_l0_test.go

@@ -117,29 +117,6 @@ func (s *L0CompactionPolicySuite) TestTriggerIdle() {
 		s.Equal(datapb.SegmentLevel_L0, view.Level)
 	}
 
-	// test for skip collection
-	s.l0_policy.AddSkipCollection(1)
-	s.l0_policy.AddSkipCollection(1)
-	// Test for skip collection
-	events, err = s.l0_policy.Trigger(context.Background())
-	s.NoError(err)
-	s.Empty(events)
-
-	// Test for skip collection with ref count
-	s.l0_policy.RemoveSkipCollection(1)
-	events, err = s.l0_policy.Trigger(context.Background())
-	s.NoError(err)
-	s.Empty(events)
-
-	s.l0_policy.RemoveSkipCollection(1)
-	events, err = s.l0_policy.Trigger(context.Background())
-	s.NoError(err)
-	s.Equal(1, len(events))
-	gotViews, ok = events[TriggerTypeLevelZeroViewIDLE]
-	s.True(ok)
-	s.NotNil(gotViews)
-	s.Equal(1, len(gotViews))
-
 	log.Info("cView", zap.String("string", cView.String()))
 }
 

internal/datacoord/compaction_trigger_v2.go

@@ -31,7 +31,6 @@ import (
 	"github.com/milvus-io/milvus/internal/util/sessionutil"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
-	"github.com/milvus-io/milvus/pkg/v2/proto/internalpb"
 	"github.com/milvus-io/milvus/pkg/v2/util/logutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
 )
@@ -97,8 +96,6 @@ type TriggerManager interface {
 	Stop()
 	OnCollectionUpdate(collectionID int64)
 	ManualTrigger(ctx context.Context, collectionID int64, clusteringCompaction bool, l0Compaction bool, targetSize int64) (UniqueID, error)
-	GetPauseCompactionChan(jobID, collectionID int64) <-chan struct{}
-	GetResumeCompactionChan(jobID, collectionID int64) <-chan struct{}
 	InitForceMergeMemoryQuerier(nodeManager session.NodeManager, mixCoord types.MixCoord, session sessionutil.SessionInterface)
 }
 
@@ -113,7 +110,6 @@ type CompactionTriggerManager struct {
 	allocator allocator.Allocator
 
 	meta                        *meta
-	importMeta                  ImportMeta
 	l0Policy                    *l0CompactionPolicy
 	clusteringPolicy            *clusteringCompactionPolicy
 	singlePolicy                *singleCompactionPolicy
@@ -122,28 +118,15 @@ type CompactionTriggerManager struct {
 
 	cancel  context.CancelFunc
 	closeWg sync.WaitGroup
-
-	l0Triggering bool
-	l0SigLock    *sync.Mutex
-	l0TickSig    *sync.Cond
-
-	pauseCompactionChanMap  map[int64]chan struct{}
-	resumeCompactionChanMap map[int64]chan struct{}
-	compactionChanLock      sync.Mutex
 }
 
-func NewCompactionTriggerManager(alloc allocator.Allocator, handler Handler, inspector CompactionInspector, meta *meta, importMeta ImportMeta) *CompactionTriggerManager {
+func NewCompactionTriggerManager(alloc allocator.Allocator, handler Handler, inspector CompactionInspector, meta *meta) *CompactionTriggerManager {
 	m := &CompactionTriggerManager{
-		allocator:               alloc,
-		handler:                 handler,
-		inspector:               inspector,
-		meta:                    meta,
-		importMeta:              importMeta,
-		pauseCompactionChanMap:  make(map[int64]chan struct{}),
-		resumeCompactionChanMap: make(map[int64]chan struct{}),
-	}
-	m.l0SigLock = &sync.Mutex{}
-	m.l0TickSig = sync.NewCond(m.l0SigLock)
+		allocator: alloc,
+		handler:   handler,
+		inspector: inspector,
+		meta:      meta,
+	}
 	m.l0Policy = newL0CompactionPolicy(meta, alloc)
 	m.clusteringPolicy = newClusteringCompactionPolicy(meta, m.allocator, m.handler)
 	m.singlePolicy = newSingleCompactionPolicy(meta, m.allocator, m.handler)
@@ -183,64 +166,6 @@ func (m *CompactionTriggerManager) Stop() {
 	m.closeWg.Wait()
 }
 
-func (m *CompactionTriggerManager) pauseL0SegmentCompacting(jobID, collectionID int64) {
-	m.l0Policy.AddSkipCollection(collectionID)
-	m.l0SigLock.Lock()
-	for m.l0Triggering {
-		m.l0TickSig.Wait()
-	}
-	m.l0SigLock.Unlock()
-	m.compactionChanLock.Lock()
-	if ch, ok := m.pauseCompactionChanMap[jobID]; ok {
-		close(ch)
-	}
-	m.compactionChanLock.Unlock()
-}
-
-func (m *CompactionTriggerManager) resumeL0SegmentCompacting(jobID, collectionID int64) {
-	m.compactionChanLock.Lock()
-	m.l0Policy.RemoveSkipCollection(collectionID)
-	if ch, ok := m.resumeCompactionChanMap[jobID]; ok {
-		close(ch)
-		delete(m.pauseCompactionChanMap, jobID)
-		delete(m.resumeCompactionChanMap, jobID)
-	}
-	m.compactionChanLock.Unlock()
-}
-
-func (m *CompactionTriggerManager) GetPauseCompactionChan(jobID, collectionID int64) <-chan struct{} {
-	m.compactionChanLock.Lock()
-	defer m.compactionChanLock.Unlock()
-	if ch, ok := m.pauseCompactionChanMap[jobID]; ok {
-		return ch
-	}
-	ch := make(chan struct{})
-	m.pauseCompactionChanMap[jobID] = ch
-	go m.pauseL0SegmentCompacting(jobID, collectionID)
-	return ch
-}
-
-func (m *CompactionTriggerManager) GetResumeCompactionChan(jobID, collectionID int64) <-chan struct{} {
-	m.compactionChanLock.Lock()
-	defer m.compactionChanLock.Unlock()
-	if ch, ok := m.resumeCompactionChanMap[jobID]; ok {
-		return ch
-	}
-	ch := make(chan struct{})
-	m.resumeCompactionChanMap[jobID] = ch
-	go m.resumeL0SegmentCompacting(jobID, collectionID)
-	return ch
-}
-
-func (m *CompactionTriggerManager) setL0Triggering(b bool) {
-	m.l0SigLock.Lock()
-	defer m.l0SigLock.Unlock()
-	m.l0Triggering = b
-	if !b {
-		m.l0TickSig.Broadcast()
-	}
-}
-
 func (m *CompactionTriggerManager) loop(ctx context.Context) {
 	defer logutil.LogPanic()
 
@@ -267,19 +192,16 @@ func (m *CompactionTriggerManager) loop(ctx context.Context) {
 				log.RatedInfo(10, "Skip trigger l0 compaction since inspector is full")
 				continue
 			}
-			m.setL0Triggering(true)
 			events, err := m.l0Policy.Trigger(ctx)
 			if err != nil {
 				log.Warn("Fail to trigger L0 policy", zap.Error(err))
-				m.setL0Triggering(false)
 				continue
 			}
 			if len(events) > 0 {
 				for triggerType, views := range events {
 					m.notify(ctx, triggerType, views)
 				}
 			}
-			m.setL0Triggering(false)
 		case <-clusteringTicker.C:
 			if !m.clusteringPolicy.Enable() {
 				continue
@@ -362,8 +284,6 @@ func (m *CompactionTriggerManager) ManualTrigger(ctx context.Context, collection
 		views, triggerID, err = m.forceMergePolicy.triggerOneCollection(ctx, collectionID, targetSize)
 		events[TriggerTypeForceMerge] = views
 	} else if isL0 {
-		m.setL0Triggering(true)
-		defer m.setL0Triggering(false)
 		views, triggerID, err = m.l0Policy.triggerOneCollection(ctx, collectionID)
 		events[TriggerTypeLevelZeroViewManual] = views
 	} else if isClustering {
@@ -444,12 +364,6 @@ func (m *CompactionTriggerManager) SubmitL0ViewToScheduler(ctx context.Context,
 		return
 	}
 
-	err = m.addL0ImportTaskForImport(ctx, collection, view)
-	if err != nil {
-		log.Warn("Failed to submit compaction view to scheduler because add l0 import task fail", zap.Error(err))
-		return
-	}
-
 	totalRows := lo.SumBy(view.GetSegmentsView(), func(segView *SegmentView) int64 {
 		return segView.NumOfRows
 	})
@@ -486,82 +400,6 @@ func (m *CompactionTriggerManager) SubmitL0ViewToScheduler(ctx context.Context,
 	)
 }
 
-func (m *CompactionTriggerManager) addL0ImportTaskForImport(ctx context.Context, collection *collectionInfo, view CompactionView) error {
-	// add l0 import task for the collection if the collection is importing
-	importJobs := m.importMeta.GetJobBy(ctx,
-		WithCollectionID(collection.ID),
-		WithoutJobStates(internalpb.ImportJobState_Completed, internalpb.ImportJobState_Failed),
-		WithoutL0Job(),
-	)
-	if len(importJobs) > 0 {
-		partitionID := view.GetGroupLabel().PartitionID
-		var (
-			fileSize        int64 = 0
-			totalRows       int64 = 0
-			totalMemorySize int64 = 0
-			importPaths     []string
-		)
-		idStart := time.Now().UnixMilli()
-		for _, segmentView := range view.GetSegmentsView() {
-			segInfo := m.meta.GetSegment(ctx, segmentView.ID)
-			if segInfo == nil {
-				continue
-			}
-			totalRows += int64(segmentView.DeltaRowCount)
-			totalMemorySize += int64(segmentView.DeltaSize)
-			for _, deltaLogs := range segInfo.GetDeltalogs() {
-				for _, binlog := range deltaLogs.GetBinlogs() {
-					fileSize += binlog.GetLogSize()
-					importPaths = append(importPaths, binlog.GetLogPath())
-				}
-			}
-		}
-		if len(importPaths) == 0 {
-			return nil
-		}
-
-		for i, job := range importJobs {
-			newTasks, err := NewImportTasks([][]*datapb.ImportFileStats{
-				{
-					{
-						ImportFile: &internalpb.ImportFile{
-							Id:    idStart + int64(i),
-							Paths: importPaths,
-						},
-						FileSize:        fileSize,
-						TotalRows:       totalRows,
-						TotalMemorySize: totalMemorySize,
-						HashedStats: map[string]*datapb.PartitionImportStats{
-							// which is vchannel
-							view.GetGroupLabel().Channel: {
-								PartitionRows: map[int64]int64{
-									partitionID: totalRows,
-								},
-								PartitionDataSize: map[int64]int64{
-									partitionID: totalMemorySize,
-								},
-							},
-						},
-					},
-				},
-			}, job, m.allocator, m.meta, m.importMeta, paramtable.Get().DataNodeCfg.FlushDeleteBufferBytes.GetAsInt())
-			if err != nil {
-				log.Warn("new import tasks failed", zap.Error(err))
-				return err
-			}
-			for _, t := range newTasks {
-				err = m.importMeta.AddTask(ctx, t)
-				if err != nil {
-					log.Warn("add new l0 import task from l0 compaction failed", WrapTaskLog(t, zap.Error(err))...)
-					return err
-				}
-				log.Info("add new l0 import task from l0 compaction", WrapTaskLog(t)...)
-			}
-		}
-	}
-	return nil
-}
-
 func (m *CompactionTriggerManager) SubmitClusteringViewToScheduler(ctx context.Context, view CompactionView) {
 	log := log.Ctx(ctx).With(zap.String("view", view.String()))
 	taskID, _, err := m.allocator.AllocN(2)