Add support for reading from snapshot files

src/cmd/tools/dtest/util/seed/generator.go

@@ -71,7 +71,7 @@ func (g *generator) Generate(namespace ident.ID, shard uint32) error {
 	g.logger.Debug("created fake data")
 
 	writer := generate.NewWriter(gOpts)
-	err := writer.Write(namespace, shardSet, data)
+	err := writer.WriteData(namespace, shardSet, data)
 	if err != nil {
 		return fmt.Errorf("unable to write data: %v", err)
 	}

src/dbnode/generated-source-files.mk

@@ -34,7 +34,8 @@ genny-map-all:                         \
 	genny-map-storage                    \
 	genny-map-storage-namespace-metadata \
 	genny-map-storage-repair             \
-	genny-map-storage-index-results
+	genny-map-storage-index-results			 \
+	genny-map-storage-bootstrap-bootstrapper-commitlog
 
 # Map generation rule for client/receivedBlocksMap
 .PHONY: genny-map-client-received-blocks
@@ -127,6 +128,19 @@ genny-map-storage-repair: install-m3x-repo
 		value_type=ReplicaSeriesBlocksMetadata        \
 		target_package=$(m3db_package)/src/dbnode/storage/repair
 
+# Map generation rule for storage/bootstrap/bootstrapper/commitlog
+.PHONY: genny-map-storage-bootstrap-bootstrapper-commitlog
+genny-map-storage-bootstrap-bootstrapper-commitlog: install-m3x-repo
+	cd $(m3x_package_path) && make idhashmap-gen    \
+		pkg=commitlog                                    \
+		value_type=metadataAndEncodersByTime        \
+		target_package=$(m3db_package)/src/dbnode/storage/bootstrap/bootstrapper/commitlog \
+		rename_constructor=newMetadataAndEncodersByTimeMap \
+		rename_constructor_options=newMetadataAndEncodersByTimeMapOptions
+	# Rename both generated map and constructor files
+	mv -f $(m3db_package_path)/src/dbnode/storage/bootstrap/bootstrapper/commitlog/map_gen.go $(m3db_package_path)/src/dbnode/storage/bootstrap/bootstrapper/commitlog/metadata_and_encoders_by_time_map_gen.go
+	mv -f $(m3db_package_path)/src/dbnode/storage/bootstrap/bootstrapper/commitlog/new_map_gen.go $(m3db_package_path)/src/dbnode/storage/bootstrap/bootstrapper/commitlog/metadata_and_encoders_by_time_new_map_gen.go
+
 # Map generation rule for storage/index/ResultsMap
 .PHONY: genny-map-storage-index-results
 genny-map-storage-index-results: install-m3x-repo

src/dbnode/integration/admin_session_fetch_blocks_test.go

@@ -167,6 +167,7 @@ func testSetupToSeriesMaps(
 	nsMetadata namespace.Metadata,
 	metadatasByShard map[uint32][]block.ReplicaMetadata,
 ) map[xtime.UnixNano]generate.SeriesBlock {
+	blockSize := nsMetadata.Options().RetentionOptions().BlockSize()
 	seriesMap := make(map[xtime.UnixNano]generate.SeriesBlock)
 	resultOpts := newDefaulTestResultOptions(testSetup.storageOpts)
 	consistencyLevel := testSetup.storageOpts.RepairOptions().RepairConsistencyLevel()
@@ -201,12 +202,12 @@ func testSetupToSeriesMaps(
 			ctx.Close()
 
 			firstTs := datapoints[0].Timestamp
-			seriesMapList := seriesMap[xtime.ToUnixNano(firstTs)]
+			seriesMapList := seriesMap[xtime.ToUnixNano(firstTs.Truncate(blockSize))]
 			seriesMapList = append(seriesMapList, generate.Series{
 				ID:   id,
 				Data: datapoints,
 			})
-			seriesMap[xtime.ToUnixNano(firstTs)] = seriesMapList
+			seriesMap[xtime.ToUnixNano(firstTs.Truncate(blockSize))] = seriesMapList
 		}
 		require.NoError(t, blocksIter.Err())
 	}

src/dbnode/integration/bootstrap_after_buffer_rotation_regression_test.go

@@ -100,7 +100,7 @@ func TestBootstrapAfterBufferRotation(t *testing.T) {
 			},
 		},
 	}
-	writeCommitLogData(t, setup, commitLogOpts, seriesMaps, ns1.ID())
+	writeCommitLogData(t, setup, commitLogOpts, seriesMaps, ns1, false)
 
 	// Setup bootstrappers - We order them such that the custom test bootstrapper runs first
 	// which does not bootstrap any data, but simply waits until it is signaled, allowing us
@@ -182,7 +182,7 @@ func TestBootstrapAfterBufferRotation(t *testing.T) {
 	// Verify in-memory data match what we expect - both commitlog and memory write
 	// should be present.
 	expectedSeriesMaps := map[xtime.UnixNano]generate.SeriesBlock{
-		xtime.ToUnixNano(commitlogWrite.Timestamp): generate.SeriesBlock{
+		xtime.ToUnixNano(commitlogWrite.Timestamp.Truncate(blockSize)): generate.SeriesBlock{
 			generate.Series{
 				ID:   testID,
 				Data: []ts.Datapoint{commitlogWrite, memoryWrite},

src/dbnode/integration/bootstrap_before_buffer_rotation_no_tick_regression_test.go

@@ -113,7 +113,7 @@ func TestBootstrapBeforeBufferRotationNoTick(t *testing.T) {
 			},
 		},
 	}
-	writeCommitLogData(t, setup, commitLogOpts, seriesMaps, ns1.ID())
+	writeCommitLogData(t, setup, commitLogOpts, seriesMaps, ns1, false)
 
 	// Setup bootstrappers - We order them such that the custom test bootstrapper runs first
 	// which does not bootstrap any data, but simply waits until it is signaled, allowing us
@@ -202,7 +202,7 @@ func TestBootstrapBeforeBufferRotationNoTick(t *testing.T) {
 
 	// Verify in-memory data match what we expect - commitlog write should not be lost
 	expectedSeriesMaps := map[xtime.UnixNano]generate.SeriesBlock{
-		xtime.ToUnixNano(commitlogWrite.Timestamp): generate.SeriesBlock{
+		xtime.ToUnixNano(commitlogWrite.Timestamp.Truncate(blockSize)): generate.SeriesBlock{
 			generate.Series{
 				ID:   testID,
 				Data: []ts.Datapoint{commitlogWrite},

src/dbnode/integration/commitlog_bootstrap_helpers.go

@@ -30,6 +30,7 @@ import (
 	"github.com/m3db/m3db/src/dbnode/integration/generate"
 	"github.com/m3db/m3db/src/dbnode/persist/fs"
 	"github.com/m3db/m3db/src/dbnode/persist/fs/commitlog"
+	"github.com/m3db/m3db/src/dbnode/storage/namespace"
 	"github.com/m3db/m3x/context"
 	"github.com/m3db/m3x/ident"
 	xtime "github.com/m3db/m3x/time"
@@ -89,7 +90,7 @@ func generateSeriesMaps(numBlocks int, starts ...time.Time) generate.SeriesBlock
 		blockConfig = append(blockConfig, generate.BlockConfig{
 			IDs:       name,
 			NumPoints: rand.Intn(100) + 1,
-			Start:     start.Add(time.Duration(i) * time.Minute),
+			Start:     start,
 		})
 	}
 	return generate.BlocksByStart(blockConfig)
@@ -100,30 +101,48 @@ func writeCommitLogData(
 	s *testSetup,
 	opts commitlog.Options,
 	data generate.SeriesBlocksByStart,
-	namespace ident.ID,
+	namespace namespace.Metadata,
+	genSnapshots bool,
 ) {
-	writeCommitLogDataBase(t, s, opts, data, namespace, nil)
+	writeCommitLogDataBase(t, s, opts, data, namespace, nil, nil)
 }
 
 func writeCommitLogDataSpecifiedTS(
 	t *testing.T,
 	s *testSetup,
 	opts commitlog.Options,
 	data generate.SeriesBlocksByStart,
-	namespace ident.ID,
+	namespace namespace.Metadata,
 	ts time.Time,
+	genSnapshots bool,
 ) {
-	writeCommitLogDataBase(t, s, opts, data, namespace, &ts)
+	writeCommitLogDataBase(t, s, opts, data, namespace, &ts, nil)
+}
+
+func writeCommitLogDataWithPredicate(
+	t *testing.T,
+	s *testSetup,
+	opts commitlog.Options,
+	data generate.SeriesBlocksByStart,
+	namespace namespace.Metadata,
+	pred generate.WriteDatapointPredicate,
+) {
+	writeCommitLogDataBase(t, s, opts, data, namespace, nil, pred)
 }
 
 func writeCommitLogDataBase(
 	t *testing.T,
 	s *testSetup,
 	opts commitlog.Options,
 	data generate.SeriesBlocksByStart,
-	namespace ident.ID,
+	namespace namespace.Metadata,
 	specifiedTS *time.Time,
+	pred generate.WriteDatapointPredicate,
 ) {
+	if pred == nil {
+		pred = generate.WriteAllPredicate
+	}
+
 	// ensure commit log is flushing frequently
 	require.Equal(t, defaultIntegrationTestFlushInterval, opts.FlushInterval())
 
@@ -132,6 +151,7 @@ func writeCommitLogDataBase(
 		shardSet     = s.shardSet
 	)
 
+	// Write out commit log data
 	for ts, blk := range data {
 		if specifiedTS != nil {
 			s.setNowFn(*specifiedTS)
@@ -160,20 +180,38 @@ func writeCommitLogDataBase(
 			series, ok := seriesLookup[point.ID.String()]
 			require.True(t, ok)
 			cId := commitlog.Series{
-				Namespace:   namespace,
+				Namespace:   namespace.ID(),
 				Shard:       shardSet.Lookup(point.ID),
 				ID:          point.ID,
 				Tags:        series.tags,
 				UniqueIndex: series.uniqueIndex,
 			}
-			require.NoError(t, commitLog.Write(ctx, cId, point.Datapoint, xtime.Second, nil))
+			if pred(point.Datapoint) {
+				require.NoError(t, commitLog.Write(ctx, cId, point.Datapoint, xtime.Second, nil))
+			}
 		}
 
 		// ensure writes finished
 		require.NoError(t, commitLog.Close())
 	}
 }
 
+func writeSnapshotsWithPredicate(
+	t *testing.T,
+	s *testSetup,
+	opts commitlog.Options,
+	data generate.SeriesBlocksByStart,
+	namespace namespace.Metadata,
+	specifiedTS *time.Time,
+	pred generate.WriteDatapointPredicate,
+	snapshotInterval time.Duration,
+) {
+	// Write out snapshots
+	err := writeTestSnapshotsToDiskWithPredicate(
+		namespace, s, data, pred, snapshotInterval)
+	require.NoError(t, err)
+}
+
 func mustInspectFilesystem(fsOpts fs.Options) fs.Inspection {
 	inspection, err := fs.InspectFilesystem(fsOpts)
 	if err != nil {

src/dbnode/integration/commitlog_bootstrap_index_test.go

@@ -132,8 +132,8 @@ func TestCommitLogIndexBootstrap(t *testing.T) {
 	})
 
 	log.Info("writing data")
-	writeCommitLogData(t, setup, commitLogOpts, seriesMaps, testNamespaces[0])
-	writeCommitLogData(t, setup, commitLogOpts, seriesMaps, testNamespaces[1])
+	writeCommitLogData(t, setup, commitLogOpts, seriesMaps, ns1, false)
+	writeCommitLogData(t, setup, commitLogOpts, seriesMaps, ns2, false)
 	log.Info("finished writing data")
 
 	// Setup bootstrapper after writing data so filesystem inspection can find it.

src/dbnode/integration/commitlog_bootstrap_merge_test.go

@@ -118,7 +118,7 @@ func TestCommitLogAndFSMergeBootstrap(t *testing.T) {
 		t1Nano: seriesMaps[t1Nano],
 		t2Nano: seriesMaps[t2Nano],
 	}
-	writeCommitLogData(t, setup, commitLogOpts, commitlogSeriesMaps, ns1.ID())
+	writeCommitLogData(t, setup, commitLogOpts, commitlogSeriesMaps, ns1, false)
 
 	// commit log bootstrapper (must be after writing out commitlog files so inspection finds files)
 	noOpAll := bootstrapper.NewNoOpAllBootstrapperProvider()

src/dbnode/integration/commitlog_bootstrap_multi_ns_test.go

@@ -83,7 +83,7 @@ func TestCommitLogBootstrapMultipleNamespaces(t *testing.T) {
 
 	setup.namespaceMetadataOrFail(testNamespaces[0])
 	log.Info("writing data - ns1")
-	writeCommitLogData(t, setup, commitLogOpts, ns1SeriesMap, testNamespaces[0])
+	writeCommitLogData(t, setup, commitLogOpts, ns1SeriesMap, ns1, false)
 	log.Info("written data - ns1")
 
 	// Write test data for ns2
@@ -96,7 +96,7 @@ func TestCommitLogBootstrapMultipleNamespaces(t *testing.T) {
 	})
 	setup.namespaceMetadataOrFail(testNamespaces[1])
 	log.Info("writing data - ns2")
-	writeCommitLogData(t, setup, commitLogOpts, ns2SeriesMap, testNamespaces[1])
+	writeCommitLogData(t, setup, commitLogOpts, ns2SeriesMap, ns2, false)
 	log.Info("written data - ns2")
 
 	// Setup bootstrapper after writing data so filesystem inspection can find it

src/dbnode/integration/commitlog_bootstrap_only_reads_required_files_test.go

@@ -68,11 +68,11 @@ func TestCommitLogBootstrapOnlyReadsRequiredFiles(t *testing.T) {
 	now := setup.getNowFn()
 	seriesMaps := generateSeriesMaps(30, now.Add(-2*blockSize), now.Add(-blockSize))
 	log.Info("writing data")
-	writeCommitLogData(t, setup, commitLogOpts, seriesMaps, testNamespaces[0])
+	writeCommitLogData(t, setup, commitLogOpts, seriesMaps, ns1, false)
 	log.Info("finished writing data")
 
 	// The datapoints in this generated data are within the retention period and
-	// would oridinarly be bootstrapped, however, we intentionally write them to a
+	// would ordinarily be bootstrapped, however, we intentionally write them to a
 	// commitlog file that has a timestamp outside of the retention period. This
 	// allows us to verify the commitlog bootstrapping logic will not waste time
 	// reading commitlog files that are outside of the retention period.
@@ -84,8 +84,9 @@ func TestCommitLogBootstrapOnlyReadsRequiredFiles(t *testing.T) {
 		setup,
 		commitLogOpts,
 		seriesMapsExpiredCommitlog,
-		testNamespaces[0],
+		ns1,
 		now.Add(-2*ropts.RetentionPeriod()),
+		false,
 	)
 	log.Info("finished writing data to commitlog file with out of range timestamp")
 

src/dbnode/integration/commitlog_bootstrap_test.go

@@ -71,7 +71,7 @@ func TestCommitLogBootstrap(t *testing.T) {
 	now := setup.getNowFn()
 	seriesMaps := generateSeriesMaps(30, now.Add(-2*blockSize), now.Add(-blockSize))
 	log.Info("writing data")
-	writeCommitLogData(t, setup, commitLogOpts, seriesMaps, testNamespaces[0])
+	writeCommitLogData(t, setup, commitLogOpts, seriesMaps, ns1, false)
 	log.Info("finished writing data")
 
 	// Setup bootstrapper after writing data so filesystem inspection can find it.
@@ -106,7 +106,7 @@ func TestCommitLogBootstrap(t *testing.T) {
 	verifySeriesMapsEqual(t, seriesMaps, observedSeriesMaps)
 
 	// Verify in-memory data match what we expect - no writes should be present
-	// because we didn't issue any writes for this namespaces
+	// because we didn't issue any writes for this namespace.
 	emptySeriesMaps := make(generate.SeriesBlocksByStart)
 	metadatasByShard2 := testSetupMetadatas(t, setup, testNamespaces[1], now.Add(-2*blockSize), now)
 	observedSeriesMaps2 := testSetupToSeriesMaps(t, setup, ns2, metadatasByShard2)