dbnode Aggregate RPC implementation #1483

prateek · 2019-03-21T03:05:15Z

No description provided.

arnikola

I like the interfaces in general, but still a bit confused about how the filter function will work. Didn't really look into the remainder of the code very much, since it's still WIP; if some of it is good to go gimme a message 👍

arnikola · 2019-03-21T19:09:16Z

src/dbnode/client/aggregate_attempt.go

+)
+
+var (
+	aggregateAttemptArgsZeroed aggregateAttemptArgs


nit: Is there any risk of issues from this being mutable?

nah, this is a common pattern we use all over this package. we only use it to set values to zero without an extra alloc on the stack.

Nice, I like this pattern

Only asking coz got cautioned against doing this in a previous PR I did :p

src/dbnode/generated-source-files.mk

arnikola · 2019-03-21T19:15:07Z

src/dbnode/client/types.go

@@ -78,6 +78,9 @@ type Session interface {
 	// FetchTaggedIDs resolves the provided query to known IDs.
 	FetchTaggedIDs(namespace ident.ID, q index.Query, opts index.QueryOptions) (iter TaggedIDsIterator, exhaustive bool, err error)

+	// Aggregate aggregates values from the database for the given set of contraints.


nit: constraints

At what level in the stack do we start referring to this as a TagCompletion query or something similar v.s. an Aggregate?

arnikola · 2019-03-21T19:20:07Z

src/dbnode/storage/index/types.go

+}
+
+// TagNameFilterFn provides a predicate using which aggregation results can be filtered.
+type TagNameFilterFn func(tagName []byte) bool


Just wondering how would we go about serializing this function to send to the index? Unless you're thinking we'd do the filtering in the client side rather than the index side?

arnikola · 2019-03-21T19:21:25Z

src/dbnode/client/types.go

@@ -90,6 +93,22 @@ type Session interface {
 	Close() error
 }

+// AggregatedTagsIterator iterates over a collection of tag names and associated tag values.
+type AggregatedTagsIterator interface {


Return type looks nice, I like it 👍

arnikola · 2019-03-27T16:06:26Z

src/dbnode/client/types.go

+
+	// Err returns any error encountered.
+	Err() error
+


Would be nice to add a Remaining() int method

codecov · 2019-03-28T21:53:18Z

Codecov Report

Merging #1483 into master will decrease coverage by 0.3%.
The diff coverage is 72%.

@@           Coverage Diff            @@
##           master   #1483     +/-   ##
========================================
- Coverage    71.3%   70.9%   -0.4%     
========================================
  Files         852     852             
  Lines       74159   72799   -1360     
========================================
- Hits        52908   51687   -1221     
+ Misses      17867   17731    -136     
+ Partials     3384    3381      -3

Flag	Coverage Δ
#aggregator	`82.3% <ø> (-0.1%)`	⬇️
#cluster	`85.7% <ø> (ø)`	⬆️
#collector	`63.7% <ø> (ø)`	⬆️
#dbnode	`80.6% <72%> (-0.5%)`	⬇️
#m3em	`73.2% <ø> (ø)`	⬆️
#m3ninx	`74.3% <ø> (-0.1%)`	⬇️
#m3nsch	`51.1% <ø> (ø)`	⬆️
#metrics	`17.5% <ø> (ø)`	⬆️
#msg	`74.9% <ø> (ø)`	⬆️
#query	`66.1% <ø> (ø)`	⬆️
#x	`76.2% <ø> (ø)`	⬆️

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 9456f7e...be8bcd5. Read the comment docs.

arnikola · 2019-03-29T15:52:01Z

src/dbnode/client/aggregate_attempt.go

+
+func (f *aggregateAttempt) reset() {
+	f.args = aggregateAttemptArgsZeroed
+	f.resultIter = nil


Does this need to finalize?

Nah, the caller takes ownership of the result iter after the attempt is done.

nah these *Attempt objects have much simpler lifecycles so the pooling is done manually in session.go. e..g

func (s *session) Aggregate( ns ident.ID, q index.Query, opts index.AggregationOptions, ) (AggregatedTagsIterator, bool, error) { f := s.pools.aggregateAttempt.Get() f.args.ns = ns f.args.query = q f.args.opts = opts err := s.fetchRetrier.Attempt(f.attemptFn) iter, exhaustive := f.resultIter, f.resultExhaustive s.pools.aggregateAttempt.Put(f) return iter, exhaustive, err }

arnikola · 2019-03-29T15:57:45Z

src/dbnode/client/aggregate_attempt.go

+	return err
+}
+
+type aggregateAttemptPool interface {


Does this need to be an interface? Looking at some of the other _attempt files, they mostly just use the struct

This matches the other attempt pools:

type fetchTaggedAttemptPool interface { ... }

So for consistency at least for the attempt pools let's just keep these as interfaces for now.

aye. it's an interface so we can replace it with leakcheck versions of the pool during tests (the fetch-tagged-attempt pool does the same thing). these are both newer than the other codepaths, at somepoint we'd add this type of tests to the other *attempts too

arnikola · 2019-03-29T16:00:04Z

src/dbnode/client/aggregate_attempt.go

+)
+
+var (
+	aggregateAttemptArgsZeroed aggregateAttemptArgs


Nice, I like this pattern

Only asking coz got cautioned against doing this in a previous PR I did :p

arnikola · 2019-03-29T16:13:12Z

src/dbnode/client/fetch_state.go

+		done = true
+		err = fmt.Errorf(
+			"[invariant violated] expected result to be one of %v, received: %v",
+			[]string{"fetchTaggedResultAccumulatorOpts"}, result)


Might be a c/p error: "fetchTaggedResultAccumulatorOpts" -> "fetchTaggedFetchState", "aggregateFetchState" ?

Ah yes, good catch thanks.

arnikola · 2019-03-29T16:15:10Z

src/dbnode/client/fetch_state.go

 	return f.tagResultAccumulator.AsEncodingSeriesIterators(limit, pools)
 }

+func (f *fetchState) asAggregatedTagsIterator(pools fetchTaggedPools) (AggregatedTagsIterator, bool, error) {


nit: Since we added a result type, maybe this and asTaggedIDsIterator should check that?

Sure thing, added.

hm how bout testing f.stateType in all three as...() methods?

arnikola · 2019-03-29T16:15:41Z

src/dbnode/client/fetch_tagged_results_accumulator.go

@@ -26,6 +26,8 @@ import (
 	"sort"
 	"time"

+	"github.com/m3db/m3x/ident"
+


Extra newline, linter may complain here

Yup good call, thanks.

arnikola · 2019-03-29T16:34:52Z

src/dbnode/integration/integration_index_verify.go

+type tagValueSeen bool
+
+type verifyQueryAggregateMetadataResultsOptions struct {
+	exhausitive bool


nit: exhausitive -> exhaustive

Good catch.

arnikola · 2019-03-29T16:35:11Z

src/dbnode/integration/integration_index_verify.go

+func verifyQueryAggregateMetadataResults(
+	t *testing.T,
+	iter client.AggregatedTagsIterator,
+	exhausitive bool,


nit: exhausitive -> exhaustive

Good catch.

arnikola · 2019-03-29T16:36:36Z

src/dbnode/integration/integration_index_verify.go

+	for name, values := range opts.expected {
+		expected[name] = map[tagValue]tagValueSeen{}
+		for value := range values {
+			expected[name][value] = tagValueSeen(false)


Nice, I like this idea a lot (was using dumb stuff with delete(..) when doing similar tests)

arnikola · 2019-03-29T16:40:40Z

src/dbnode/integration/integration_index_verify.go

+		}
+	}
+
+	assert.Equal(t, len(expected), compared,


I think this might be better as assert.Equal(t, 0, len(notMatched), ...) and then we can drop the compared var

Sure thing.

arnikola · 2019-03-29T17:15:46Z

src/dbnode/integration/peers_bootstrap_index_aggregate_test.go

+					"seattle":    struct{}{},
+				},
+			},
+		})


Can we add this test for filtering:

// Match all new_*r*, filtering on `foo` regexpQuery, err = idx.NewRegexpQuery([]byte("city"), []byte("new_.*r.*")) require.NoError(t, err) queryOpts.TermFilter = index.AggregateTermFilter([][]byte{[]byte("foo")}) iter, exhausitive, err = session.Aggregate(ns1.ID(), index.Query{regexpQuery}, queryOpts) require.NoError(t, err) require.True(t, exhausitive) defer iter.Finalize() verifyQueryAggregateMetadataResults(t, iter, exhausitive, verifyQueryAggregateMetadataResultsOptions{ exhausitive: true, expected: map[tagName]aggregateTagValues{ "foo": aggregateTagValues{ "foo": struct{}{}, }, }, })

// Match all new_*r*, names only, filtering on `city` regexpQuery, err = idx.NewRegexpQuery([]byte("city"), []byte("new_.*r.*")) require.NoError(t, err) queryOpts.TermFilter = index.AggregateTermFilter([][]byte{[]byte("city")}) queryOpts.Type = index.AggregateTagNames iter, exhausitive, err = session.Aggregate(ns1.ID(), index.Query{regexpQuery}, queryOpts) require.NoError(t, err) require.True(t, exhausitive) defer iter.Finalize() verifyQueryAggregateMetadataResults(t, iter, exhausitive, verifyQueryAggregateMetadataResultsOptions{ exhausitive: true, expected: map[tagName]aggregateTagValues{ "city": nil, }, })

TY for writing this, good call.

arnikola · 2019-03-29T17:23:28Z

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

+
+	// sort to order results to make test deterministic.
+	sort.Slice(r.Results, func(i, j int) bool {
+		return bytes.Compare(r.Results[i].TagName, r.Results[j].TagName) < 0


Nice, should start using this (mostly just make a throwaway sort.Sort type haha)

arnikola · 2019-03-29T17:27:45Z

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

@@ -359,6 +500,10 @@ func (w *writeTaggedIter) Duplicate() ident.TagIterator {

 // FromRPCQuery will create a m3ninx index query from an RPC query
 func FromRPCQuery(query *rpc.Query) (idx.Query, error) {
+	if query == nil {
+		return idx.NewAllQuery(), nil


nice 👍

Might be worth calling out in the comment though?

// FromRPCQuery will create a m3ninx index query from an RPC query. // // NB: a nil query is considered equivalent to an `All` query

Sure thing.

arnikola · 2019-03-29T17:29:03Z

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

+	// TODO(prateek): pool the []byte underlying opts.TermFilter
+	filters := make([][]byte, 0, len(opts.TermFilter))
+	for _, f := range opts.TermFilter {
+		copied := append([]byte(nil), f...)


More for my own benefit, but how does this compare to:

copied := make([]byte, len(f)) copy(copied, f)

It's optimized to be just as good by the compiler these days thankfully, and it's a one liner =]

arnikola · 2019-03-29T17:33:06Z

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

+		}
+		tagValues := entry.Value()
+		tagValuesMap := tagValues.Map()
+		responseElem.TagValues = make([]*rpc.AggregateQueryResultTagValueElement, 0, tagValuesMap.Len())


meganit: may be nicer to use indexed insertion here rather than appending?

Appending is always safer, heh.

arnikola · 2019-03-29T17:44:16Z

src/dbnode/client/fetch_tagged_results_accumulator_merge_test.go

@@ -359,6 +359,24 @@ func (ts testSerieses) assertMatchesEncodingIters(t *testing.T, iters encoding.S
 	}
 }

+func (ts testSerieses) assertMatchesAggregatedTagsIter(t *testing.T, iters AggregatedTagsIterator) {


arnikola · 2019-03-29T17:58:26Z

src/dbnode/client/session.go

+	// of the hostQueues responding is less than the lifecycle of the current method.
+	nsClone := s.pools.id.Clone(ns)
+
+	// FOLLOWUP(prateek): currently both `index.Query` and the returned request depend on


nit: can we add these callouts to that issue so they don't fall through the cracks?

Sure thing.

arnikola · 2019-03-29T18:00:40Z

src/dbnode/client/session.go

+	case fetchTaggedFetchState:
+		fetchOp := s.pools.fetchTaggedOp.Get()
+		fetchOp.incRef() // indicate current go-routine has a reference to the op
+		closer = func() {


nit: closer = fetchOp.decRef

Sure thing.

arnikola · 2019-03-29T18:00:56Z

src/dbnode/client/session.go

+	case aggregateFetchState:
+		aggOp := s.pools.aggregateOp.Get()
+		aggOp.incRef() // indicate current go-routine has a reference to the op
+		closer = func() {


nit: closer = aggOp.decRef()

Sure thing.

arnikola · 2019-03-29T18:01:50Z

src/dbnode/client/session.go

 			return nil, wrappedErr
 		}
 	}
-
-	op.decRef() // release the ref for the current go-routine
+	closer()


nit: re-add the // release the ref for the current go-routine comment here and above

Yeah agreed.

prateek · 2019-03-29T18:45:38Z

src/dbnode/client/aggregate_tags_iterator.go

+		return false
+	}
+	i.currentIdx++
+	if i.currentIdx >= len(i.backing) {


should we release/reset i.current here too

Sure thing.

prateek · 2019-03-29T18:45:56Z

src/dbnode/client/aggregate_tags_iterator.go

+	return tag
+}
+
+func (i *aggregateTagsIterator) Finalize() {


should we finalize i.current if it exists here

Sure thing.

prateek · 2019-03-29T18:48:13Z

src/dbnode/client/aggregate_tags_iterator.go

+	if i.currentIdx < 0 {
+		return len(i.backing)
+	}
+	return len(i.backing) - i.currentIdx


hm this has an off by one. could replace entire function with return len(i.backing) - (i.currentIdx+1)

Sure thing.

prateek · 2019-03-29T18:50:57Z

src/dbnode/client/fetch_state.go

+		err  error
+	)
+	switch r := result.(type) {
+	case fetchTaggedResultAccumulatorOpts:


should we also test that f.stateType == fetchTaggedFetchState here (just paranoia)

prateek · 2019-03-29T18:51:03Z

src/dbnode/client/fetch_state.go

+	switch r := result.(type) {
+	case fetchTaggedResultAccumulatorOpts:
+		done, err = f.tagResultAccumulator.AddFetchTaggedResponse(r, resultErr)
+	case aggregateResultAccumulatorOpts:


prateek · 2019-03-29T18:54:45Z

src/dbnode/client/fetch_tagged_results_accumulator.go

-	responses  fetchTaggedIDResults
-	exhaustive bool
+	errors       xerrors.Errors
+	responses    fetchTaggedIDResults


super nit: mind renaming var to fetchResponses or something

Sure thing.

prateek · 2019-03-29T18:58:47Z

src/dbnode/client/fetch_tagged_results_accumulator.go

+
+	accum.aggResponses = aggregateResults(results)
+	accum.aggResponses.forEachTag(func(elems aggregateResults, hasMore bool) bool {
+		tagResult := iter.addTag(elems[0].TagName)


just checking my understanding: we're guaranteed elems[0] exists because we'd only call the lambda from the forEachTag implementation if the group was non-empty yea?

Correct yeah.

I'll add a note.

robskillington

LGTM

prateek mentioned this pull request Mar 21, 2019

[query] Plumb potential tag completion/aggregate through to m3query endpoints #1481

Merged

arnikola mentioned this pull request Mar 21, 2019

Poor performance for tag completion endpoints #1453

Closed

6 tasks

arnikola reviewed Mar 21, 2019

View reviewed changes

arnikola mentioned this pull request Mar 27, 2019

[index] Standardizes aggregate interfaces #1502

Merged

arnikola reviewed Mar 27, 2019

View reviewed changes

prateek force-pushed the prateek/dbnode/aggregate-rpc branch 3 times, most recently from 6e42d05 to 0448b1c Compare March 28, 2019 21:52

prateek force-pushed the prateek/dbnode/aggregate-rpc branch from 8993097 to 0b20cdc Compare March 28, 2019 22:22

prateek changed the title ~~[WIP] dbnode/client.Aggregate RPC implementation~~ [WIP] dbnode Aggregate RPC implementation Mar 29, 2019

robskillington changed the title ~~[WIP] dbnode Aggregate RPC implementation~~ dbnode Aggregate RPC implementation Mar 29, 2019

robskillington mentioned this pull request Mar 29, 2019

[WIP] Merge r/dbnode/aggregate-rpc into current aggregate RPC PR #1514

Closed

[dbnode/client] Add Aggregate RPC impl

8a163c4

prateek force-pushed the prateek/dbnode/aggregate-rpc branch from 4f0cca2 to 8a163c4 Compare March 29, 2019 16:45

arnikola reviewed Mar 29, 2019

View reviewed changes

prateek commented Mar 29, 2019

View reviewed changes

Rob Skillington and others added 6 commits March 29, 2019 15:04

Addressed feedback

cb9ed88

fixes

f6df98f

bf

5750102

redo test

a9ae121

set empty

f3030b9

Remove nil checks for values map, always assume map is there

be8bcd5

robskillington approved these changes Mar 29, 2019

View reviewed changes

prateek merged commit 2427bbb into master Mar 29, 2019

justinjc deleted the prateek/dbnode/aggregate-rpc branch July 12, 2019 20:27

dbnode Aggregate RPC implementation #1483

dbnode Aggregate RPC implementation #1483

Conversation

prateek commented Mar 21, 2019

arnikola 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

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 Mar 28, 2019 • 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

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

robskillington Mar 29, 2019 • 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

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

robskillington left a comment

Choose a reason for hiding this comment

codecov bot commented Mar 28, 2019 •

edited

Loading

robskillington Mar 29, 2019 •

edited

Loading