diff --git a/src/dbnode/persist/fs/merger.go b/src/dbnode/persist/fs/merger.go
index 555bbcc6d8..c014af9753 100644
--- a/src/dbnode/persist/fs/merger.go
+++ b/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,
+					},
 				})
 			}
 
diff --git a/src/dbnode/persist/types.go b/src/dbnode/persist/types.go
index f93f28bb56..052161d0fc 100644
--- a/src/dbnode/persist/types.go
+++ b/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.
diff --git a/src/dbnode/storage/namespace.go b/src/dbnode/storage/namespace.go
index 6b07920c19..739af94825 100644
--- a/src/dbnode/storage/namespace.go
+++ b/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)
+}
diff --git a/src/dbnode/storage/storage_mock.go b/src/dbnode/storage/storage_mock.go
index ab66619c1d..ccc6a21831 100644
--- a/src/dbnode/storage/storage_mock.go
+++ b/src/dbnode/storage/storage_mock.go
@@ -1150,6 +1150,22 @@ func (mr *MockNamespaceMockRecorder) SetReadOnly(value interface{}) *gomock.Call
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetReadOnly", reflect.TypeOf((*MockNamespace)(nil).SetReadOnly), value)
 }
 
+// DocRef mocks base method
+func (m *MockNamespace) DocRef(id ident.ID) (doc.Document, bool, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "DocRef", id)
+	ret0, _ := ret[0].(doc.Document)
+	ret1, _ := ret[1].(bool)
+	ret2, _ := ret[2].(error)
+	return ret0, ret1, ret2
+}
+
+// DocRef indicates an expected call of DocRef
+func (mr *MockNamespaceMockRecorder) DocRef(id interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DocRef", reflect.TypeOf((*MockNamespace)(nil).DocRef), id)
+}
+
 // MockdatabaseNamespace is a mock of databaseNamespace interface
 type MockdatabaseNamespace struct {
 	ctrl     *gomock.Controller
@@ -1326,6 +1342,22 @@ func (mr *MockdatabaseNamespaceMockRecorder) SetReadOnly(value interface{}) *gom
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetReadOnly", reflect.TypeOf((*MockdatabaseNamespace)(nil).SetReadOnly), value)
 }
 
+// DocRef mocks base method
+func (m *MockdatabaseNamespace) DocRef(id ident.ID) (doc.Document, bool, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "DocRef", id)
+	ret0, _ := ret[0].(doc.Document)
+	ret1, _ := ret[1].(bool)
+	ret2, _ := ret[2].(error)
+	return ret0, ret1, ret2
+}
+
+// DocRef indicates an expected call of DocRef
+func (mr *MockdatabaseNamespaceMockRecorder) DocRef(id interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DocRef", reflect.TypeOf((*MockdatabaseNamespace)(nil).DocRef), id)
+}
+
 // Close mocks base method
 func (m *MockdatabaseNamespace) Close() error {
 	m.ctrl.T.Helper()
diff --git a/src/dbnode/storage/types.go b/src/dbnode/storage/types.go
index 579cae8c7d..87aa6b3da1 100644
--- a/src/dbnode/storage/types.go
+++ b/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.