[dbnode] Remove unused Shard.ScanData method (#3148)

src/dbnode/generated/mocks/generate.go

@@ -20,7 +20,7 @@
 
 // mockgen rules for generating mocks for exported interfaces (reflection mode)
 
-//go:generate sh -c "mockgen -package=fs $PACKAGE/src/dbnode/persist/fs DataFileSetWriter,DataFileSetReader,DataFileSetSeeker,IndexFileSetWriter,IndexFileSetReader,IndexSegmentFileSetWriter,IndexSegmentFileSet,IndexSegmentFile,SnapshotMetadataFileWriter,DataFileSetSeekerManager,ConcurrentDataFileSetSeeker,MergeWith,StreamingWriter,DataEntryProcessor | genclean -pkg $PACKAGE/src/dbnode/persist/fs -out $GOPATH/src/$PACKAGE/src/dbnode/persist/fs/fs_mock.go"
+//go:generate sh -c "mockgen -package=fs $PACKAGE/src/dbnode/persist/fs DataFileSetWriter,DataFileSetReader,DataFileSetSeeker,IndexFileSetWriter,IndexFileSetReader,IndexSegmentFileSetWriter,IndexSegmentFileSet,IndexSegmentFile,SnapshotMetadataFileWriter,DataFileSetSeekerManager,ConcurrentDataFileSetSeeker,MergeWith,StreamingWriter | genclean -pkg $PACKAGE/src/dbnode/persist/fs -out $GOPATH/src/$PACKAGE/src/dbnode/persist/fs/fs_mock.go"
 //go:generate sh -c "mockgen -package=xio $PACKAGE/src/dbnode/x/xio SegmentReader,SegmentReaderPool | genclean -pkg $PACKAGE/src/dbnode/x/xio -out $GOPATH/src/$PACKAGE/src/dbnode/x/xio/io_mock.go"
 //go:generate sh -c "mockgen -package=digest -destination=$GOPATH/src/$PACKAGE/src/dbnode/digest/digest_mock.go $PACKAGE/src/dbnode/digest ReaderWithDigest"
 //go:generate sh -c "mockgen -package=series $PACKAGE/src/dbnode/storage/series DatabaseSeries,QueryableBlockRetriever | genclean -pkg $PACKAGE/src/dbnode/storage/series -out $GOPATH/src/$PACKAGE/src/dbnode/storage/series/series_mock.go"

src/dbnode/persist/fs/types.go

@@ -704,11 +704,3 @@ type StreamedMetadataEntry struct {
 
 // NewReaderFn creates a new DataFileSetReader.
 type NewReaderFn func(bytesPool pool.CheckedBytesPool, opts Options) (DataFileSetReader, error)
-
-// DataEntryProcessor processes StreamedDataEntries.
-type DataEntryProcessor interface {
-	// SetEntriesCount sets the number of entries to be processed.
-	SetEntriesCount(int)
-	// ProcessEntry processes a single StreamedDataEntry.
-	ProcessEntry(StreamedDataEntry) error
-}

src/dbnode/storage/shard.go

@@ -2839,66 +2839,6 @@ func (s *dbShard) OpenStreamingReader(blockStart time.Time) (fs.DataFileSetReade
 	return reader, nil
 }
 
-func (s *dbShard) ScanData(
-	blockStart time.Time,
-	processor fs.DataEntryProcessor,
-) error {
-	latestVolume, err := s.LatestVolume(blockStart)
-	if err != nil {
-		return err
-	}
-
-	reader, err := s.newReaderFn(s.opts.BytesPool(), s.opts.CommitLogOptions().FilesystemOptions())
-	if err != nil {
-		return 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 err
-	}
-
-	readEntriesErr := s.scanDataWithReader(reader, processor)
-	// Always close the reader regardless of if failed, but
-	// make sure to propagate if an error occurred closing the reader too.
-	readCloseErr := reader.Close()
-	if err := readEntriesErr; err != nil {
-		return readEntriesErr
-	}
-	return readCloseErr
-}
-
-func (s *dbShard) scanDataWithReader(
-	reader fs.DataFileSetReader,
-	processor fs.DataEntryProcessor,
-) error {
-	processor.SetEntriesCount(reader.Entries())
-
-	for {
-		entry, err := reader.StreamingRead()
-		if err != nil {
-			if errors.Is(err, io.EOF) {
-				return nil
-			}
-			return err
-		}
-
-		if err := processor.ProcessEntry(entry); err != nil {
-			return err
-		}
-	}
-}
-
 func (s *dbShard) logFlushResult(r dbShardFlushResult) {
 	s.logger.Debug("shard flush outcome",
 		zap.Uint32("shard", s.ID()),

src/dbnode/storage/shard_test.go

@@ -23,7 +23,6 @@ package storage
 import (
 	"errors"
 	"fmt"
-	"io"
 	"io/ioutil"
 	"os"
 	"strconv"
@@ -1991,52 +1990,6 @@ func TestOpenStreamingReader(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func TestShardScan(t *testing.T) {
-	ctrl := xtest.NewController(t)
-	defer ctrl.Finish()
-
-	var (
-		blockSize = time.Hour
-		start     = time.Now().Truncate(blockSize)
-		testOpts  = DefaultTestOptions()
-	)
-
-	shard := testDatabaseShard(t, testOpts)
-	defer assert.NoError(t, shard.Close())
-
-	shardEntries := []fs.StreamedDataEntry{
-		{
-			ID:           ident.BytesID("id1"),
-			EncodedTags:  ts.EncodedTags("tags1"),
-			Data:         []byte{1},
-			DataChecksum: 11,
-		},
-		{
-			ID:           ident.BytesID("id2"),
-			EncodedTags:  ts.EncodedTags("tags2"),
-			Data:         []byte{2},
-			DataChecksum: 22,
-		},
-	}
-
-	processor := fs.NewMockDataEntryProcessor(ctrl)
-	processor.EXPECT().SetEntriesCount(len(shardEntries))
-
-	reader, _ := getMockReader(ctrl, t, shard, start, nil)
-	reader.EXPECT().Entries().Return(len(shardEntries))
-	for _, entry := range shardEntries {
-		reader.EXPECT().StreamingRead().Return(entry, nil)
-		processor.EXPECT().ProcessEntry(entry)
-	}
-	reader.EXPECT().StreamingRead().Return(fs.StreamedDataEntry{}, io.EOF)
-
-	shard.newReaderFn = func(pool.CheckedBytesPool, fs.Options) (fs.DataFileSetReader, error) {
-		return reader, nil
-	}
-
-	require.NoError(t, shard.ScanData(start, processor))
-}
-
 func getMockReader(
 	ctrl *gomock.Controller,
 	t *testing.T,

src/dbnode/storage/types.go

@@ -509,13 +509,6 @@ type Shard interface {
 	// BootstrapState returns the shards' bootstrap state.
 	BootstrapState() BootstrapState
 
-	// ScanData performs a "full table scan" on the given block,
-	// calling processor function on every entry read.
-	ScanData(
-		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)