[dbnode] Add Shard.OpenStreamingReader method (#3119)

src/dbnode/storage/shard.go

@@ -2810,6 +2810,35 @@ func (s *dbShard) LatestVolume(blockStart time.Time) (int, error) {
 	return s.namespaceReaderMgr.latestVolume(s.shard, blockStart)
 }
 
+func (s *dbShard) OpenStreamingReader(blockStart time.Time) (fs.DataFileSetReader, error) {
+	latestVolume, err := s.LatestVolume(blockStart)
+	if err != nil {
+		return nil, err
+	}
+
+	reader, err := s.newReaderFn(s.opts.BytesPool(), s.opts.CommitLogOptions().FilesystemOptions())
+	if err != nil {
+		return nil, err
+	}
+
+	openOpts := fs.DataReaderOpenOptions{
+		Identifier: fs.FileSetFileIdentifier{
+			Namespace:   s.namespace.ID(),
+			Shard:       s.ID(),
+			BlockStart:  blockStart,
+			VolumeIndex: latestVolume,
+		},
+		FileSetType:      persist.FileSetFlushType,
+		StreamingEnabled: true,
+	}
+
+	if err := reader.Open(openOpts); err != nil {
+		return nil, err
+	}
+
+	return reader, nil
+}
+
 func (s *dbShard) ScanData(
 	blockStart time.Time,
 	processor fs.DataEntryProcessor,

src/dbnode/storage/shard_test.go

@@ -1954,6 +1954,43 @@ func TestShardAggregateTilesVerifySliceLengths(t *testing.T) {
 	require.EqualError(t, err, "blockReaders and sourceBlockVolumes length mismatch (0 != 1)")
 }
 
+func TestOpenStreamingReader(t *testing.T) {
+	ctrl := xtest.NewController(t)
+	defer ctrl.Finish()
+
+	var (
+		blockStart = time.Now().Truncate(time.Hour)
+		testOpts   = DefaultTestOptions()
+	)
+
+	shard := testDatabaseShard(t, testOpts)
+	defer assert.NoError(t, shard.Close())
+
+	latestSourceVolume, err := shard.LatestVolume(blockStart)
+	require.NoError(t, err)
+
+	openOpts := fs.DataReaderOpenOptions{
+		Identifier: fs.FileSetFileIdentifier{
+			Namespace:   shard.namespace.ID(),
+			Shard:       shard.ID(),
+			BlockStart:  blockStart,
+			VolumeIndex: latestSourceVolume,
+		},
+		FileSetType:      persist.FileSetFlushType,
+		StreamingEnabled: true,
+	}
+
+	reader := fs.NewMockDataFileSetReader(ctrl)
+	reader.EXPECT().Open(openOpts).Return(nil)
+
+	shard.newReaderFn = func(pool.CheckedBytesPool, fs.Options) (fs.DataFileSetReader, error) {
+		return reader, nil
+	}
+
+	_, err = shard.OpenStreamingReader(blockStart)
+	require.NoError(t, err)
+}
+
 func TestShardScan(t *testing.T) {
 	ctrl := xtest.NewController(t)
 	defer ctrl.Finish()

src/dbnode/storage/types.go

@@ -283,6 +283,9 @@ type Namespace interface {
 	// Shards returns the shard description.
 	Shards() []Shard
 
+	// ReadableShardAt returns a readable (bootstrapped) shard by id.
+	ReadableShardAt(shardID uint32) (databaseShard, namespace.Context, error)
+
 	// SetIndex sets and enables reverse index for this namespace.
 	SetIndex(reverseIndex NamespaceIndex) error
 
@@ -475,9 +478,6 @@ type databaseNamespace interface {
 		sourceNs databaseNamespace,
 		opts AggregateTilesOptions,
 	) (int64, error)
-
-	// ReadableShardAt returns a shard of this namespace by shardID.
-	ReadableShardAt(shardID uint32) (databaseShard, namespace.Context, error)
 }
 
 // SeriesReadWriteRef is a read/write reference for a series,
@@ -515,6 +515,10 @@ type Shard interface {
 		blockStart time.Time,
 		processor fs.DataEntryProcessor,
 	) error
+
+	// OpenStreamingDataReader creates and opens a streaming fs.DataFileSetReader
+	// on the latest volume of the given block.
+	OpenStreamingReader(blockStart time.Time) (fs.DataFileSetReader, error)
 }
 
 type databaseShard interface {