[dbnode] Cache the results of the commitlog bootstapper between runs

src/dbnode/storage/bootstrap/bootstrapper/commitlog/source.go

@@ -72,6 +72,14 @@ type commitLogSource struct {
 	newReaderFn     newReaderFn
 
 	metrics commitLogSourceMetrics
+
+	// Cache the results so the commit log is not read twice. This source is special since it reads the entire
+	// time series range, irrespective of the requested time range. This is required since the commit log may hold
+	// cold writes that have not been committed to a cold block. Another source might report it fulfilled a cold block,
+	// but it could be missing writes only available in the commit log.
+	//
+	// The bootstrapper is single threaded so we don't need a mutex for this.
+	bootstrapResults bootstrap.NamespaceResults
 }
 
 type bootstrapNamespace struct {
@@ -175,6 +183,12 @@ func (s *commitLogSource) Read(
 	ctx, span, _ := ctx.StartSampledTraceSpan(tracepoint.BootstrapperCommitLogSourceRead)
 	defer span.Finish()
 
+	// bail early if this source already ran before.
+	if s.bootstrapResults.Results != nil {
+		s.log.Info("the entire range of the commit has already been read. returning previous results")
+		return s.bootstrapResults, nil
+	}
+
 	var (
 		// Emit bootstrapping gauge for duration of ReadData.
 		doneReadingData         = s.metrics.emitBootstrapping()
@@ -521,9 +535,8 @@ func (s *commitLogSource) Read(
 		return bootstrap.NamespaceResults{}, err
 	}
 
-	bootstrapResult := bootstrap.NamespaceResults{
-		Results: bootstrap.NewNamespaceResultsMap(bootstrap.NamespaceResultsMapOptions{}),
-	}
+	s.bootstrapResults.Results = bootstrap.NewNamespaceResultsMap(bootstrap.NamespaceResultsMapOptions{})
+
 	for _, ns := range namespaceResults {
 		id := ns.namespace.Metadata.ID()
 		dataResult := result.NewDataBootstrapResult()
@@ -539,15 +552,15 @@ func (s *commitLogSource) Read(
 				indexResult = shardTimeRanges.ToUnfulfilledIndexResult()
 			}
 		}
-		bootstrapResult.Results.Set(id, bootstrap.NamespaceResult{
+		s.bootstrapResults.Results.Set(id, bootstrap.NamespaceResult{
 			Metadata:    ns.namespace.Metadata,
 			Shards:      ns.namespace.Shards,
 			DataResult:  dataResult,
 			IndexResult: indexResult,
 		})
 	}
 
-	return bootstrapResult, nil
+	return s.bootstrapResults, nil
 }
 
 func (s *commitLogSource) snapshotFilesByShard(

src/dbnode/storage/bootstrap/bootstrapper/commitlog/source_data_test.go

@@ -87,6 +87,66 @@ func TestReadEmpty(t *testing.T) {
 	tester.EnsureNoWrites()
 }
 
+func TestCachedResults(t *testing.T) {
+	opts := testDefaultOpts
+	md := testNsMetadata(t)
+
+	nsCtx := namespace.NewContextFrom(md)
+
+	src := newCommitLogSource(opts, fs.Inspection{}).(*commitLogSource)
+
+	blockSize := md.Options().RetentionOptions().BlockSize()
+	now := time.Now()
+	start := now.Truncate(blockSize).Add(-blockSize)
+	end := now.Truncate(blockSize)
+
+	ranges := xtime.NewRanges(xtime.Range{Start: start, End: end})
+
+	foo := ts.Series{Namespace: nsCtx.ID, Shard: 0, ID: ident.StringID("foo")}
+	bar := ts.Series{Namespace: nsCtx.ID, Shard: 1, ID: ident.StringID("bar")}
+	baz := ts.Series{Namespace: nsCtx.ID, Shard: 2, ID: ident.StringID("baz")}
+
+	values := testValues{
+		{foo, start, 1.0, xtime.Second, nil},
+		{foo, start.Add(1 * time.Minute), 2.0, xtime.Second, nil},
+		{bar, start.Add(2 * time.Minute), 1.0, xtime.Second, nil},
+		{bar, start.Add(3 * time.Minute), 2.0, xtime.Second, nil},
+		// "baz" is in shard 2 and should not be returned
+		{baz, start.Add(4 * time.Minute), 1.0, xtime.Second, nil},
+	}
+
+	iterated := false
+	src.newIteratorFn = func(
+		_ commitlog.IteratorOpts,
+	) (commitlog.Iterator, []commitlog.ErrorWithPath, error) {
+		if iterated == true {
+			return nil, nil, errors.New("expected results to be cached")
+		}
+		iterated = true
+		return newTestCommitLogIterator(values, nil), nil, nil
+	}
+
+	targetRanges := result.NewShardTimeRanges().Set(0, ranges).Set(1, ranges)
+	tester := bootstrap.BuildNamespacesTester(t, testDefaultRunOpts, targetRanges, md)
+	defer tester.Finish()
+
+	tester.TestReadWith(src)
+	tester.TestUnfulfilledForNamespaceIsEmpty(md)
+
+	read := tester.EnsureDumpWritesForNamespace(md)
+	require.Equal(t, 2, len(read))
+	enforceValuesAreCorrect(t, values[:4], read)
+	tester.EnsureNoLoadedBlocks()
+
+	tester.TestReadWith(src)
+	tester.TestUnfulfilledForNamespaceIsEmpty(md)
+
+	read = tester.EnsureDumpWritesForNamespace(md)
+	require.Equal(t, 2, len(read))
+	enforceValuesAreCorrect(t, values[:4], read)
+	tester.EnsureNoLoadedBlocks()
+}
+
 func TestReadErrorOnNewIteratorError(t *testing.T) {
 	opts := testDefaultOpts
 	src := newCommitLogSource(opts, fs.Inspection{}).(*commitLogSource)