fix: align target size compaction selection (#50114)

issue: #49991
See also: #49991

Target-size manual compaction should use the same candidate rules as
normal manual compaction. Share the candidate predicate so force merge
only plans healthy, flushed, visible, sorted, non-L0, non-L2 segments
and skips compacting, importing, or snapshot-protected segments.

Signed-off-by: yangxuan <xuan.yang@zilliz.com>

Signed-off-by: yangxuan <xuan.yang@zilliz.com>

internal/datacoord/compaction_policy_forcemerge.go

@@ -15,7 +15,6 @@ import (
 	"github.com/milvus-io/milvus/internal/util/sessionutil"
 	"github.com/milvus-io/milvus/pkg/v3/common"
 	"github.com/milvus-io/milvus/pkg/v3/log"
-	"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/metricsinfo"
 	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
@@ -107,12 +106,7 @@ func (policy *forceMergeCompactionPolicy) triggerOneCollection(
 	}
 
 	segments := policy.meta.SelectSegments(ctx, WithCollection(collectionID), SegmentFilterFunc(func(segment *SegmentInfo) bool {
-		return isSegmentHealthy(segment) &&
-			isFlushed(segment) &&
-			!segment.isCompacting &&
-			!segment.GetIsImporting() &&
-			segment.GetLevel() != datapb.SegmentLevel_L0 &&
-			!policy.meta.isSegmentCompactionProtected(segment.GetID())
+		return isNormalManualCompactionCandidate(policy.meta, segment)
 	}))
 
 	if len(segments) == 0 {

internal/datacoord/compaction_policy_forcemerge_test.go

@@ -40,6 +40,9 @@ func (s *ForceMergeCompactionPolicySuite) SetupTest() {
 	meta, err := newMemoryMeta(s.T())
 	s.Require().NoError(err)
 	for id, segment := range segments {
+		if segment.GetLevel() != datapb.SegmentLevel_L0 && segment.GetState() == commonpb.SegmentState_Flushed {
+			segment.IsSorted = true
+		}
 		meta.segments.SetSegment(id, segment)
 	}
 
@@ -372,3 +375,68 @@ func (s *ForceMergeCompactionPolicySuite) TestTriggerOneCollection_FilterSegment
 		}
 	}
 }
+
+func (s *ForceMergeCompactionPolicySuite) TestTriggerOneCollection_AlignsTargetSizeWithManualCompactionSelection() {
+	ctx := context.Background()
+	collectionID := int64(1)
+	targetSize := int64(1024 * 2) // 2GB in MB
+	triggerID := int64(100)
+
+	meta, err := newMemoryMeta(s.T())
+	s.Require().NoError(err)
+	policy := newForceMergeCompactionPolicy(meta, s.mockAlloc, s.mockHandler)
+	policy.SetTopologyQuerier(s.mockQuerier)
+
+	newSegment := func(id int64, level datapb.SegmentLevel, sorted bool, sortedByNamespace bool, compacting bool) *SegmentInfo {
+		return &SegmentInfo{
+			SegmentInfo: &datapb.SegmentInfo{
+				ID:                  id,
+				CollectionID:        collectionID,
+				PartitionID:         s.testLabel.PartitionID,
+				InsertChannel:       s.testLabel.Channel,
+				State:               commonpb.SegmentState_Flushed,
+				Level:               level,
+				IsSorted:            sorted,
+				IsSortedByNamespace: sortedByNamespace,
+				Binlogs:             genTestBinlogs(1, 10*MB),
+			},
+			isCompacting: compacting,
+		}
+	}
+
+	for _, segment := range []*SegmentInfo{
+		newSegment(10, datapb.SegmentLevel_L1, true, false, false),
+		newSegment(11, datapb.SegmentLevel_L1, false, true, false),
+		newSegment(12, datapb.SegmentLevel_L1, false, false, false),
+		newSegment(13, datapb.SegmentLevel_L2, true, false, false),
+		newSegment(14, datapb.SegmentLevel_L1, true, false, true),
+	} {
+		meta.segments.SetSegment(segment.GetID(), segment)
+	}
+
+	coll := &collectionInfo{
+		ID:         collectionID,
+		Schema:     newTestSchema(),
+		Properties: nil,
+	}
+	topology := &CollectionTopology{
+		CollectionID: collectionID,
+		NumReplicas:  1,
+	}
+
+	s.mockHandler.EXPECT().GetCollection(mock.Anything, collectionID).Return(coll, nil)
+	s.mockAlloc.EXPECT().AllocID(mock.Anything).Return(triggerID, nil)
+	s.mockQuerier.EXPECT().GetCollectionTopology(mock.Anything, collectionID).Return(topology, nil)
+
+	views, gotTriggerID, err := policy.triggerOneCollection(ctx, collectionID, targetSize)
+
+	s.NoError(err)
+	s.Equal(triggerID, gotTriggerID)
+	s.Require().Len(views, 1)
+
+	gotSegmentIDs := make([]int64, 0, len(views[0].GetSegmentsView()))
+	for _, segment := range views[0].GetSegmentsView() {
+		gotSegmentIDs = append(gotSegmentIDs, segment.ID)
+	}
+	s.ElementsMatch([]int64{10, 11}, gotSegmentIDs)
+}

internal/datacoord/compaction_trigger.go

@@ -560,15 +560,7 @@ func (t *compactionTrigger) getCandidates(signal *compactionSignal) ([]chanPartS
 	// default filter, select segments which could be compacted
 	filters := []SegmentFilter{
 		SegmentFilterFunc(func(segment *SegmentInfo) bool {
-			return isSegmentHealthy(segment) &&
-				isFlushed(segment) &&
-				!segment.isCompacting && // not compacting now
-				!segment.GetIsImporting() && // not importing now
-				segment.GetLevel() != datapb.SegmentLevel_L0 && // ignore level zero segments
-				segment.GetLevel() != datapb.SegmentLevel_L2 && // ignore l2 segment
-				!segment.GetIsInvisible() &&
-				(segment.GetIsSorted() || segment.GetIsSortedByNamespace()) &&
-				!t.meta.isSegmentCompactionProtected(segment.GetID()) // not protected by snapshot
+			return isNormalManualCompactionCandidate(t.meta, segment)
 		}),
 	}
 

internal/datacoord/compaction_util.go

@@ -104,6 +104,18 @@ func WrapPluginContext(collectionID int64, properties []*commonpb.KeyValuePair,
 	}
 }
 
+func isNormalManualCompactionCandidate(meta *meta, segment *SegmentInfo) bool {
+	return isSegmentHealthy(segment) &&
+		isFlushed(segment) &&
+		!segment.isCompacting &&
+		!segment.GetIsImporting() &&
+		segment.GetLevel() != datapb.SegmentLevel_L0 &&
+		segment.GetLevel() != datapb.SegmentLevel_L2 &&
+		!segment.GetIsInvisible() &&
+		(segment.GetIsSorted() || segment.GetIsSortedByNamespace()) &&
+		!meta.isSegmentCompactionProtected(segment.GetID())
+}
+
 // isCompactionTaskFinished returns true if the task has reached a terminal state
 // (timeout, completed, cleaned, or unknown) and requires no further processing.
 func isCompactionTaskFinished(t *datapb.CompactionTask) bool {