[dbnode] Add DocRef() to NS and multiple series metadata types (#2931)

src/dbnode/persist/fs/merger.go

@@ -249,10 +249,15 @@ func (m *merger) Merge(
 
 			if err == nil {
 				err = onFlush.OnFlushNewSeries(persist.OnFlushNewSeriesEvent{
-					Shard:          shard,
-					BlockStart:     startTime,
-					FirstWrite:     mergeWithData.FirstWrite,
-					SeriesMetadata: seriesMetadata,
+					Shard:      shard,
+					BlockStart: startTime,
+					FirstWrite: mergeWithData.FirstWrite,
+					SeriesMetadata: persist.SeriesMetadata{
+						Type:     persist.SeriesDocumentType,
+						Document: seriesMetadata,
+						// The lifetime of the shard series metadata is longly lived.
+						LifeTime: persist.SeriesMetadataLifeTimeLong,
+					},
 				})
 			}
 

src/dbnode/persist/types.go

@@ -35,9 +35,7 @@ import (
 	"github.com/pborman/uuid"
 )
 
-var (
-	errReuseableTagIteratorRequired = errors.New("reuseable tags iterator is required")
-)
+var errReuseableTagIteratorRequired = errors.New("reuseable tags iterator is required")
 
 // Metadata is metadata for a time series, it can
 // have several underlying sources.
@@ -322,12 +320,49 @@ const (
 	FileSetIndexContentType
 )
 
+// SeriesMetadataLifeTime describes the memory life time type.
+type SeriesMetadataLifeTime uint8
+
+const (
+	// SeriesMetadataLifeTimeLong means the underlying memory's life time is long lived and exceeds
+	// the execution duration of the series metadata receiver.
+	SeriesMetadataLifeTimeLong SeriesMetadataLifeTime = iota
+	// SeriesMetadataLifeTimeShortLived means that the underlying memory is only valid for the duration
+	// of the OnFlushNewSeries call. Must clone the underlying bytes in order to extend the life time.
+	SeriesMetadataLifeTimeShortLived
+)
+
+// SeriesMetadataType describes the type of series metadata.
+type SeriesMetadataType uint8
+
+const (
+	// SeriesDocumentType means the metadata is in doc.Document form.
+	SeriesDocumentType SeriesMetadataType = iota
+	// SeriesIDAndEncodedTagsType means the metadata is in IDAndEncodedTags form.
+	SeriesIDAndEncodedTagsType
+)
+
+// IDAndEncodedTags contains a series ID and encoded tags.
+type IDAndEncodedTags struct {
+	ID          ident.BytesID
+	EncodedTags ts.EncodedTags
+}
+
+// SeriesMetadata captures different representations of series metadata and
+// the ownership status of the underlying memory.
+type SeriesMetadata struct {
+	Document         doc.Document
+	IDAndEncodedTags IDAndEncodedTags
+	Type             SeriesMetadataType
+	LifeTime         SeriesMetadataLifeTime
+}
+
 // OnFlushNewSeriesEvent is the fields related to a flush of a new series.
 type OnFlushNewSeriesEvent struct {
 	Shard          uint32
 	BlockStart     time.Time
 	FirstWrite     time.Time
-	SeriesMetadata doc.Document
+	SeriesMetadata SeriesMetadata
 }
 
 // OnFlushSeries performs work on a per series level.

src/dbnode/storage/namespace.go

@@ -41,6 +41,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/ts"
 	"github.com/m3db/m3/src/dbnode/ts/writes"
 	"github.com/m3db/m3/src/dbnode/x/xio"
+	"github.com/m3db/m3/src/m3ninx/doc"
 	"github.com/m3db/m3/src/x/clock"
 	"github.com/m3db/m3/src/x/context"
 	xerrors "github.com/m3db/m3/src/x/errors"
@@ -1793,3 +1794,11 @@ func (n *dbNamespace) aggregateTiles(
 
 	return processedTileCount, nil
 }
+
+func (n *dbNamespace) DocRef(id ident.ID) (doc.Document, bool, error) {
+	shard, _, err := n.readableShardFor(id)
+	if err != nil {
+		return doc.Document{}, false, err
+	}
+	return shard.DocRef(id)
+}

src/dbnode/storage/types.go

@@ -285,6 +285,9 @@ type Namespace interface {
 
 	// SetReadOnly sets the value of ReadOnly option.
 	SetReadOnly(value bool)
+
+	// DocRef returns the doc if already present in a namespace shard.
+	DocRef(id ident.ID) (doc.Document, bool, error)
 }
 
 // NamespacesByID is a sortable slice of namespaces by ID.