Enable to gen compaction plan for stale segments (#20742)

See also: #20502

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

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

internal/datacoord/compaction_trigger.go

@@ -449,11 +449,6 @@ func (t *compactionTrigger) generatePlans(segments []*SegmentInfo, force bool, c
 	// TODO, currently we lack of the measurement of data distribution, there should be another compaction help on redistributing segment based on scalar/vector field distribution
 	for _, segment := range segments {
 		segment := segment.ShadowClone()
-		// by-pass stale segments
-		if !force && t.isStaleSegment(segment) {
-			log.Debug("generate plans skip stale segment", zap.Int64("segmentID", segment.GetID()), zap.Time("lastFlushTime", segment.lastFlushTime))
-			continue
-		}
 		// TODO should we trigger compaction periodically even if the segment has no obvious reason to be compacted?
 		if force || t.ShouldDoSingleCompaction(segment, compactTime) {
 			prioritizedCandidates = append(prioritizedCandidates, segment)

internal/datacoord/compaction_trigger_test.go

@@ -1057,7 +1057,7 @@ func Test_compactionTrigger_smallfiles(t *testing.T) {
 										},
 									},
 								},
-								lastFlushTime: time.Now(),
+								lastFlushTime: time.Now().Add(-100 * time.Minute),
 							},
 							2: {
 								SegmentInfo: &datapb.SegmentInfo{