From 8e318efc9a0d23a541020ab17a61ecf4d0523de5 Mon Sep 17 00:00:00 2001
From: Richard Artoul <richardartoul@gmail.com>
Date: Thu, 17 Jan 2019 17:25:54 -0500
Subject: [PATCH] Dont use pool for allocating tag slices (#1300)

---
 src/dbnode/storage/index/convert/convert.go   | 28 ++++++++++++++-----
 .../storage/index/convert/convert_test.go     | 15 ++++++++++
 src/dbnode/storage/shard.go                   |  7 +++--
 src/dbnode/storage/shard_test.go              |  1 +
 4 files changed, 42 insertions(+), 9 deletions(-)

diff --git a/src/dbnode/storage/index/convert/convert.go b/src/dbnode/storage/index/convert/convert.go
index 3d641ba929..2ed87118e5 100644
--- a/src/dbnode/storage/index/convert/convert.go
+++ b/src/dbnode/storage/index/convert/convert.go
@@ -183,11 +183,15 @@ func TagsFromTagsIter(
 	iter ident.TagIterator,
 	idPool ident.Pool,
 ) (ident.Tags, error) {
-	var (
-		seriesIDBytes = ident.BytesID(seriesID.Bytes())
-		tags          = idPool.Tags()
-	)
+	var tags ident.Tags
+	if idPool != nil {
+		tags = idPool.Tags()
+	} else {
+		tagSlice := make([]ident.Tag, 0, iter.Len())
+		tags = ident.NewTags(tagSlice...)
+	}
 
+	seriesIDBytes := ident.BytesID(seriesID.Bytes())
 	for iter.Next() {
 		curr := iter.Current()
 
@@ -200,17 +204,27 @@ func TagsFromTagsIter(
 			tag.Name = seriesIDBytes[idx : idx+len(nameBytes)]
 			idRef = true
 		} else {
-			tag.Name = idPool.Clone(curr.Name)
+			if idPool != nil {
+				tag.Name = idPool.Clone(curr.Name)
+			} else {
+				copiedBytes := append([]byte(nil), curr.Name.Bytes()...)
+				tag.Name = ident.BytesID(copiedBytes)
+			}
 		}
 		if idx := bytes.Index(seriesIDBytes, valueBytes); idx != -1 {
 			tag.Value = seriesIDBytes[idx : idx+len(valueBytes)]
 			idRef = true
 		} else {
-			tag.Value = idPool.Clone(curr.Value)
+			if idPool != nil {
+				tag.Value = idPool.Clone(curr.Value)
+			} else {
+				copiedBytes := append([]byte(nil), curr.Value.Bytes()...)
+				tag.Value = ident.BytesID(copiedBytes)
+			}
 		}
 
 		if idRef {
-			tag.NoFinalize() // Taken ref, cannot finalize this
+			tag.NoFinalize() // Taken ref, cannot finalize this.
 		}
 
 		tags.Append(tag)
diff --git a/src/dbnode/storage/index/convert/convert_test.go b/src/dbnode/storage/index/convert/convert_test.go
index cbfc49dfd8..6bb40a36d2 100644
--- a/src/dbnode/storage/index/convert/convert_test.go
+++ b/src/dbnode/storage/index/convert/convert_test.go
@@ -143,6 +143,21 @@ func TestTagsFromTagsIter(t *testing.T) {
 	require.True(t, true, expectedTags.Equal(tags))
 }
 
+func TestTagsFromTagsIterNoPool(t *testing.T) {
+	var (
+		id           = ident.StringID("foo")
+		expectedTags = ident.NewTags(
+			ident.StringTag("bar", "baz"),
+			ident.StringTag("foo", "m3"),
+		)
+		tagsIter = ident.NewTagsIterator(expectedTags)
+	)
+
+	tags, err := convert.TagsFromTagsIter(id, tagsIter, nil)
+	require.NoError(t, err)
+	require.True(t, true, expectedTags.Equal(tags))
+}
+
 func TestToMetricInvalidID(t *testing.T) {
 	d := doc.Document{
 		Fields: []doc.Field{
diff --git a/src/dbnode/storage/shard.go b/src/dbnode/storage/shard.go
index bf731be216..4fb843c2ac 100644
--- a/src/dbnode/storage/shard.go
+++ b/src/dbnode/storage/shard.go
@@ -1013,8 +1013,11 @@ func (s *dbShard) newShardEntry(
 		if tagsIter.CurrentIndex() != 0 {
 			return nil, errNewShardEntryTagsIterNotAtIndexZero
 		}
-		seriesTags, err = convert.TagsFromTagsIter(
-			seriesID, tagsIter, s.identifierPool)
+
+		// Pass nil for the identifier pool because the pool will force us to use an array
+		// with a large capacity to store the tags. Since these tags are long-lived, it's
+		// better to allocate an array of the exact size to save memory.
+		seriesTags, err = convert.TagsFromTagsIter(seriesID, tagsIter, nil)
 		tagsIter.Close()
 		if err != nil {
 			return nil, err
diff --git a/src/dbnode/storage/shard_test.go b/src/dbnode/storage/shard_test.go
index 42bc7c8e29..97c44d2ed6 100644
--- a/src/dbnode/storage/shard_test.go
+++ b/src/dbnode/storage/shard_test.go
@@ -1097,6 +1097,7 @@ func TestShardNewInvalidShardEntry(t *testing.T) {
 	gomock.InOrder(
 		iter.EXPECT().Duplicate().Return(iter),
 		iter.EXPECT().CurrentIndex().Return(0),
+		iter.EXPECT().Len().Return(0),
 		iter.EXPECT().Next().Return(false),
 		iter.EXPECT().Err().Return(fmt.Errorf("random err")),
 		iter.EXPECT().Close(),