Check if segments require background GC before compacting

src/dbnode/storage/entry.go

@@ -234,6 +234,16 @@ func (entry *Entry) IfAlreadyIndexedMarkIndexSuccessAndFinalize(
 // TryMarkIndexGarbageCollected checks if the entry is eligible to be garbage collected
 // from the index. If so, it marks the entry as GCed and returns true. Otherwise returns false.
 func (entry *Entry) TryMarkIndexGarbageCollected() bool {
+	return entry.checkNeedsIndexGarbageCollected(true)
+}
+
+// NeedsIndexGarbageCollected checks if the entry is eligible to be garbage collected
+// from the index. If so, it marks the entry as GCed and returns true. Otherwise returns false.
+func (entry *Entry) NeedsIndexGarbageCollected() bool {
+	return entry.checkNeedsIndexGarbageCollected(false)
+}
+
+func (entry *Entry) checkNeedsIndexGarbageCollected(mark bool) bool {
 	// Since series insertions + index insertions are done separately async, it is possible for
 	// a series to be in the index but not have data written yet, and so any series not in the
 	// lookup yet we cannot yet consider empty.
@@ -255,8 +265,10 @@ func (entry *Entry) TryMarkIndexGarbageCollected() bool {
 		return false
 	}
 
-	// Mark as GCed from index so the entry can be safely cleaned up elsewhere.
-	entry.IndexGarbageCollected.Store(true)
+	if mark {
+		// Mark as GCed from index so the entry can be safely cleaned up elsewhere.
+		entry.IndexGarbageCollected.Store(true)
+	}
 
 	return true
 }

src/dbnode/storage/index/mutable_segments.go

@@ -23,6 +23,7 @@ package index
 import (
 	"errors"
 	"fmt"
+	"github.com/m3db/m3/src/m3ninx/x"
 	"math"
 	"runtime"
 	"sync"
@@ -82,6 +83,8 @@ type mutableSegments struct {
 	optsListener             xresource.SimpleCloser
 	writeIndexingConcurrency int
 
+	seriesActiveFn segment.DocumentsFilter
+
 	metrics mutableSegmentsMetrics
 	logger  *zap.Logger
 }
@@ -135,6 +138,7 @@ func newMutableSegments(
 		metrics:    newMutableSegmentsMetrics(iopts.MetricsScope()),
 		logger:     iopts.Logger(),
 	}
+	m.seriesActiveFn = segment.DocumentsFilterFn(m.seriesActive)
 	m.optsListener = namespaceRuntimeOptsMgr.RegisterListener(m)
 	return m
 }
@@ -161,6 +165,19 @@ func (m *mutableSegments) SetNamespaceRuntimeOptions(opts namespace.RuntimeOptio
 	builder.SetSortConcurrency(m.writeIndexingConcurrency)
 }
 
+func (m *mutableSegments) seriesActive(d doc.Metadata) bool {
+	// Filter out any documents that only were indexed for
+	// sealed blocks.
+	if d.OnIndexSeries == nil {
+		instrument.EmitAndLogInvariantViolation(m.iopts, func(l *zap.Logger) {
+			l.Error("unexpected nil for document index entry for background compact")
+		})
+		return true
+	}
+
+	return !d.OnIndexSeries.TryMarkIndexGarbageCollected()
+}
+
 func (m *mutableSegments) WriteBatch(inserts *WriteBatch) (MutableSegmentsStats, error) {
 	m.Lock()
 	if m.state == mutableSegmentsStateClosed {
@@ -397,7 +414,6 @@ func (m *mutableSegments) backgroundCompactWithLock() {
 		gcPlan           = &compaction.Plan{}
 		gcAlreadyRunning = m.compact.compactingBackgroundGarbageCollect
 	)
-
 	if !gcAlreadyRunning {
 		gcRequired = true
 
@@ -416,6 +432,20 @@ func (m *mutableSegments) backgroundCompactWithLock() {
 				continue
 			}
 
+			// Ensure that segment has some series that need to be GC'd.
+			hasAnyInactiveSeries, err := m.segmentAnyInactiveSeries(seg.Segment())
+			if err != nil {
+				instrument.EmitAndLogInvariantViolation(m.iopts, func(l *zap.Logger) {
+					l.Error("error detecting needs background gc segment", zap.Error(err))
+				})
+				continue
+			}
+			if !hasAnyInactiveSeries {
+				// Skip background GC since all series are still active and no
+				// series need to be removed.
+				continue
+			}
+
 			// The active block starts are outdated, need to compact
 			// and remove any old data from the segment.
 			var task compaction.Task
@@ -477,6 +507,45 @@ func (m *mutableSegments) backgroundCompactWithLock() {
 	}
 }
 
+func (m *mutableSegments) segmentAnyInactiveSeries(seg segment.Segment) (bool, error) {
+	reader, err := seg.Reader()
+	if err != nil {
+		return false, err
+	}
+
+	defer reader.Close()
+
+	docs, err := reader.AllDocs()
+	if err != nil {
+		return false, err
+	}
+
+	docsCloser := x.NewSafeCloser(docs)
+	defer func() {
+		// In case of early return cleanup
+		_ = docsCloser.Close()
+	}()
+
+	var result bool
+	for docs.Next() {
+		d := docs.Current()
+		indexEntry := d.OnIndexSeries
+		if indexEntry == nil {
+			return false, fmt.Errorf("document has no index entry: %s", d.ID)
+		}
+		if indexEntry.NeedsIndexGarbageCollected() {
+			result = true
+			break
+		}
+	}
+
+	if err := docs.Err(); err != nil {
+		return false, err
+	}
+
+	return result, docsCloser.Close()
+}
+
 func (m *mutableSegments) shouldEvictCompactedSegmentsWithLock() bool {
 	return m.state == mutableSegmentsStateClosed
 }
@@ -608,18 +677,7 @@ func (m *mutableSegments) backgroundCompactWithTask(
 	var documentsFilter segment.DocumentsFilter
 	if gcRequired {
 		// Only actively filter out documents if GC is required.
-		documentsFilter = segment.DocumentsFilterFn(func(d doc.Metadata) bool {
-			// Filter out any documents that only were indexed for
-			// sealed blocks.
-			if d.OnIndexSeries == nil {
-				instrument.EmitAndLogInvariantViolation(m.iopts, func(l *zap.Logger) {
-					l.Error("unexpected nil for document index entry for background compact")
-				})
-				return true
-			}
-
-			return !d.OnIndexSeries.TryMarkIndexGarbageCollected()
-		})
+		documentsFilter = m.seriesActiveFn
 	}
 
 	start := time.Now()

src/m3ninx/doc/types.go

@@ -118,6 +118,10 @@ type OnIndexSeries interface {
 	// from the index. If so, it marks the entry as GCed and returns true. Otherwise returns false.
 	TryMarkIndexGarbageCollected() bool
 
+	// NeedsIndexGarbageCollected returns if the entry is eligible to be garbage collected
+	// from the index.
+	NeedsIndexGarbageCollected() bool
+
 	// IndexedForBlockStart returns true if the blockStart has been indexed.
 	IndexedForBlockStart(blockStart xtime.UnixNano) bool
 }