[dbnode] Check ref count when considering if a series is empty

src/dbnode/storage/entry.go

@@ -245,6 +245,12 @@ func (entry *Entry) RelookupAndCheckIsEmpty() (bool, bool) {
 	}
 	defer e.DecrementReaderWriterCount()
 
+	// Consider non-empty if the entry is still being held since this could indicate
+	// another thread holding a new series prior to writing to it.
+	if e.ReaderWriterCount() > 1 {
+		return false, true
+	}
+
 	return e.Series.IsEmpty(), true
 }
 

src/dbnode/storage/flush_test.go

@@ -361,7 +361,6 @@ func TestFlushManagerNamespaceIndexingEnabled(t *testing.T) {
 	ctrl := xtest.NewController(t)
 	defer ctrl.Finish()
 
-	blocks := 24
 	nsOpts := defaultTestNs1Opts.SetIndexOptions(namespace.NewIndexOptions().SetEnabled(true))
 	s1 := NewMockdatabaseShard(ctrl)
 	s2 := NewMockdatabaseShard(ctrl)
@@ -376,7 +375,7 @@ func TestFlushManagerNamespaceIndexingEnabled(t *testing.T) {
 	// Order is important to avoid any edge case where data is GCed from memory without all flushing operations
 	// being completed.
 	gomock.InOrder(
-		ns.EXPECT().WarmFlush(gomock.Any(), gomock.Any()).Return(nil).Times(blocks),
+		ns.EXPECT().WarmFlush(gomock.Any(), gomock.Any()).Return(nil).AnyTimes(),
 		ns.EXPECT().Snapshot(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes(),
 		ns.EXPECT().FlushIndex(gomock.Any()).Return(nil),
 	)

src/dbnode/storage/index.go

@@ -1112,6 +1112,18 @@ func (i *nsIndex) ColdFlush(shards []databaseShard) (OnColdFlushDone, error) {
 	}, nil
 }
 
+// WarmFlushBlockStarts returns all index blockStarts which have been flushed to disk.
+func (i *nsIndex) WarmFlushBlockStarts() []xtime.UnixNano {
+	flushed := make([]xtime.UnixNano, 0)
+	infoFiles := i.readInfoFilesAsMap()
+	for blockStart := range infoFiles {
+		if i.hasIndexWarmFlushedToDisk(infoFiles, blockStart) {
+			flushed = append(flushed, blockStart)
+		}
+	}
+	return flushed
+}
+
 func (i *nsIndex) readInfoFilesAsMap() map[xtime.UnixNano]fs.ReadIndexInfoFileResult {
 	fsOpts := i.opts.CommitLogOptions().FilesystemOptions()
 	infoFiles := i.readIndexInfoFilesFn(fs.ReadIndexInfoFilesOptions{

src/dbnode/storage/index/mutable_segments.go

@@ -618,17 +618,11 @@ func (m *mutableSegments) backgroundCompactWithTask(
 				return true
 			}
 
-			isEmpty, ok := d.OnIndexSeries.RelookupAndCheckIsEmpty()
-			if !ok {
-				// Should not happen since shard will not expire until
-				// no more block starts are indexed.
-				// We do not GC this series if shard is missing since
-				// we open up a race condition where the entry is not
-				// in the shard yet and we GC it since we can't find it
-				// due to an asynchronous insert.
-				instrument.EmitAndLogInvariantViolation(m.iopts, func(l *zap.Logger) {
-					l.Error("unexpected checking series entry does not exist")
-				})
+			isEmpty, isPresent := d.OnIndexSeries.RelookupAndCheckIsEmpty()
+			if !isPresent {
+				// 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 we skip any series
+				// that aren't yet present in the entry lookup.
 				return true
 			}
 

src/dbnode/storage/shard.go

@@ -841,14 +841,10 @@ func (s *dbShard) purgeExpiredSeries(expiredEntries []*Entry) {
 		}
 		// If there have been datapoints written to the series since its
 		// last empty check, we don't remove it.
-		// Also check if still indexed, if so will be GC'd soon from
-		// index and shouldn't evict here since we need to let the index
-		// know this series definitely should be GC'd (it's ambiguous
-		// if the series is missing from the shard, do not know whether
-		// race to insert or whether it actually expired or not).
-		if !series.IsEmpty() || entry.IndexedOrAttemptedAny() {
+		if !series.IsEmpty() {
 			continue
 		}
+
 		// NB(xichen): if we get here, we are guaranteed that there can be
 		// no more reads/writes to this series while the lock is held, so it's
 		// safe to remove it.
@@ -1967,18 +1963,9 @@ func (s *dbShard) UpdateFlushStates() {
 		at := xtime.UnixNano(info.BlockStart)
 		currState := s.flushStateNoBootstrapCheck(at)
 
-		// When initializing from disk, the data files being present are sufficient
-		// for considering the data+index are flushed because that distinction is only
-		// needed to account for the raciness surrounding GCing series based on when
-		// data + index flushes have occurred. For the purposes of just initializing
-		// the state of which blocks have been flushed when bootstrapping, we can
-		// just use the data being present as the indicator.
 		if currState.WarmStatus.DataFlushed != fileOpSuccess {
 			s.markWarmDataFlushStateSuccess(at)
 		}
-		if currState.WarmStatus.IndexFlushed != fileOpSuccess {
-			s.markWarmIndexFlushStateSuccess(at)
-		}
 
 		// Cold version needs to get bootstrapped so that the 1:1 relationship
 		// between volume number and cold version is maintained and the volume
@@ -1992,6 +1979,27 @@ func (s *dbShard) UpdateFlushStates() {
 			s.setFlushStateColdVersionFlushed(at, info.VolumeIndex)
 		}
 	}
+
+	// Populate index flush state only if enabled.
+	if !s.namespace.Options().IndexOptions().Enabled() {
+		return
+	}
+
+	blockSize := s.namespace.Options().RetentionOptions().BlockSize()
+	indexBlockSize := s.namespace.Options().IndexOptions().BlockSize()
+
+	indexFlushedBlockStarts := s.reverseIndex.WarmFlushBlockStarts()
+	for _, blockStart := range indexFlushedBlockStarts {
+		// Index block size is wider than data block size, so we want to set all data blockStarts
+		// within the range of a given index blockStart
+		blockEnd := blockStart.Add(indexBlockSize)
+		for at := blockStart; at < blockEnd; at = at.Add(blockSize) {
+			currState := s.flushStateNoBootstrapCheck(at)
+			if currState.WarmStatus.IndexFlushed != fileOpSuccess {
+				s.markWarmIndexFlushStateSuccess(at)
+			}
+		}
+	}
 }
 
 func (s *dbShard) Bootstrap(

src/dbnode/storage/types.go

@@ -769,6 +769,9 @@ type NamespaceIndex interface {
 		shards []databaseShard,
 	) error
 
+	// WarmFlushBlockStarts returns all index blockStarts which have been flushed to disk.
+	WarmFlushBlockStarts() []xtime.UnixNano
+
 	// ColdFlush performs any cold flushes that the index has outstanding using
 	// the owned shards of the database. Also returns a callback to be called when
 	// cold flushing completes to perform houskeeping.