[dbnode] Use StreamingReadMetadata for bootstrapping

.golangci.yml

@@ -222,7 +222,7 @@ linters:
     - godox 
     # New line required before return would require a large fraction of the
     # code base to need updating, it's not worth the perceived benefit.
-    - nlreturn 
+    - nlreturn
     # Opinionated and sometimes wrong.
     - paralleltest
   disable-all: false

src/dbnode/persist/fs/read.go

@@ -53,9 +53,6 @@ var (
 
 	errUnexpectedSortByOffset = errors.New("should not sort index by offsets when doing reads sorted by id")
 
-	// errReadMetadataOptimizedForRead returned when we optimized for only reading metadata but are attempting a regular read
-	errReadMetadataOptimizedForRead = errors.New("read metadata optimized for regular read")
-
 	errStreamingRequired    = errors.New("streaming must be enabled for streaming read methods")
 	errStreamingUnsupported = errors.New("streaming mode be disabled for non streaming read methods")
 )
@@ -112,10 +109,6 @@ type reader struct {
 	volume                    int
 	open                      bool
 	streamingEnabled          bool
-	// NB(bodu): Informs whether or not we optimize for only reading
-	// metadata. We don't need to sort for reading metadata but sorting is
-	// required if we are performing regulars reads.
-	optimizedReadMetadataOnly bool
 }
 
 // NewReader returns a new reader and expects all files to exist. Will read the
@@ -292,7 +285,6 @@ func (r *reader) Open(opts DataReaderOpenOptions) error {
 	r.open = true
 	r.namespace = namespace
 	r.shard = shard
-	r.optimizedReadMetadataOnly = opts.OptimizedReadMetadataOnly
 
 	return nil
 }
@@ -363,12 +355,10 @@ func (r *reader) readIndexAndSortByOffsetAsc() error {
 		}
 		r.indexEntriesByOffsetAsc = append(r.indexEntriesByOffsetAsc, entry)
 	}
-	// This is false by default so we always sort unless otherwise specified.
-	if !r.optimizedReadMetadataOnly {
-		// NB(r): As we decode each block we need access to each index entry
-		// in the order we decode the data. This is only required for regular reads.
-		sort.Sort(indexEntriesByOffsetAsc(r.indexEntriesByOffsetAsc))
-	}
+	// NB(r): As we decode each block we need access to each index entry
+	// in the order we decode the data. This is only required for regular reads.
+	sort.Sort(indexEntriesByOffsetAsc(r.indexEntriesByOffsetAsc))
+
 	return nil
 }
 
@@ -405,25 +395,19 @@ func (r *reader) StreamingRead() (StreamedDataEntry, error) {
 
 	r.entriesRead++
 
-	dataEntry := StreamedDataEntry{
+	return StreamedDataEntry{
 		ID:           r.streamingID,
 		EncodedTags:  r.streamingTags,
 		Data:         r.streamingData,
 		DataChecksum: uint32(entry.DataChecksum),
-	}
-
-	return dataEntry, nil
+	}, nil
 }
 
 func (r *reader) Read() (ident.ID, ident.TagIterator, checked.Bytes, uint32, error) {
 	if r.streamingEnabled {
 		return nil, nil, nil, 0, errStreamingUnsupported
 	}
 
-	// NB(bodu): We cannot perform regular reads if we're optimizing for only reading metadata.
-	if r.optimizedReadMetadataOnly {
-		return nil, nil, nil, 0, errReadMetadataOptimizedForRead
-	}
 	if r.entries > 0 && len(r.indexEntriesByOffsetAsc) < r.entries {
 		// Have not read the index yet, this is required when reading
 		// data as we need each index entry in order by by the offset ascending
@@ -465,6 +449,33 @@ func (r *reader) Read() (ident.ID, ident.TagIterator, checked.Bytes, uint32, err
 	return id, tags, data, uint32(entry.DataChecksum), nil
 }
 
+func (r *reader) StreamingReadMetadata() (StreamedMetadataEntry, error) {
+	if !r.streamingEnabled {
+		return StreamedMetadataEntry{}, errStreamingRequired
+	}
+
+	if r.metadataRead >= r.entries {
+		return StreamedMetadataEntry{}, io.EOF
+	}
+
+	entry, err := r.decoder.DecodeIndexEntry(nil)
+	if err != nil {
+		return StreamedMetadataEntry{}, err
+	}
+
+	r.streamingID = append(r.streamingID[:0], entry.ID...)
+	r.streamingTags = append(r.streamingTags[:0], entry.EncodedTags...)
+
+	r.metadataRead++
+
+	return StreamedMetadataEntry{
+		ID:           r.streamingID,
+		EncodedTags:  r.streamingTags,
+		Length:       int(entry.Size),
+		DataChecksum: uint32(entry.DataChecksum),
+	}, nil
+}
+
 func (r *reader) ReadMetadata() (ident.ID, ident.TagIterator, int, uint32, error) {
 	if r.streamingEnabled {
 		return nil, nil, 0, 0, errStreamingUnsupported

src/dbnode/persist/fs/read_write_test.go

@@ -31,6 +31,7 @@ import (
 
 	"github.com/m3db/m3/src/dbnode/digest"
 	"github.com/m3db/m3/src/dbnode/persist"
+	"github.com/m3db/m3/src/dbnode/ts"
 	"github.com/m3db/m3/src/x/checked"
 	"github.com/m3db/m3/src/x/ident"
 	xtime "github.com/m3db/m3/src/x/time"
@@ -196,11 +197,6 @@ func readTestDataWithStreamingOpt(
 	streamingEnabled bool,
 ) {
 	for _, underTest := range readTestTypes {
-		if underTest == readTestTypeMetadata && streamingEnabled {
-			// ATM there is no streaming support for metadata.
-			continue
-		}
-
 		rOpenOpts := DataReaderOpenOptions{
 			Identifier: FileSetFileIdentifier{
 				Namespace:  testNs1ID,
@@ -263,7 +259,7 @@ func readTestDataWithStreamingOpt(
 				data.Finalize()
 
 			case readTestTypeMetadata:
-				id, tags, length, checksum, err := r.ReadMetadata()
+				id, tags, length, checksum, err := readMetadata(t, r)
 				require.NoError(t, err)
 
 				// Assert id
@@ -571,26 +567,47 @@ func TestWriterOnlyWritesNonNilBytes(t *testing.T) {
 	})
 }
 
-func readData(t *testing.T, reader DataFileSetReader) (id ident.ID, tags ident.TagIterator, data checked.Bytes, checksum uint32, err error) {
+func readData(
+	t *testing.T,
+	reader DataFileSetReader,
+) (id ident.ID, tags ident.TagIterator, data checked.Bytes, checksum uint32, err error) {
 	if reader.StreamingEnabled() {
 		entry, err := reader.StreamingRead()
 		if err != nil {
 			return nil, nil, nil, 0, err
 		}
-		var tags = ident.EmptyTagIterator
-		if len(entry.EncodedTags) > 0 {
-			tagsDecoder := testTagDecoderPool.Get()
-			tagsDecoder.Reset(checkedBytes(entry.EncodedTags))
-			require.NoError(t, tagsDecoder.Err())
-			tags = tagsDecoder
-		}
-
+		tags := decodeTags(t, entry.EncodedTags)
 		return entry.ID, tags, checked.NewBytes(entry.Data, nil), entry.DataChecksum, err
 	}
 
 	return reader.Read()
 }
 
+func readMetadata(
+	t *testing.T,
+	reader DataFileSetReader,
+) (id ident.ID, tags ident.TagIterator, length int, checksum uint32, err error) {
+	if reader.StreamingEnabled() {
+		entry, err := reader.StreamingReadMetadata()
+		tags := decodeTags(t, entry.EncodedTags)
+		return entry.ID, tags, entry.Length, entry.DataChecksum, err
+	}
+
+	return reader.ReadMetadata()
+}
+
+func decodeTags(t *testing.T, encodedTags ts.EncodedTags) ident.TagIterator {
+	tags := ident.EmptyTagIterator
+	if len(encodedTags) > 0 {
+		tagsDecoder := testTagDecoderPool.Get()
+		tagsDecoder.Reset(checkedBytes(encodedTags))
+		require.NoError(t, tagsDecoder.Err())
+		tags = tagsDecoder
+	}
+
+	return tags
+}
+
 func checkedBytes(b []byte) checked.Bytes {
 	r := checked.NewBytes(b, nil)
 	r.IncRef()

src/dbnode/persist/fs/reader_open_options_matcher.go

@@ -28,8 +28,9 @@ import (
 
 // ReaderOpenOptionsMatcher is a matcher for the DataReaderOpenOptions struct
 type ReaderOpenOptionsMatcher struct {
-	ID          FileSetFileIdentifier
-	FileSetType persist.FileSetType
+	ID               FileSetFileIdentifier
+	FileSetType      persist.FileSetType
+	StreamingEnabled bool
 }
 
 // Matches determine whether m matches a DataWriterOpenOptions
@@ -54,13 +55,17 @@ func (m ReaderOpenOptionsMatcher) Matches(x interface{}) bool {
 	if m.FileSetType != readerOpenOptions.FileSetType {
 		return false
 	}
+	if m.StreamingEnabled != readerOpenOptions.StreamingEnabled {
+		return false
+	}
 
 	return true
 }
 
 func (m ReaderOpenOptionsMatcher) String() string {
 	return fmt.Sprintf(
-		"namespace: %s, shard: %d, blockstart: %d, volumeIndex: %d, filesetType: %s",
-		m.ID.Namespace.String(), m.ID.Shard, m.ID.BlockStart.Unix(), m.ID.VolumeIndex, m.FileSetType,
+		"namespace: %s, shard: %d, blockstart: %d, volumeIndex: %d, filesetType: %s, streamingEnabled: %t", // nolint: lll
+		m.ID.Namespace.String(), m.ID.Shard, m.ID.BlockStart.Unix(), m.ID.VolumeIndex,
+		m.FileSetType, m.StreamingEnabled,
 	)
 }

src/dbnode/persist/fs/types.go

@@ -127,12 +127,9 @@ type DataReaderOpenOptions struct {
 	Identifier FileSetFileIdentifier
 	// FileSetType is the file set type.
 	FileSetType persist.FileSetType
-	// StreamingEnabled enables using streaming methods, such as DataFileSetReader.StreamingRead.
+	// StreamingEnabled enables using streaming methods, such as
+	// DataFileSetReader.StreamingRead and DataFileSetReader.StreamingReadMetadata.
 	StreamingEnabled bool
-	// NB(bodu): This option can inform the reader to optimize for reading
-	// only metadata by not sorting index entries. Setting this option will
-	// throw an error if a regular `Read()` is attempted.
-	OptimizedReadMetadataOnly bool
 }
 
 // DataFileSetReader provides an unsynchronized reader for a TSDB file set.
@@ -145,18 +142,27 @@ type DataFileSetReader interface {
 	// Status returns the status of the reader
 	Status() DataFileSetReaderStatus
 
-	// Read returns the next id, tags, data, checksum tuple or error, will return io.EOF at end of volume.
-	// Use either Read or ReadMetadata to progress through a volume, but not both.
-	// Note: make sure to finalize the ID, close the Tags and finalize the Data when done with
-	// them so they can be returned to their respective pools.
-	Read() (id ident.ID, tags ident.TagIterator, data checked.Bytes, checksum uint32, err error)
-
 	// StreamingRead returns the next unpooled id, encodedTags, data, checksum
 	// values ordered by id, or error, will return io.EOF at end of volume.
 	// Can only by used when DataReaderOpenOptions.StreamingEnabled is true.
+	// Use either StreamingRead or StreamingReadMetadata to progress through a volume, but not both.
 	// Note: the returned data gets invalidated on the next call to StreamingRead.
 	StreamingRead() (StreamedDataEntry, error)
 
+	// StreamingReadMetadata returns the next unpooled id, encodedTags, length checksum
+	// values ordered by id, or error; will return io.EOF at end of volume.
+	// Can only by used when DataReaderOpenOptions.StreamingEnabled is true.
+	// Use either StreamingRead or StreamingReadMetadata to progress through a volume, but not both.
+	// Note: the returned data get invalidated on the next call to StreamingReadMetadata.
+	StreamingReadMetadata() (StreamedMetadataEntry, error)
+
+	// Read returns the next id, tags, data, checksum tuple or error,
+	// will return io.EOF at end of volume.
+	// Use either Read or ReadMetadata to progress through a volume, but not both.
+	// Note: make sure to finalize the ID, close the Tags and finalize the Data when done with
+	// them so they can be returned to their respective pools.
+	Read() (id ident.ID, tags ident.TagIterator, data checked.Bytes, checksum uint32, err error)
+
 	// ReadMetadata returns the next id and metadata or error, will return io.EOF at end of volume.
 	// Use either Read or ReadMetadata to progress through a volume, but not both.
 	// Note: make sure to finalize the ID, and close the Tags when done with them so they can
@@ -678,7 +684,7 @@ type IndexClaimsManager interface {
 	) (int, error)
 }
 
-// StreamedDataEntry contains the data of single entry returned by streaming method(s).
+// StreamedDataEntry contains the data of single entry returned by streaming method.
 // The underlying data slices are reused and invalidated on every read.
 type StreamedDataEntry struct {
 	ID           ident.BytesID
@@ -687,6 +693,15 @@ type StreamedDataEntry struct {
 	DataChecksum uint32
 }
 
+// StreamedMetadataEntry contains the metadata of single entry returned by streaming method.
+// The underlying data slices are reused and invalidated on every read.
+type StreamedMetadataEntry struct {
+	ID           ident.BytesID
+	EncodedTags  ts.EncodedTags
+	Length       int
+	DataChecksum uint32
+}
+
 // NewReaderFn creates a new DataFileSetReader.
 type NewReaderFn func(bytesPool pool.CheckedBytesPool, opts Options) (DataFileSetReader, error)