[dbnode][coordinator] Support match[] in label endpoints

scripts/docker-integration-tests/prometheus/test.sh

@@ -391,6 +391,47 @@ function test_series {
     '[[ $(curl -s "0.0.0.0:7201/api/v1/series?match[]=prometheus_remote_storage_samples_total&start=-292273086-05-16T16:47:06Z&end=292277025-08-18T07:12:54.999999999Z" | jq -r ".data | length") -eq 1 ]]'
 }
 
+function test_labels {
+  TAG_NAME_0="name_0" TAG_VALUE_0="value_0_1" \
+    TAG_NAME_1="name_1" TAG_VALUE_1="value_1_1" \
+    TAG_NAME_2="name_2" TAG_VALUE_2="value_2_1" \
+    prometheus_remote_write \
+    label_metric now 42.42 \
+    true "Expected request to succeed" \
+    200 "Expected request to return status code 200"
+
+  TAG_NAME_0="name_0" TAG_VALUE_0="value_0_2" \
+    TAG_NAME_1="name_1" TAG_VALUE_1="value_1_2" \
+    prometheus_remote_write \
+    label_metric_2 now 42.42 \
+    true "Expected request to succeed" \
+    200 "Expected request to return status code 200"
+
+  # Test label search with match
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/labels?start=0&end=9999999999999.99999" | jq -r ".data | length") -gt 3 ]]'
+
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/labels?match[]=label_metric&start=0&end=9999999999999.99999" | jq -r ".data | length") -eq 4 ]]'
+
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/labels?match[]=label_metric_2&start=0&end=9999999999999.99999" | jq -r ".data | length") -eq 3 ]]'
+
+  # Test label values search with match
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/label/name_1/values?start=0&end=9999999999999.99999" | jq -r ".data | length") -eq 2 ]]' # two values without a match
+
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/label/name_1/values?match[]=label_metric&start=0&end=9999999999999.99999" | jq -r ".data | length") -eq 1 ]]'
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/label/name_1/values?match[]=label_metric&start=0&end=9999999999999.99999" | jq -r ".data[0]") = "value_1_1" ]]'
+
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/label/name_1/values?match[]=label_metric_2&start=0&end=9999999999999.99999" | jq -r ".data | length") -eq 1 ]]'
+  ATTEMPTS=5 TIMEOUT=2 MAX_TIMEOUT=4 retry_with_backoff  \
+    '[[ $(curl -s "0.0.0.0:7201/api/v1/label/name_1/values?match[]=label_metric_2&start=0&end=9999999999999.99999" | jq -r ".data[0]") = "value_1_2" ]]'
+}
+
 echo "Running readiness test"
 test_readiness
 
@@ -409,6 +450,7 @@ test_prometheus_query_native_timeout
 test_query_restrict_tags
 test_prometheus_remote_write_map_tags
 test_series
+test_labels
 
 echo "Running function correctness tests"
 test_correctness

src/dbnode/storage/index/block.go

@@ -626,7 +626,7 @@ func (b *block) aggregateWithSpan(
 			}
 			return aggOpts.FieldFilter.Allow(field)
 		},
-		fieldIterFn: func(r segment.Reader) (segment.FieldsIterator, error) {
+		fieldIterFn: func(r segment.Reader) (segment.FieldsPostingsListIterator, error) {
 			// NB(prateek): we default to using the regular (FST) fields iterator
 			// unless we have a predefined list of fields we know we need to restrict
 			// our search to, in which case we iterate that list and check if known values
@@ -638,7 +638,7 @@ func (b *block) aggregateWithSpan(
 			// to this function is expected to have (FieldsFilter) pretty small. If that changes
 			// in the future, we can revisit this.
 			if len(aggOpts.FieldFilter) == 0 {
-				return r.Fields()
+				return r.FieldsPostingsList()
 			}
 			return newFilterFieldsIterator(r, aggOpts.FieldFilter)
 		},

src/dbnode/storage/index/fields_terms_iterator.go

@@ -49,23 +49,23 @@ func (o fieldsAndTermsIteratorOpts) allow(f []byte) bool {
 	return o.allowFn(f)
 }
 
-func (o fieldsAndTermsIteratorOpts) newFieldIter(r segment.Reader) (segment.FieldsIterator, error) {
+func (o fieldsAndTermsIteratorOpts) newFieldIter(r segment.Reader) (segment.FieldsPostingsListIterator, error) {
 	if o.fieldIterFn == nil {
-		return r.Fields()
+		return r.FieldsPostingsList()
 	}
 	return o.fieldIterFn(r)
 }
 
 type allowFn func(field []byte) bool
 
-type newFieldIterFn func(r segment.Reader) (segment.FieldsIterator, error)
+type newFieldIterFn func(r segment.Reader) (segment.FieldsPostingsListIterator, error)
 
 type fieldsAndTermsIter struct {
 	reader segment.Reader
 	opts   fieldsAndTermsIteratorOpts
 
 	err       error
-	fieldIter segment.FieldsIterator
+	fieldIter segment.FieldsPostingsListIterator
 	termIter  segment.TermsIterator
 
 	current struct {
@@ -145,10 +145,33 @@ func (fti *fieldsAndTermsIter) setNextField() bool {
 	}
 
 	for fieldIter.Next() {
-		field := fieldIter.Current()
+		field, pl := fieldIter.Current()
 		if !fti.opts.allow(field) {
 			continue
 		}
+		if fti.restrictByPostings == nil {
+			// No restrictions.
+			fti.current.field = field
+			return true
+		}
+
+		bitmap, ok := roaring.BitmapFromPostingsList(pl)
+		if !ok {
+			fti.err = errUnpackBitmapFromPostingsList
+			return false
+		}
+
+		// Check field is part of at least some of the documents we're
+		// restricted to providing results for based on intersection
+		// count.
+		// Note: IntersectionCount is significantly faster than intersecting and
+		// counting results and also does not allocate.
+		if n := fti.restrictByPostings.IntersectionCount(bitmap); n < 1 {
+			// No match, not next result.
+			continue
+		}
+
+		// Matches, this is next result.
 		fti.current.field = field
 		return true
 	}
@@ -213,7 +236,7 @@ func (fti *fieldsAndTermsIter) nextTermsIterResult() (bool, error) {
 			return false, errUnpackBitmapFromPostingsList
 		}
 
-		// Check term isn part of at least some of the documents we're
+		// Check term is part of at least some of the documents we're
 		// restricted to providing results for based on intersection
 		// count.
 		// Note: IntersectionCount is significantly faster than intersecting and

src/dbnode/storage/index/fields_terms_iterator_prop_test.go

@@ -35,6 +35,7 @@ import (
 	"github.com/leanovate/gopter"
 	"github.com/leanovate/gopter/gen"
 	"github.com/leanovate/gopter/prop"
+
 	"github.com/m3db/m3/src/m3ninx/index/segment"
 	xtest "github.com/m3db/m3/src/x/test"
 )
@@ -149,9 +150,9 @@ func genIterableSegment(ctrl *gomock.Controller) gopter.Gen {
 
 			r := segment.NewMockReader(ctrl)
 
-			fieldIterator := &stubFieldIterator{points: fields}
+			fieldsPostingsListIterator := &stubFieldsPostingsListIterator{points: fields}
 
-			r.EXPECT().Fields().Return(fieldIterator, nil).AnyTimes()
+			r.EXPECT().FieldsPostingsList().Return(fieldsPostingsListIterator, nil).AnyTimes()
 
 			for f, values := range tagValues {
 				sort.Slice(values, func(i, j int) bool {

src/dbnode/storage/index/fields_terms_iterator_test.go

@@ -270,7 +270,7 @@ func TestFieldsTermsIteratorIterateTermsAndRestrictByQuery(t *testing.T) {
 	}, slice)
 }
 
-func newMockSegmentReader(ctrl *gomock.Controller, tagValues map[string][]string) segment.Reader {
+func newMockSegmentReader(ctrl *gomock.Controller, tagValues map[string][]string) *segment.MockReader {
 	fields := make([]iterpoint, 0, len(tagValues))
 	for k := range tagValues {
 		fields = append(fields, iterpoint{
@@ -282,9 +282,9 @@ func newMockSegmentReader(ctrl *gomock.Controller, tagValues map[string][]string
 	})
 
 	r := segment.NewMockReader(ctrl)
-	fieldIterator := &stubFieldIterator{points: fields}
+	fieldsPostingsListIterator := &stubFieldsPostingsListIterator{points: fields}
 
-	r.EXPECT().Fields().Return(fieldIterator, nil).AnyTimes()
+	r.EXPECT().FieldsPostingsList().Return(fieldsPostingsListIterator, nil).AnyTimes()
 
 	for _, f := range fields {
 		termValues := tagValues[f.value]
@@ -302,6 +302,40 @@ func newMockSegmentReader(ctrl *gomock.Controller, tagValues map[string][]string
 	return r
 }
 
+type stubFieldsPostingsListIterator struct {
+	current iterpoint
+	points  []iterpoint
+}
+
+func (s *stubFieldsPostingsListIterator) Next() bool {
+	if len(s.points) == 0 {
+		return false
+	}
+	s.current = s.points[0]
+	s.points = s.points[1:]
+	return true
+}
+
+func (s *stubFieldsPostingsListIterator) Current() ([]byte, postings.List) {
+	return []byte(s.current.value), nil
+}
+
+func (s *stubFieldsPostingsListIterator) Err() error {
+	return s.current.err
+}
+
+func (s *stubFieldsPostingsListIterator) Close() error {
+	if s.current.err != nil {
+		return s.current.err
+	}
+	for s.Next() {
+		if err := s.Err(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 type stubTermIterator struct {
 	current iterpoint
 	points  []iterpoint

src/dbnode/storage/index/filter_fields_iterator.go

@@ -21,9 +21,12 @@
 package index
 
 import (
+	"bytes"
 	"errors"
 
 	"github.com/m3db/m3/src/m3ninx/index/segment"
+	"github.com/m3db/m3/src/m3ninx/postings"
+	xerrors "github.com/m3db/m3/src/x/errors"
 )
 
 var (
@@ -33,54 +36,71 @@ var (
 func newFilterFieldsIterator(
 	reader segment.Reader,
 	fields AggregateFieldFilter,
-) (segment.FieldsIterator, error) {
+) (segment.FieldsPostingsListIterator, error) {
 	if len(fields) == 0 {
 		return nil, errNoFiltersSpecified
 	}
+	fieldsIter, err := reader.FieldsPostingsList()
+	if err != nil {
+		return nil, err
+	}
 	return &filterFieldsIterator{
 		reader:     reader,
+		fieldsIter: fieldsIter,
 		fields:     fields,
 		currentIdx: -1,
 	}, nil
 }
 
 type filterFieldsIterator struct {
-	reader segment.Reader
-	fields AggregateFieldFilter
+	reader     segment.Reader
+	fieldsIter segment.FieldsPostingsListIterator
+	fields     AggregateFieldFilter
 
 	err        error
 	currentIdx int
 }
 
-var _ segment.FieldsIterator = &filterFieldsIterator{}
+var _ segment.FieldsPostingsListIterator = &filterFieldsIterator{}
 
 func (f *filterFieldsIterator) Next() bool {
 	if f.err != nil {
 		return false
 	}
 
-	f.currentIdx++ // required because we start at -1
-	for f.currentIdx < len(f.fields) {
-		field := f.fields[f.currentIdx]
+	for f.fieldsIter.Next() {
+		field, _ := f.fieldsIter.Current()
 
-		ok, err := f.reader.ContainsField(field)
-		if err != nil {
-			f.err = err
-			return false
+		found := false
+		for _, f := range f.fields {
+			if bytes.Equal(field, f) {
+				found = true
+				break
+			}
 		}
-
-		// i.e. we found a field from the filter list contained in the segment.
-		if ok {
+		if found {
 			return true
 		}
-
-		// the current field is unsuitable, so we skip to the next possiblity.
-		f.currentIdx++
 	}
 
 	return false
 }
 
-func (f *filterFieldsIterator) Current() []byte { return f.fields[f.currentIdx] }
-func (f *filterFieldsIterator) Err() error      { return f.err }
-func (f *filterFieldsIterator) Close() error    { return nil }
+func (f *filterFieldsIterator) Current() ([]byte, postings.List) {
+	return f.fieldsIter.Current()
+}
+
+func (f *filterFieldsIterator) Err() error {
+	return f.err
+}
+
+func (f *filterFieldsIterator) Close() error {
+	multiErr := xerrors.NewMultiError()
+	if err := f.reader.Close(); err != nil {
+		multiErr = multiErr.Add(err)
+	}
+	if err := f.fieldsIter.Close(); err != nil {
+		multiErr = multiErr.Add(err)
+	}
+	return multiErr.FinalError()
+}