[dbnode] Remove tag decoders from write tagged requests.

src/dbnode/integration/fetch_tagged_quorum_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/encoding"
 	"github.com/m3db/m3/src/dbnode/namespace"
 	"github.com/m3db/m3/src/dbnode/storage/index"
+	"github.com/m3db/m3/src/dbnode/storage/index/convert"
 	"github.com/m3db/m3/src/dbnode/topology"
 	"github.com/m3db/m3/src/m3ninx/idx"
 	"github.com/m3db/m3/src/x/context"
@@ -317,7 +318,7 @@ func writeTagged(
 	defer ctx.BlockingClose()
 	for _, n := range nodes {
 		require.NoError(t, n.DB().WriteTagged(ctx, testNamespaces[0], ident.StringID("quorumTest"),
-			ident.NewTagsIterator(ident.NewTags(ident.StringTag("foo", "bar"), ident.StringTag("boo", "baz"))),
+			convert.NewTagsMetadataResolver(ident.NewTags(ident.StringTag("foo", "bar"), ident.StringTag("boo", "baz"))),
 			n.NowFn()(), 42, xtime.Second, nil))
 	}
 }

src/dbnode/network/server/tchannelthrift/node/service.go

@@ -249,7 +249,6 @@ func (s *serviceState) DecNumOutstandingReadRPCs() {
 type pools struct {
 	id                      ident.Pool
 	tagEncoder              serialize.TagEncoderPool
-	tagDecoder              serialize.TagDecoderPool
 	checkedBytesWrapper     xpool.CheckedBytesWrapperPool
 	segmentsArray           segmentsArrayPool
 	writeBatchPooledReqPool *writeBatchPooledReqPool
@@ -321,7 +320,7 @@ func NewService(db storage.Database, opts tchannelthrift.Options) Service {
 		writeBatchPoolSize = maxWriteReqs
 	}
 	writeBatchPooledReqPool := newWriteBatchPooledReqPool(writeBatchPoolSize, iopts)
-	writeBatchPooledReqPool.Init(opts.TagDecoderPool())
+	writeBatchPooledReqPool.Init()
 
 	return &service{
 		state: serviceState{
@@ -343,7 +342,6 @@ func NewService(db storage.Database, opts tchannelthrift.Options) Service {
 			id:                      opts.IdentifierPool(),
 			checkedBytesWrapper:     opts.CheckedBytesWrapperPool(),
 			tagEncoder:              opts.TagEncoderPool(),
-			tagDecoder:              opts.TagDecoderPool(),
 			segmentsArray:           segmentPool,
 			writeBatchPooledReqPool: writeBatchPooledReqPool,
 			blockMetadataV2:         opts.BlockMetadataV2Pool(),
@@ -1701,7 +1699,7 @@ func (s *service) WriteTagged(tctx thrift.Context, req *rpc.WriteTaggedRequest)
 	if err = db.WriteTagged(ctx,
 		s.pools.id.GetStringID(ctx, req.NameSpace),
 		s.pools.id.GetStringID(ctx, req.ID),
-		iter, xtime.FromNormalizedTime(dp.Timestamp, d),
+		idxconvert.NewTagsIterMetadataResolver(iter), xtime.FromNormalizedTime(dp.Timestamp, d),
 		dp.Value, unit, dp.Annotation); err != nil {
 		s.metrics.writeTagged.ReportError(s.nowFn().Sub(callStart))
 		return convert.ToRPCError(err)
@@ -1965,18 +1963,11 @@ func (s *service) WriteTaggedBatchRaw(tctx thrift.Context, req *rpc.WriteTaggedB
 			continue
 		}
 
-		dec, err := s.newPooledTagsDecoder(ctx, elem.EncodedTags, pooledReq)
-		if err != nil {
-			nonRetryableErrors++
-			pooledReq.addError(tterrors.NewBadRequestWriteBatchRawError(i, err))
-			continue
-		}
-
 		seriesID := s.newPooledID(ctx, elem.ID, pooledReq)
+
 		batchWriter.AddTagged(
 			i,
 			seriesID,
-			dec,
 			elem.EncodedTags,
 			xtime.FromNormalizedTime(elem.Datapoint.Timestamp, d),
 			elem.Datapoint.Value,
@@ -2086,18 +2077,11 @@ func (s *service) WriteTaggedBatchRawV2(tctx thrift.Context, req *rpc.WriteTagge
 			continue
 		}
 
-		dec, err := s.newPooledTagsDecoder(ctx, elem.EncodedTags, pooledReq)
-		if err != nil {
-			nonRetryableErrors++
-			pooledReq.addError(tterrors.NewBadRequestWriteBatchRawError(i, err))
-			continue
-		}
-
 		seriesID := s.newPooledID(ctx, elem.ID, pooledReq)
+
 		batchWriter.AddTagged(
 			i,
 			seriesID,
-			dec,
 			elem.EncodedTags,
 			xtime.FromNormalizedTime(elem.Datapoint.Timestamp, d),
 			elem.Datapoint.Value,
@@ -2629,31 +2613,6 @@ func readEncodedResultSegment(
 	return converted.Segments, nil
 }
 
-func (s *service) newTagsDecoder(ctx context.Context, encodedTags []byte) (serialize.TagDecoder, error) {
-	checkedBytes := s.pools.checkedBytesWrapper.Get(encodedTags)
-	dec := s.pools.tagDecoder.Get()
-	ctx.RegisterCloser(dec)
-	dec.Reset(checkedBytes)
-	if err := dec.Err(); err != nil {
-		return nil, err
-	}
-	return dec, nil
-}
-
-func (s *service) newPooledTagsDecoder(
-	ctx context.Context,
-	encodedTags []byte,
-	p *writeBatchPooledReq,
-) (serialize.TagDecoder, error) {
-	if decoder, ok := p.nextPooledTagDecoder(encodedTags); ok {
-		if err := decoder.Err(); err != nil {
-			return nil, err
-		}
-		return decoder, nil
-	}
-	return s.newTagsDecoder(ctx, encodedTags)
-}
-
 func (s *service) newCloseableMetadataV2Result(
 	res *rpc.FetchBlocksMetadataRawV2Result_,
 ) closeableMetadataV2Result {
@@ -2709,23 +2668,6 @@ func (r *writeBatchPooledReq) nextPooledID(idBytes []byte) (ident.ID, bool) {
 	return id, true
 }
 
-func (r *writeBatchPooledReq) nextPooledTagDecoder(encodedTags []byte) (serialize.TagDecoder, bool) {
-	if r.pooledIDsUsed >= len(r.pooledIDs) {
-		return nil, false
-	}
-
-	bytes := r.pooledIDs[r.pooledIDsUsed].bytes
-	bytes.IncRef()
-	bytes.Reset(encodedTags)
-
-	decoder := r.pooledIDs[r.pooledIDsUsed].tagDecoder
-	decoder.Reset(bytes)
-
-	r.pooledIDsUsed++
-
-	return decoder, true
-}
-
 func (r *writeBatchPooledReq) Finalize() {
 	// Reset the pooledIDsUsed and decrement the ref counts
 	for i := 0; i < r.pooledIDsUsed; i++ {
@@ -2829,9 +2771,8 @@ func (r *writeBatchPooledReq) numNonRetryableErrors() int {
 }
 
 type writeBatchPooledReqID struct {
-	bytes      checked.Bytes
-	id         ident.ID
-	tagDecoder serialize.TagDecoder
+	bytes checked.Bytes
+	id    ident.ID
 }
 
 type writeBatchPooledReqPool struct {
@@ -2849,14 +2790,10 @@ func newWriteBatchPooledReqPool(
 	return &writeBatchPooledReqPool{pool: pool}
 }
 
-func (p *writeBatchPooledReqPool) Init(
-	tagDecoderPool serialize.TagDecoderPool,
-) {
+func (p *writeBatchPooledReqPool) Init() {
 	p.pool.Init(func() interface{} {
-		// NB(r): Make pooled IDs 2x the default write batch size to account for
-		// write tagged which also has encoded tags, plus an extra one for the
-		// namespace
-		pooledIDs := make([]writeBatchPooledReqID, 1+(2*client.DefaultWriteBatchSize))
+		// NB(r): Make pooled IDs the default write batch size plus an extra one for the namespace
+		pooledIDs := make([]writeBatchPooledReqID, 1+client.DefaultWriteBatchSize)
 		for i := range pooledIDs {
 			pooledIDs[i].bytes = checked.NewBytes(nil, nil)
 			pooledIDs[i].id = ident.BinaryID(pooledIDs[i].bytes)
@@ -2865,8 +2802,6 @@ func (p *writeBatchPooledReqPool) Init(
 			// immediately dec a ref here to avoid calling get on this ID
 			// being a valid call
 			pooledIDs[i].bytes.DecRef()
-			// Also ready a tag decoder
-			pooledIDs[i].tagDecoder = tagDecoderPool.Get()
 		}
 		return &writeBatchPooledReq{
 			pooledIDs: pooledIDs,

src/dbnode/network/server/tchannelthrift/node/service_test.go

@@ -39,6 +39,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/storage"
 	"github.com/m3db/m3/src/dbnode/storage/block"
 	"github.com/m3db/m3/src/dbnode/storage/index"
+	conv "github.com/m3db/m3/src/dbnode/storage/index/convert"
 	"github.com/m3db/m3/src/dbnode/storage/limits"
 	"github.com/m3db/m3/src/dbnode/storage/limits/permits"
 	"github.com/m3db/m3/src/dbnode/storage/series"
@@ -52,7 +53,6 @@ import (
 	"github.com/m3db/m3/src/x/checked"
 	"github.com/m3db/m3/src/x/context"
 	"github.com/m3db/m3/src/x/ident"
-	"github.com/m3db/m3/src/x/serialize"
 	xtest "github.com/m3db/m3/src/x/test"
 	xtime "github.com/m3db/m3/src/x/time"
 
@@ -1449,14 +1449,15 @@ func TestServiceFetchBlocksMetadataEndpointV2Raw(t *testing.T) {
 		if len(expectedBlocks.tags.Values()) == 0 {
 			require.Equal(t, 0, len(block.EncodedTags))
 		} else {
-			encodedTags := checked.NewBytes(block.EncodedTags, nil)
-			decoder := service.pools.tagDecoder.Get()
-			decoder.Reset(encodedTags)
+			id := ident.BinaryID(checked.NewBytes(block.ID, nil))
 
-			expectedTags := ident.NewTagsIterator(expectedBlocks.tags)
-			require.True(t, ident.NewTagIterMatcher(expectedTags).Matches(decoder))
+			actualTags, err := conv.FromSeriesIDAndEncodedTags(id.Bytes(), block.EncodedTags)
+			require.NoError(t, err)
+
+			expectedTags, err := conv.FromSeriesIDAndTags(id, expectedBlocks.tags)
+			require.NoError(t, err)
 
-			decoder.Close()
+			require.True(t, expectedTags.Equal(actualTags))
 		}
 
 		foundMatch := false
@@ -2831,15 +2832,7 @@ func TestServiceWriteTaggedBatchRaw(t *testing.T) {
 	mockDB := storage.NewMockDatabase(ctrl)
 	mockDB.EXPECT().Options().Return(testStorageOpts).AnyTimes()
 
-	mockDecoder := serialize.NewMockTagDecoder(ctrl)
-	mockDecoder.EXPECT().Reset(gomock.Any()).AnyTimes()
-	mockDecoder.EXPECT().Err().Return(nil).AnyTimes()
-	mockDecoder.EXPECT().Close().AnyTimes()
-	mockDecoderPool := serialize.NewMockTagDecoderPool(ctrl)
-	mockDecoderPool.EXPECT().Get().Return(mockDecoder).AnyTimes()
-
-	opts := tchannelthrift.NewOptions().
-		SetTagDecoderPool(mockDecoderPool)
+	opts := tchannelthrift.NewOptions()
 
 	service := NewService(mockDB, opts).(*service)
 
@@ -2897,15 +2890,7 @@ func TestServiceWriteTaggedBatchRawV2(t *testing.T) {
 	mockDB := storage.NewMockDatabase(ctrl)
 	mockDB.EXPECT().Options().Return(testStorageOpts).AnyTimes()
 
-	mockDecoder := serialize.NewMockTagDecoder(ctrl)
-	mockDecoder.EXPECT().Reset(gomock.Any()).AnyTimes()
-	mockDecoder.EXPECT().Err().Return(nil).AnyTimes()
-	mockDecoder.EXPECT().Close().AnyTimes()
-	mockDecoderPool := serialize.NewMockTagDecoderPool(ctrl)
-	mockDecoderPool.EXPECT().Get().Return(mockDecoder).AnyTimes()
-
-	opts := tchannelthrift.NewOptions().
-		SetTagDecoderPool(mockDecoderPool)
+	opts := tchannelthrift.NewOptions()
 
 	service := NewService(mockDB, opts).(*service)
 
@@ -2964,15 +2949,7 @@ func TestServiceWriteTaggedBatchRawV2MultiNS(t *testing.T) {
 	mockDB := storage.NewMockDatabase(ctrl)
 	mockDB.EXPECT().Options().Return(testStorageOpts).AnyTimes()
 
-	mockDecoder := serialize.NewMockTagDecoder(ctrl)
-	mockDecoder.EXPECT().Reset(gomock.Any()).AnyTimes()
-	mockDecoder.EXPECT().Err().Return(nil).AnyTimes()
-	mockDecoder.EXPECT().Close().AnyTimes()
-	mockDecoderPool := serialize.NewMockTagDecoderPool(ctrl)
-	mockDecoderPool.EXPECT().Get().Return(mockDecoder).AnyTimes()
-
-	opts := tchannelthrift.NewOptions().
-		SetTagDecoderPool(mockDecoderPool)
+	opts := tchannelthrift.NewOptions()
 
 	service := NewService(mockDB, opts).(*service)
 
@@ -3079,15 +3056,7 @@ func TestServiceWriteTaggedBatchRawUnknownError(t *testing.T) {
 	mockDB := storage.NewMockDatabase(ctrl)
 	mockDB.EXPECT().Options().Return(testStorageOpts).AnyTimes()
 
-	mockDecoder := serialize.NewMockTagDecoder(ctrl)
-	mockDecoder.EXPECT().Reset(gomock.Any()).AnyTimes()
-	mockDecoder.EXPECT().Err().Return(nil).AnyTimes()
-	mockDecoder.EXPECT().Close().AnyTimes()
-	mockDecoderPool := serialize.NewMockTagDecoderPool(ctrl)
-	mockDecoderPool.EXPECT().Get().Return(mockDecoder).AnyTimes()
-
-	opts := tchannelthrift.NewOptions().
-		SetTagDecoderPool(mockDecoderPool)
+	opts := tchannelthrift.NewOptions()
 
 	service := NewService(mockDB, opts).(*service)
 

src/dbnode/network/server/tchannelthrift/options.go

@@ -71,11 +71,6 @@ func NewOptions() Options {
 		poolOptions)
 	tagEncoderPool.Init()
 
-	tagDecoderPool := serialize.NewTagDecoderPool(
-		serialize.NewTagDecoderOptions(serialize.TagDecoderOptionsConfig{}),
-		poolOptions)
-	tagDecoderPool.Init()
-
 	bytesWrapperPool := xpool.NewCheckedBytesWrapperPool(poolOptions)
 	bytesWrapperPool.Init()
 
@@ -86,7 +81,6 @@ func NewOptions() Options {
 		blockMetadataV2Pool:      NewBlockMetadataV2Pool(nil),
 		blockMetadataV2SlicePool: NewBlockMetadataV2SlicePool(nil, 0),
 		tagEncoderPool:           tagEncoderPool,
-		tagDecoderPool:           tagDecoderPool,
 		checkedBytesWrapperPool:  bytesWrapperPool,
 		queryLimits:              limits.NoOpQueryLimits(),
 		permitsOptions:           permits.NewOptions(),
@@ -163,16 +157,6 @@ func (o *options) TagEncoderPool() serialize.TagEncoderPool {
 	return o.tagEncoderPool
 }
 
-func (o *options) SetTagDecoderPool(value serialize.TagDecoderPool) Options {
-	opts := *o
-	opts.tagDecoderPool = value
-	return &opts
-}
-
-func (o *options) TagDecoderPool() serialize.TagDecoderPool {
-	return o.tagDecoderPool
-}
-
 func (o *options) SetCheckedBytesWrapperPool(value xpool.CheckedBytesWrapperPool) Options {
 	opts := *o
 	opts.checkedBytesWrapperPool = value

src/dbnode/network/server/tchannelthrift/types.go

@@ -75,12 +75,6 @@ type Options interface {
 	// TagEncoderPool returns the tag encoder pool.
 	TagEncoderPool() serialize.TagEncoderPool
 
-	// SetTagDecoderPool sets the tag encoder pool.
-	SetTagDecoderPool(value serialize.TagDecoderPool) Options
-
-	// TagDecoderPool returns the tag encoder pool.
-	TagDecoderPool() serialize.TagDecoderPool
-
 	// SetCheckedBytesWrapperPool sets the checked bytes wrapper pool.
 	SetCheckedBytesWrapperPool(value xpool.CheckedBytesWrapperPool) Options
 

src/dbnode/persist/fs/commitlog/commit_log_test.go

@@ -1077,9 +1077,9 @@ func TestCommitLogBatchWriteDoesNotAddErroredOrSkippedSeries(t *testing.T) {
 		tt := alignedStart.Add(time.Minute * time.Duration(i))
 		tagsIter := opts.FilesystemOptions().TagDecoderPool().Get()
 		tagsIter.Reset(checked.NewBytes(testSeriesWrites[i].EncodedTags, nil))
-		writes.AddTagged(i, testSeriesWrites[i].ID, tagsIter,
+		require.NoError(t, writes.AddTagged(i, testSeriesWrites[i].ID,
 			testSeriesWrites[i].EncodedTags,
-			tt, float64(i)*10.5, xtime.Second, nil)
+			tt, float64(i)*10.5, xtime.Second, nil))
 	}
 
 	writes.SetSkipWrite(0)

src/dbnode/server/server.go

@@ -742,7 +742,6 @@ func Run(runOpts RunOptions) {
 		SetTopologyInitializer(syncCfg.TopologyInitializer).
 		SetIdentifierPool(opts.IdentifierPool()).
 		SetTagEncoderPool(tagEncoderPool).
-		SetTagDecoderPool(tagDecoderPool).
 		SetCheckedBytesWrapperPool(opts.CheckedBytesWrapperPool()).
 		SetMaxOutstandingWriteRequests(cfg.Limits.MaxOutstandingWriteRequests).
 		SetMaxOutstandingReadRequests(cfg.Limits.MaxOutstandingReadRequests).

src/dbnode/storage/database.go

@@ -34,6 +34,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/storage/block"
 	dberrors "github.com/m3db/m3/src/dbnode/storage/errors"
 	"github.com/m3db/m3/src/dbnode/storage/index"
+	"github.com/m3db/m3/src/dbnode/storage/index/convert"
 	"github.com/m3db/m3/src/dbnode/storage/limits"
 	"github.com/m3db/m3/src/dbnode/storage/series"
 	"github.com/m3db/m3/src/dbnode/tracepoint"
@@ -697,7 +698,7 @@ func (d *db) WriteTagged(
 	ctx context.Context,
 	namespace ident.ID,
 	id ident.ID,
-	tags ident.TagIterator,
+	tagResolver convert.TagMetadataResolver,
 	timestamp time.Time,
 	value float64,
 	unit xtime.Unit,
@@ -709,7 +710,7 @@ func (d *db) WriteTagged(
 		return err
 	}
 
-	seriesWrite, err := n.WriteTagged(ctx, id, tags, timestamp, value, unit, annotation)
+	seriesWrite, err := n.WriteTagged(ctx, id, tagResolver, timestamp, value, unit, annotation)
 	if err != nil {
 		return err
 	}
@@ -802,7 +803,7 @@ func (d *db) writeBatch(
 			seriesWrite, err = n.WriteTagged(
 				ctx,
 				write.Write.Series.ID,
-				write.TagIter,
+				convert.NewEncodedTagsMetadataResolver(write.EncodedTags),
 				write.Write.Datapoint.Timestamp,
 				write.Write.Datapoint.Value,
 				write.Write.Unit,