[dbnode] Repopulate cached searched in index active block (#3671)

src/cmd/services/m3dbnode/config/cache.go

@@ -23,10 +23,11 @@ package config
 import "github.com/m3db/m3/src/dbnode/storage/series"
 
 var (
-	defaultPostingsListCacheSize   = 2 << 11 // 4096
+	defaultPostingsListCacheSize   = 2 << 15 // ~65k
 	defaultPostingsListCacheRegexp = true
 	defaultPostingsListCacheTerms  = true
-	defaultRegexpCacheSize         = 256
+	defaultPostingsListCacheSearch = true
+	defaultRegexpCacheSize         = 1024
 )
 
 // CacheConfigurations is the cache configurations.
@@ -87,6 +88,7 @@ type PostingsListCacheConfiguration struct {
 	Size        *int  `yaml:"size"`
 	CacheRegexp *bool `yaml:"cacheRegexp"`
 	CacheTerms  *bool `yaml:"cacheTerms"`
+	CacheSearch *bool `yaml:"cacheSearch"`
 }
 
 // SizeOrDefault returns the provided size or the default value is none is
@@ -119,6 +121,16 @@ func (p PostingsListCacheConfiguration) CacheTermsOrDefault() bool {
 	return *p.CacheTerms
 }
 
+// CacheSearchOrDefault returns the provided cache search configuration value
+// or the default value is none is provided.
+func (p PostingsListCacheConfiguration) CacheSearchOrDefault() bool {
+	if p.CacheSearch == nil {
+		return defaultPostingsListCacheSearch
+	}
+
+	return *p.CacheSearch
+}
+
 // RegexpCacheConfiguration is a compiled regexp cache for query regexps.
 type RegexpCacheConfiguration struct {
 	Size *int `yaml:"size"`

src/cmd/services/m3dbnode/config/config_test.go

@@ -53,6 +53,7 @@ db:
           size: 100
           cacheRegexp: false
           cacheTerms: false
+          cacheSearch: null
 
   metrics:
       prometheus:
@@ -429,6 +430,7 @@ func TestConfiguration(t *testing.T) {
       size: 100
       cacheRegexp: false
       cacheTerms: false
+      cacheSearch: null
     regexp: null
   filesystem:
     filePathPrefix: /var/lib/m3db

src/dbnode/server/server.go

@@ -497,11 +497,22 @@ func Run(runOpts RunOptions) {
 				SetMetricsScope(scope.SubScope("postings-list-cache")),
 		}
 	)
-	postingsListCache, err := index.NewPostingsListCache(plCacheSize, plCacheOptions)
+	segmentPostingsListCache, err := index.NewPostingsListCache(plCacheSize, plCacheOptions)
 	if err != nil {
-		logger.Fatal("could not construct postings list cache", zap.Error(err))
+		logger.Fatal("could not construct segment postings list cache", zap.Error(err))
 	}
 
+	segmentStopReporting := segmentPostingsListCache.Start()
+	defer segmentStopReporting()
+
+	searchPostingsListCache, err := index.NewPostingsListCache(plCacheSize, plCacheOptions)
+	if err != nil {
+		logger.Fatal("could not construct searches postings list cache", zap.Error(err))
+	}
+
+	searchStopReporting := searchPostingsListCache.Start()
+	defer searchStopReporting()
+
 	// Setup index regexp compilation cache.
 	m3ninxindex.SetRegexpCacheOptions(m3ninxindex.RegexpCacheOptions{
 		Size:  cfg.Cache.RegexpConfiguration().SizeOrDefault(),
@@ -521,7 +532,6 @@ func Run(runOpts RunOptions) {
 	defer queryLimits.Stop()
 	seriesReadPermits.Start()
 	defer seriesReadPermits.Stop()
-	defer postingsListCache.Start()()
 
 	// FOLLOWUP(prateek): remove this once we have the runtime options<->index wiring done
 	indexOpts := opts.IndexOptions()
@@ -531,10 +541,12 @@ func Run(runOpts RunOptions) {
 		insertMode = index.InsertAsync
 	}
 	indexOpts = indexOpts.SetInsertMode(insertMode).
-		SetPostingsListCache(postingsListCache).
+		SetPostingsListCache(segmentPostingsListCache).
+		SetSearchPostingsListCache(searchPostingsListCache).
 		SetReadThroughSegmentOptions(index.ReadThroughSegmentOptions{
-			CacheRegexp: plCacheConfig.CacheRegexpOrDefault(),
-			CacheTerms:  plCacheConfig.CacheTermsOrDefault(),
+			CacheRegexp:   plCacheConfig.CacheRegexpOrDefault(),
+			CacheTerms:    plCacheConfig.CacheTermsOrDefault(),
+			CacheSearches: plCacheConfig.CacheSearchOrDefault(),
 		}).
 		SetMmapReporter(mmapReporter).
 		SetQueryLimits(queryLimits)

src/dbnode/storage/entry.go

@@ -65,6 +65,7 @@ type Entry struct {
 	Series                   series.DatabaseSeries
 	Index                    uint64
 	IndexGarbageCollected    *xatomic.Bool
+	insertTime               *xatomic.Int64
 	indexWriter              IndexWriter
 	curReadWriters           int32
 	reverseIndex             entryIndexState
@@ -101,6 +102,7 @@ func NewEntry(opts NewEntryOptions) *Entry {
 		Series:                   opts.Series,
 		Index:                    opts.Index,
 		IndexGarbageCollected:    xatomic.NewBool(false),
+		insertTime:               xatomic.NewInt64(0),
 		indexWriter:              opts.IndexWriter,
 		nowFn:                    nowFn,
 		pendingIndexBatchSizeOne: make([]writes.PendingIndexInsert, 1),
@@ -234,16 +236,6 @@ 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.
@@ -265,14 +257,32 @@ func (entry *Entry) checkNeedsIndexGarbageCollected(mark bool) bool {
 		return false
 	}
 
-	if mark {
-		// Mark as GCed from index so the entry can be safely cleaned up elsewhere.
-		entry.IndexGarbageCollected.Store(true)
-	}
+	// Mark as GCed from index so the entry can be safely cleaned up elsewhere.
+	entry.IndexGarbageCollected.Store(true)
 
 	return true
 }
 
+// NeedsIndexGarbageCollected checks if the entry is eligible to be garbage collected
+// from the index. Otherwise returns false.
+func (entry *Entry) NeedsIndexGarbageCollected() bool {
+	// This is a cheaper check that loading the entry from the shard again
+	// which makes it cheaper to run frequently.
+	// It may not be as accurate, but it's fine for an approximation since
+	// only a single series in a segment needs to return true to trigger an
+	// index segment to be garbage collected.
+	if entry.insertTime.Load() == 0 {
+		return false // Not inserted, does not need garbage collection.
+	}
+	// Check that a write is not potentially pending and the series is empty.
+	return entry.ReaderWriterCount() == 0 && entry.Series.IsEmpty()
+}
+
+// SetInsertTime marks the entry as having been inserted into the shard at a given timestamp.
+func (entry *Entry) SetInsertTime(t time.Time) {
+	entry.insertTime.Store(t.UnixNano())
+}
+
 // Write writes a new value.
 func (entry *Entry) Write(
 	ctx context.Context,

src/dbnode/storage/index/block.go

@@ -25,9 +25,15 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
+	"runtime"
 	"sync"
 	"time"
 
+	opentracinglog "github.com/opentracing/opentracing-go/log"
+	"github.com/uber-go/tally"
+	"go.uber.org/zap"
+
 	"github.com/m3db/m3/src/dbnode/namespace"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/result"
 	"github.com/m3db/m3/src/dbnode/storage/limits"
@@ -44,11 +50,8 @@ import (
 	"github.com/m3db/m3/src/x/ident"
 	"github.com/m3db/m3/src/x/instrument"
 	xresource "github.com/m3db/m3/src/x/resource"
+	xsync "github.com/m3db/m3/src/x/sync"
 	xtime "github.com/m3db/m3/src/x/time"
-
-	opentracinglog "github.com/opentracing/opentracing-go/log"
-	"github.com/uber-go/tally"
-	"go.uber.org/zap"
 )
 
 var (
@@ -133,6 +136,8 @@ type block struct {
 
 	state blockState
 
+	cachedSearchesWorkers xsync.WorkerPool
+
 	mutableSegments                 *mutableSegments
 	coldMutableSegments             []*mutableSegments
 	shardRangesSegmentsByVolumeType shardRangesSegmentsByVolumeType
@@ -235,11 +240,16 @@ func NewBlock(
 	scope := iopts.MetricsScope().SubScope("index").SubScope("block")
 	iopts = iopts.SetMetricsScope(scope)
 
+	cpus := int(math.Max(1, math.Ceil(0.25*float64(runtime.NumCPU()))))
+	cachedSearchesWorkers := xsync.NewWorkerPool(cpus)
+	cachedSearchesWorkers.Init()
+
 	segs := newMutableSegments(
 		md,
 		blockStart,
 		opts,
 		blockOpts,
+		cachedSearchesWorkers,
 		namespaceRuntimeOptsMgr,
 		iopts,
 	)
@@ -249,6 +259,7 @@ func NewBlock(
 		blockStart,
 		opts,
 		blockOpts,
+		cachedSearchesWorkers,
 		namespaceRuntimeOptsMgr,
 		iopts,
 	)
@@ -261,6 +272,7 @@ func NewBlock(
 		blockEnd:                        blockStart.Add(blockSize),
 		blockSize:                       blockSize,
 		blockOpts:                       blockOpts,
+		cachedSearchesWorkers:           cachedSearchesWorkers,
 		mutableSegments:                 segs,
 		coldMutableSegments:             coldMutableSegments,
 		shardRangesSegmentsByVolumeType: make(shardRangesSegmentsByVolumeType),
@@ -982,7 +994,10 @@ func (b *block) addResults(
 	}
 
 	var (
-		plCache         = b.opts.PostingsListCache()
+		plCaches = ReadThroughSegmentCaches{
+			SegmentPostingsListCache: b.opts.PostingsListCache(),
+			SearchPostingsListCache:  b.opts.SearchPostingsListCache(),
+		}
 		readThroughOpts = b.opts.ReadThroughSegmentOptions()
 		segments        = results.Segments()
 	)
@@ -991,7 +1006,7 @@ func (b *block) addResults(
 		elem := seg.Segment()
 		if immSeg, ok := elem.(segment.ImmutableSegment); ok {
 			// only wrap the immutable segments with a read through cache.
-			elem = NewReadThroughSegment(immSeg, plCache, readThroughOpts)
+			elem = NewReadThroughSegment(immSeg, plCaches, readThroughOpts)
 		}
 		readThroughSegments = append(readThroughSegments, elem)
 	}
@@ -1225,6 +1240,7 @@ func (b *block) RotateColdMutableSegments() error {
 		b.blockStart,
 		b.opts,
 		b.blockOpts,
+		b.cachedSearchesWorkers,
 		b.namespaceRuntimeOptsMgr,
 		b.iopts,
 	)

src/dbnode/storage/index/fields_terms_iterator.go

@@ -29,6 +29,7 @@ import (
 	"github.com/m3db/m3/src/m3ninx/index/segment"
 	"github.com/m3db/m3/src/m3ninx/postings"
 	"github.com/m3db/m3/src/m3ninx/postings/roaring"
+	"github.com/m3db/m3/src/m3ninx/search"
 	"github.com/m3db/m3/src/x/context"
 	xerrors "github.com/m3db/m3/src/x/errors"
 )
@@ -109,13 +110,21 @@ func newFieldsAndTermsIterator(
 	}
 
 	// If need to restrict by query, run the query on the segment first.
-	searcher, err := opts.restrictByQuery.SearchQuery().Searcher()
+	searchQuery := opts.restrictByQuery.SearchQuery()
+	searcher, err := searchQuery.Searcher()
 	if err != nil {
 		return nil, err
 	}
 
-	_, sp := ctx.StartTraceSpan(tracepoint.FieldTermsIteratorIndexSearch)
-	pl, err := searcher.Search(iter.reader)
+	var (
+		_, sp = ctx.StartTraceSpan(tracepoint.FieldTermsIteratorIndexSearch)
+		pl    postings.List
+	)
+	if readThrough, ok := reader.(search.ReadThroughSegmentSearcher); ok {
+		pl, err = readThrough.Search(searchQuery, searcher)
+	} else {
+		pl, err = searcher.Search(reader)
+	}
 	sp.Finish()
 	if err != nil {
 		return nil, err