Serialization now uses compressed tags when possible #745

arnikola · 2018-06-04T19:19:43Z

No description provided.

benraskin92 · 2018-06-04T19:22:44Z

src/coordinator/storage/converter.go

 	"fmt"
 	"time"

+	"github.com/m3db/m3coordinator/util/execution"
+


nit: newline

benraskin92 · 2018-06-04T19:25:42Z

src/coordinator/storage/converter.go

+
+const (
+	initRawFetchAllocSize = 32
+	workerPoolSize        = 10


Will we want to make this config driven in the future? If so, maybe add a comment.

benraskin92 · 2018-06-04T19:34:34Z

src/coordinator/storage/converter.go

+	datapoints := make(ts.Datapoints, 0, initRawFetchAllocSize)
+	for iter.Next() {
+		dp, _, _ := iter.Current()
+		datapoints = append(datapoints, ts.Datapoint{Timestamp: dp.Timestamp, Value: dp.Value})


Since you're pre-allocating above, is it better to use an index. e.g.

for i := 0; iter.Next(); i++ { dp, _, _ := iter.Current() datapoints[i] = ts.Datapoint{Timestamp: dp.Timestamp, Value: dp.Value} }

I need it to be able to expand if initRawFetchAllocSize is too small

benraskin92 · 2018-06-04T19:36:29Z

src/coordinator/storage/converter.go

+	result    *ts.Series
+}
+
+func (w *decompressRequest) Process(ctx context.Context) error {


nit: add a comment here

Surprised the linter let me get away with that

codecov · 2018-06-04T21:02:27Z

Codecov Report

Merging #745 into master will decrease coverage by 0.29%.
The diff coverage is 93.33%.

@@            Coverage Diff            @@
##           master     #745     +/-   ##
=========================================
- Coverage   79.43%   79.14%   -0.3%     
=========================================
  Files         280      283      +3     
  Lines       26569    26844    +275     
=========================================
+ Hits        21105    21245    +140     
- Misses       4186     4313    +127     
- Partials     1278     1286      +8

Flag	Coverage Δ
#cmd	`62.26% <ø> (ø)`	⬆️
#coordinator	`68.76% <93.25%> (-2.04%)`	⬇️
#dbnode	`80.78% <100%> (-0.05%)`	⬇️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 06d3ecc...477b41d. Read the comment docs.

nikunjgit · 2018-06-04T21:42:00Z

src/coordinator/storage/converter.go

+	iterLength := seriesIterators.Len()
+
+	seriesList := make([]*ts.Series, 0, iterLength)
+	div, remainder := iterLength/workerPoolSize, iterLength%workerPoolSize


this logic should probably not live here. Maybe consider moving this within execution.ExecuteParallel ?

nikunjgit · 2018-06-04T21:42:42Z

src/coordinator/storage/converter.go

+	iterLength := seriesIterators.Len()
+
+	seriesList := make([]*ts.Series, 0, iterLength)
+	div, remainder := iterLength/workerPoolSize, iterLength%workerPoolSize


maybe push the responsibility of providing the workerpoolsize to the caller, that way we can eventually centrally control all these resources

nikunjgit · 2018-06-04T21:43:11Z

src/coordinator/storage/converter.go

+	seriesList := make([]*ts.Series, 0, iterLength)
+	div, remainder := iterLength/workerPoolSize, iterLength%workerPoolSize
+	var executionPools [][]execution.Request
+	if remainder > 0 {


i'm hoping we can leverage a library for this instead of doing the math ourselves

nikunjgit · 2018-06-04T21:44:45Z

src/coordinator/tsdb/remote/codecs.go

@@ -40,6 +40,17 @@ func toTime(t int64) time.Time {
 	return storage.TimestampToTime(t)
 }

+func encodeTags(tagMap map[string]string) []*rpc.Tag {


why not just have []rpc.Tag ?

Required to conform to what protobuf expects

nikunjgit · 2018-06-04T21:44:59Z

src/coordinator/tsdb/remote/codecs.go

@@ -80,6 +91,15 @@ func DecodeFetchResult(_ context.Context, rpcSeries []*rpc.Series) ([]*ts.Series
 	return tsSeries, nil
 }

+func decodeTags(tags []*rpc.Tag) models.Tags {


why not just have []rpc.Tag ?

Required to conform to what protobuf expects

nikunjgit · 2018-06-04T21:45:51Z

src/coordinator/tsdb/remote/compressed_codecs.go

+	if iterPools == nil {
+		return nil, false
+	}
+	encoder := iterPools.TagEncoder().Get()


nil check ?

That implies a failing in the iterPools, but I'll add a sanity check

nikunjgit · 2018-06-04T21:46:20Z

src/coordinator/tsdb/remote/compressed_codecs.go

+}
+
+func tagsFromTagIterator(tagIter ident.TagIterator) ([]*rpc.Tag, error) {
+	tags := make([]*rpc.Tag, 0, tagIter.Remaining())


avoid pointer slices

Required to conform to what protobuf expects

Would we ever need to use this func elsewhere? Might be worth it to export it?

Unlikely, as this casts directly to the rpc.Tag type which should only be used for cross DC stuff

nikunjgit · 2018-06-04T21:48:07Z

src/coordinator/tsdb/remote/compressed_codecs.go

+
+	var tags []*rpc.Tag
+	var err error
+	if !canCompress {


maybe not rely on an error to switch logic ? Ideally, if we have compressedDatapoints we should always have compressedTags ? Otherwise, its confusing.

This is going the other way; if we can make compressedTags, we do, otherwise we use regular tags

benraskin92 · 2018-06-06T15:52:33Z

src/coordinator/tsdb/remote/compressed_codecs.go

+	compressedTags, canCompress := compressedTagsFromTagIteratorWithEncoder(tagIter, iterPools)
+
+	var tags []*rpc.Tag
+	var err error


var ( tags []*rpc.Tag err error )

benraskin92 · 2018-06-06T15:54:32Z

src/coordinator/tsdb/remote/compressed_codecs.go

+	iterators encoding.SeriesIterators,
+	iterPools encoding.IteratorPools,
+) (*rpc.FetchResult, error) {
+	iters := iterators.Iters()


defer iterators.Close()?

That's going to be the responsibility of the caller, since tags id and namespace bytes are shared between the iterators and *rpc.FetchResult.

benraskin92 · 2018-06-06T16:04:43Z

src/coordinator/tsdb/remote/compressed_codecs.go

+func tagIteratorFromSeries(series *rpc.Series, iteratorPools encoding.IteratorPools) (ident.TagIterator, error) {
+	var (
+		idPool ident.Pool
+	)


benraskin92 · 2018-06-06T16:05:30Z

src/coordinator/tsdb/remote/compressed_codecs.go

+	var (
+		pooledIterators encoding.MutableSeriesIterators
+		seriesIterators []encoding.SeriesIterator
+		numSeries       = len(rpcSeries)


benraskin92 · 2018-06-06T18:55:43Z

src/coordinator/storage/converter_test.go

+	"github.com/m3db/m3db/src/coordinator/util/logging"
+
+	"github.com/golang/mock/gomock"
+	"github.com/m3db/m3x/ident"


nit: importorder

benraskin92 · 2018-06-06T18:56:17Z

src/coordinator/storage/converter_test.go

+
+	results, err := SeriesIteratorsToFetchResult(ctx, iters, ident.StringID("strID"))
+	assert.NoError(t, err)
+


remove newline

benraskin92 · 2018-06-06T18:56:50Z

src/coordinator/storage/fanout/storage_test.go


 	"github.com/golang/mock/gomock"
+	"github.com/m3db/m3x/ident"


importorder

nikunjgit

So eventually, we want both local and remote storages to return blocks. Is that going to be a separate diff ? Once we start only dealing with blocks, how much of this is going to be used.

nikunjgit · 2018-06-07T17:40:14Z

src/coordinator/storage/converter.go

+		return decompressSequentially(iterLength, iters, namespace)
+	}
+
+	var wg sync.WaitGroup


can you extract this out into separate function and maybe call it decompressConcurrently ?

nikunjgit · 2018-06-07T17:51:36Z

src/coordinator/tsdb/remote/compressed_codecs.go

+		err  error
+	)
+
+	if !canCompress {


Its strange to make it optional. Clients might be simpler if compressed datapoints always correspond to compressedTags as well. Clients can then just always go one path instead of switching based on error

nikunjgit · 2018-06-07T17:52:20Z

src/coordinator/tsdb/remote/compressed_codecs.go

+	decoder := iterPools.TagDecoder().Get()
+	decoder.Reset(checkedBytes)
+	defer decoder.Close()
+	return decoder.Duplicate(), nil


can you comment about why duplicate ?

arnikola · 2018-06-07T19:43:08Z

@nikunjgit

So eventually, we want both local and remote storages to return blocks. Is that going to be a separate diff ? Once we start only dealing with blocks, how much of this is going to be used.��

The meat of this will lead to returning encoding.SeriesIterators; these will be passed through the conversion function in #726 to return the blocks we need

nikunjgit · 2018-06-09T04:47:48Z

src/coordinator/tsdb/remote/compressed_codecs.go

@@ -226,6 +289,18 @@ func tagIteratorFromCompressedTags(compressedTags []*rpc.CompressedTag, idPool i
 	return tagIter
 }

+func tagIteratorFromSeries(series *rpc.Series, iteratorPools encoding.IteratorPools) (ident.TagIterator, error) {
+	compressedValues := series.GetCompressed()
+	if len(compressedValues.GetCompressedTags()) > 0 {


nil check for compressedValues ? Might be better to check for isSet ?

Good call on nil check, for some reason didn't think .GetCompressed() could return nil

nikunjgit · 2018-06-09T04:55:21Z

src/coordinator/tsdb/remote/compressed_codecs.go

 	initialize.Do(initializeVars)

+	// Attempt to decompress compressed tags first as this is the only scenario that is expected to fail
+	tagIter, err := tagIteratorFromSeries(timeSeries, iteratorPools)


tagIteratorFromSeries can either decompress tags or create them using series.GetTags(). Should we only allow the first method ? This will enforce that if the series is compressed, then its tags will also always be compressed.

Not necessarily, since series can be compressed but not have compressed tags if no iterator pools are passed in.

nikunjgit

looks good

arnikola added 2 commits June 4, 2018 15:10

Send encoded tag iterators if possible

c33bfd1

Rebase on master, test fixes

bc0c1d7

arnikola requested review from nikunjgit and benraskin92 June 4, 2018 19:19

benraskin92 reviewed Jun 4, 2018

View reviewed changes

src/coordinator/storage/converter.go Outdated

"fmt"

"time"

"github.com/m3db/m3coordinator/util/execution"

Copy link

Collaborator

benraskin92 Jun 4, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: newline

benraskin92 reviewed Jun 4, 2018

View reviewed changes

linter

fb7c82a

Response

1fa5b6e

nikunjgit reviewed Jun 4, 2018

View reviewed changes

benraskin92 reviewed Jun 6, 2018

View reviewed changes

Response

e457080

benraskin92 reviewed Jun 6, 2018

View reviewed changes

arnikola added 2 commits June 6, 2018 16:47

WIP

ead77e9

Use worker pool for expanding series iterators

b8bbb3c

nikunjgit reviewed Jun 7, 2018

View reviewed changes

PR response

076061e

arnikola added 2 commits June 8, 2018 12:36

Add decode pools to config options

918a085

Return workers to pool when finished

116d987

nikunjgit reviewed Jun 9, 2018

View reviewed changes

nikunjgit approved these changes Jun 9, 2018

View reviewed changes

Nil check

477b41d

arnikola merged commit f70b6b1 into master Jun 11, 2018

arnikola deleted the arnikola/compress_tags branch June 11, 2018 20:35


		results, err := SeriesIteratorsToFetchResult(ctx, iters, ident.StringID("strID"))
		assert.NoError(t, err)

Serialization now uses compressed tags when possible #745

Serialization now uses compressed tags when possible #745

Conversation

arnikola commented Jun 4, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Jun 4, 2018 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nikunjgit left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arnikola commented Jun 7, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nikunjgit left a comment

Choose a reason for hiding this comment

codecov bot commented Jun 4, 2018 •

edited

Loading

arnikola commented Jun 7, 2018 •

edited

Loading